testtools/source/bridgetest/pyuno/impl.py

   1 #*************************************************************************
   2 #
   3 # DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4 #
   5 # Copyright 2008 by Sun Microsystems, Inc.
   6 #
   7 # OpenOffice.org - a multi-platform office productivity suite
   8 #
   9 # $RCSfile: impl.py,v $
  10 #
  11 # $Revision: 1.6 $
  12 #
  13 # This file is part of OpenOffice.org.
  14 #
  15 # OpenOffice.org is free software: you can redistribute it and/or modify
  16 # it under the terms of the GNU Lesser General Public License version 3
  17 # only, as published by the Free Software Foundation.
  18 #
  19 # OpenOffice.org is distributed in the hope that it will be useful,
  20 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  21 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  22 # GNU Lesser General Public License version 3 for more details
  23 # (a copy is included in the LICENSE file that accompanied this code).
  24 #
  25 # You should have received a copy of the GNU Lesser General Public License
  26 # version 3 along with OpenOffice.org.  If not, see
  27 # <http://www.openoffice.org/license.html>
  28 # for a copy of the LGPLv3 License.
  29 #
  30 #*************************************************************************
  31 "tests bridging python implementations of UNO objects"
  32 import unittest
  33 import uno
  34 import unohelper
  35 import os
  36 import sys
  37
  38 from com.sun.star.io import XOutputStream, XInputStream, typeOfXOutputStream, typeOfXInputStream
  39 from com.sun.star.lang import XTypeProvider, typeOfXTypeProvider, XEventListener
  40 from com.sun.star.uno import XCurrentContext
  41
  42 class SequenceOutputStream( unohelper.Base, XOutputStream ):
  43       def __init__( self ):
  44           self.s = uno.ByteSequence("")
  45           self.closed = 0
  46
  47       def closeOutput(self):
  48           self.closed = 1
  49
  50       def writeBytes( self, seq ):
  51           self.s = self.s + seq
  52
  53       def flush( self ):
  54           pass
  55
  56       def getSequence( self ):
  57           return self.s
  58
  59
  60 class SequenceInputStream( XInputStream, unohelper.Base ):
  61       def __init__( self, seq ):
  62           self.s = seq
  63           self.nIndex = 0
  64           self.closed = 0
  65
  66       def closeInput( self):
  67           self.closed = 1
  68           self.s = None
  69
  70       def skipBytes( self, nByteCount ):
  71           if( nByteCount + self.nIndex > len(self.s) ):
  72               nByteCount = len(self.s) - self.nIndex
  73           self.nIndex += nByteCount
  74
  75       def readBytes( self, retSeq, nByteCount ):
  76           nRet = 0
  77           if( self.nIndex + nByteCount > len(self.s) ):
  78               nRet = len(self.s) - self.nIndex
  79           else:
  80               nRet = nByteCount
  81           retSeq = uno.ByteSequence(self.s.value[self.nIndex : self.nIndex + nRet ])
  82           self.nIndex = self.nIndex + nRet
  83           return nRet, retSeq
  84
  85       def readSomeBytes( self, retSeq , nByteCount ):
  86           #as we never block !
  87           return readBytes( retSeq, nByteCount )
  88
  89       def available( self ):
  90           return len( self.s ) - self.nIndex
  91
  92 class SequenceInputStream2( SequenceInputStream ):
  93       def __init__( self, seq ):
  94             SequenceInputStream.__init__( self, seq )
  95
  96 class TestCase(unittest.TestCase):
  97       def __init__(self,method,ctx):
  98           unittest.TestCase.__init__(self,method)
  99           self.ctx = ctx
 100
 101       def setUp(self):
 102           self.tobj = self.ctx.ServiceManager.createInstanceWithContext( \
 103                            "com.sun.star.test.bridge.CppTestObject",self.ctx)
 104           self.pipe = self.ctx.ServiceManager.createInstanceWithContext( \
 105                            "com.sun.star.io.Pipe" , self.ctx )
 106
 107       def testStandard( self ):
 108           dataOut = self.ctx.ServiceManager.createInstanceWithContext( \
 109                         "com.sun.star.io.DataOutputStream", self.ctx )
 110           streamOut = SequenceOutputStream()
 111           dataOut.setOutputStream( streamOut )
 112           dataOut.writeShort( 42 )
 113           dataOut.writeLong( 43 )
 114           dataOut.closeOutput()
 115
 116           dataInput = self.ctx.ServiceManager.createInstanceWithContext( \
 117                    "com.sun.star.io.DataInputStream", self.ctx )
 118
 119           dataInput.setInputStream( SequenceInputStream2( streamOut.getSequence() ) )
 120
 121           self.failUnless( 42 == dataInput.readShort() )
 122           self.failUnless( 43 == dataInput.readLong() )
 123           self.failUnless( self.tobj.transportAny( streamOut ) == streamOut )
 124
 125
 126 class NullDevice:
 127       def write( self, string ):
 128             pass
 129
 130
 131 class EventListener( unohelper.Base, XEventListener ):
 132     def __init__( self ):
 133         self.disposingCalled = False
 134
 135     def disposing( self , eventObject ):
 136         self.disposingCalled = True
 137
 138 class TestHelperCase( unittest.TestCase ):
 139
 140       def __init__(self,method):
 141             unittest.TestCase.__init__(self,method)
 142
 143       def testUrlHelper( self ):
 144             systemPath = os.getcwd()
 145             if systemPath.startswith( "/" ):
 146                   self.failUnless( "/tmp" == unohelper.fileUrlToSystemPath( "file:///tmp" ) )
 147                   self.failUnless( "file:///tmp" == unohelper.systemPathToFileUrl( "/tmp" ))
 148             else:
 149                   self.failUnless( "c:\\temp" == unohelper.fileUrlToSystemPath( "file:///c:/temp" ) )
 150                   self.failUnless( "file:///c:/temp" == unohelper.systemPathToFileUrl( "c:\\temp" ) )
 151
 152             systemPath = unohelper.systemPathToFileUrl( systemPath )
 153             self.failUnless( systemPath + "/a" == unohelper.absolutize( systemPath, "a" ) )
 154       def testInspect( self ):
 155             dev = NullDevice()
 156 #            dev = sys.stdout
 157             unohelper.inspect( uno.getComponentContext() , dev )
 158             unohelper.inspect( uno.getComponentContext().ServiceManager , dev )
 159             unohelper.inspect( uno.getTypeByName( "com.sun.star.lang.XComponent" ) , dev )
 160
 161       def testListener( self ):
 162             smgr = uno.getComponentContext().ServiceManager.createInstance(
 163                   "com.sun.star.lang.ServiceManager" )
 164
 165             # check, whether listeners
 166             listener = EventListener()
 167             smgr.addEventListener( listener )
 168             smgr.dispose()
 169             self.failUnless( listener.disposingCalled )
 170
 171             # check, whether listeners can be removed
 172             smgr = uno.getComponentContext().ServiceManager.createInstance(
 173                   "com.sun.star.lang.ServiceManager" )
 174             listener = EventListener()
 175             smgr.addEventListener( listener )
 176             smgr.removeEventListener( listener )
 177             smgr.dispose()
 178             self.failUnless( not listener.disposingCalled )
 179
 180       def testCurrentContext( self ):
 181             oldContext = uno.getCurrentContext()
 182             try:
 183                   uno.setCurrentContext(
 184                         unohelper.CurrentContext( oldContext,{"My42":42}) )
 185                   self.failUnless( 42 == uno.getCurrentContext().getValueByName( "My42" ) )
 186                   self.failUnless( None == uno.getCurrentContext().getValueByName( "My43" ) )
 187             finally:
 188                   uno.setCurrentContext( oldContext )
 189
 190
 191
 192 def suite( ctx ):
 193     suite = unittest.TestSuite()
 194     suite.addTest(TestCase("testStandard",ctx))
 195     suite.addTest(TestHelperCase( "testUrlHelper" ))
 196     suite.addTest(TestHelperCase( "testInspect" ))
 197     suite.addTest(TestHelperCase( "testListener" ) )
 198     suite.addTest(TestHelperCase( "testCurrentContext" ) )
 199     return suite
 200