testtools/source/bridgetest/pyuno/impl.py

   1 #
   2 # This file is part of the LibreOffice project.
   3 #
   4 # This Source Code Form is subject to the terms of the Mozilla Public
   5 # License, v. 2.0. If a copy of the MPL was not distributed with this
   6 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   7 #
   8 # This file incorporates work covered by the following license notice:
   9 #
  10 #   Licensed to the Apache Software Foundation (ASF) under one or more
  11 #   contributor license agreements. See the NOTICE file distributed
  12 #   with this work for additional information regarding copyright
  13 #   ownership. The ASF licenses this file to you under the Apache
  14 #   License, Version 2.0 (the "License"); you may not use this file
  15 #   except in compliance with the License. You may obtain a copy of
  16 #   the License at http://www.apache.org/licenses/LICENSE-2.0 .
  17 #
  18 "tests bridging python implementations of UNO objects"
  19 import unittest
  20 import uno
  21 import unohelper
  22 import os
  23 import sys
  24
  25 from com.sun.star.io import XOutputStream, XInputStream, typeOfXOutputStream, typeOfXInputStream
  26 from com.sun.star.lang import XTypeProvider, typeOfXTypeProvider, XEventListener
  27 from com.sun.star.uno import XCurrentContext
  28
  29 class SequenceOutputStream( unohelper.Base, XOutputStream ):
  30       def __init__( self ):
  31           self.s = uno.ByteSequence("")
  32           self.closed = 0
  33
  34       def closeOutput(self):
  35           self.closed = 1
  36
  37       def writeBytes( self, seq ):
  38           self.s = self.s + seq
  39
  40       def flush( self ):
  41           pass
  42
  43       def getSequence( self ):
  44           return self.s
  45
  46
  47 class SequenceInputStream( XInputStream, unohelper.Base ):
  48       def __init__( self, seq ):
  49           self.s = seq
  50           self.nIndex = 0
  51           self.closed = 0
  52
  53       def closeInput( self):
  54           self.closed = 1
  55           self.s = None
  56
  57       def skipBytes( self, nByteCount ):
  58           if( nByteCount + self.nIndex > len(self.s) ):
  59               nByteCount = len(self.s) - self.nIndex
  60           self.nIndex += nByteCount
  61
  62       def readBytes( self, retSeq, nByteCount ):
  63           nRet = 0
  64           if( self.nIndex + nByteCount > len(self.s) ):
  65               nRet = len(self.s) - self.nIndex
  66           else:
  67               nRet = nByteCount
  68           retSeq = uno.ByteSequence(self.s.value[self.nIndex : self.nIndex + nRet ])
  69           self.nIndex = self.nIndex + nRet
  70           return nRet, retSeq
  71
  72       def readSomeBytes( self, retSeq , nByteCount ):
  73           #as we never block !
  74           return readBytes( retSeq, nByteCount )
  75
  76       def available( self ):
  77           return len( self.s ) - self.nIndex
  78
  79 class SequenceInputStream2( SequenceInputStream ):
  80       def __init__( self, seq ):
  81             SequenceInputStream.__init__( self, seq )
  82
  83 class TestCase(unittest.TestCase):
  84       def __init__(self,method,ctx):
  85           unittest.TestCase.__init__(self,method)
  86           self.ctx = ctx
  87
  88       def setUp(self):
  89           self.tobj = self.ctx.ServiceManager.createInstanceWithContext( \
  90                            "com.sun.star.test.bridge.CppTestObject",self.ctx)
  91           self.pipe = self.ctx.ServiceManager.createInstanceWithContext( \
  92                            "com.sun.star.io.Pipe" , self.ctx )
  93
  94       def testStandard( self ):
  95           dataOut = self.ctx.ServiceManager.createInstanceWithContext( \
  96                         "com.sun.star.io.DataOutputStream", self.ctx )
  97           streamOut = SequenceOutputStream()
  98           dataOut.setOutputStream( streamOut )
  99           dataOut.writeShort( 42 )
 100           dataOut.writeLong( 43 )
 101           dataOut.closeOutput()
 102
 103           dataInput = self.ctx.ServiceManager.createInstanceWithContext( \
 104                    "com.sun.star.io.DataInputStream", self.ctx )
 105
 106           dataInput.setInputStream( SequenceInputStream2( streamOut.getSequence() ) )
 107
 108           self.failUnless( 42 == dataInput.readShort() )
 109           self.failUnless( 43 == dataInput.readLong() )
 110           self.failUnless( self.tobj.transportAny( streamOut ) == streamOut )
 111
 112
 113 class NullDevice:
 114       def write( self, string ):
 115             pass
 116
 117
 118 class EventListener( unohelper.Base, XEventListener ):
 119     def __init__( self ):
 120         self.disposingCalled = False
 121
 122     def disposing( self , eventObject ):
 123         self.disposingCalled = True
 124
 125 class TestHelperCase( unittest.TestCase ):
 126
 127       def __init__(self,method):
 128             unittest.TestCase.__init__(self,method)
 129
 130       def testUrlHelper( self ):
 131             systemPath = os.getcwd()
 132             if systemPath.startswith( "/" ):
 133                   self.failUnless( "/tmp" == unohelper.fileUrlToSystemPath( "file:///tmp" ) )
 134                   self.failUnless( "file:///tmp" == unohelper.systemPathToFileUrl( "/tmp" ))
 135             else:
 136                   self.failUnless( "c:\\temp" == unohelper.fileUrlToSystemPath( "file:///c:/temp" ) )
 137                   self.failUnless( "file:///c:/temp" == unohelper.systemPathToFileUrl( "c:\\temp" ) )
 138
 139             systemPath = unohelper.systemPathToFileUrl( systemPath )
 140             self.failUnless( systemPath + "/a" == unohelper.absolutize( systemPath, "a" ) )
 141       def testInspect( self ):
 142             dev = NullDevice()
 143 #            dev = sys.stdout
 144             unohelper.inspect( uno.getComponentContext() , dev )
 145             unohelper.inspect( uno.getComponentContext().ServiceManager , dev )
 146             unohelper.inspect( uno.getTypeByName( "com.sun.star.lang.XComponent" ) , dev )
 147
 148       def testListener( self ):
 149             smgr = uno.getComponentContext().ServiceManager.createInstance(
 150                   "com.sun.star.lang.ServiceManager" )
 151
 152             # check, whether listeners
 153             listener = EventListener()
 154             smgr.addEventListener( listener )
 155             smgr.dispose()
 156             self.failUnless( listener.disposingCalled )
 157
 158             # check, whether listeners can be removed
 159             smgr = uno.getComponentContext().ServiceManager.createInstance(
 160                   "com.sun.star.lang.ServiceManager" )
 161             listener = EventListener()
 162             smgr.addEventListener( listener )
 163             smgr.removeEventListener( listener )
 164             smgr.dispose()
 165             self.failUnless( not listener.disposingCalled )
 166
 167       def testCurrentContext( self ):
 168             oldContext = uno.getCurrentContext()
 169             try:
 170                   uno.setCurrentContext(
 171                         unohelper.CurrentContext( oldContext,{"My42":42}) )
 172                   self.failUnless( 42 == uno.getCurrentContext().getValueByName( "My42" ) )
 173                   self.failUnless( None == uno.getCurrentContext().getValueByName( "My43" ) )
 174             finally:
 175                   uno.setCurrentContext( oldContext )
 176
 177
 178
 179 def suite( ctx ):
 180     suite = unittest.TestSuite()
 181     suite.addTest(TestCase("testStandard",ctx))
 182     suite.addTest(TestHelperCase( "testUrlHelper" ))
 183     suite.addTest(TestHelperCase( "testInspect" ))
 184     suite.addTest(TestHelperCase( "testListener" ) )
 185     suite.addTest(TestHelperCase( "testCurrentContext" ) )
 186     return suite
 187