← Back to branch summary

~ubuntu-branches/debian/sid/pyro/sid

« back to all changes in this revision

Viewing changes to tests/PyroTests/coretests.py

  • Committer: Bazaar Package Importer
  • Author(s): Carl Chenet, Carl Chenet, Jakub Wilk
  • Date: 2010-09-14 01:04:28 UTC
  • Revision ID: james.westby@ubuntu.com-20100914010428-02r7p1rzr7jvw94z
Tags: 1:3.9.1-2
http://bugs.debian.org/589172
http://bugs.debian.org/589053
[Carl Chenet]
* revert to 3.9.1-1 package because of the development status 
  of the 4.1 package is unsuitable for stable use
  DPMT svn #8557 revision (Closes: #589172) 
* added debian/source
* added debian/source/format
* package is now 3.0 (quilt) source format
* debian/control
  - Bump Standards-Version to 3.9.1

[Jakub Wilk]
* Add ‘XS-Python-Version: >= 2.5’ to prevent bytecompilation with python2.4
  (closes: #589053).

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/PyroTests/coretests.py
1
 
"""
2
 
Tests for the core logic.
3
 
 
4
 
Pyro - Python Remote Objects.  Copyright by Irmen de Jong.
5
 
irmen@razorvine.net - http://www.razorvine.net/python/Pyro
6
 
"""
7
 
 
8
 
from __future__ import with_statement
9
 
import unittest
10
 
import copy
11
 
import Pyro.core
12
 
import Pyro.config
13
 
import Pyro.errors
14
 
 
15
 
class Thing(object):
16
 
    def __init__(self, arg):
17
 
        self.arg=arg
18
 
    def __eq__(self,other):
19
 
        return self.arg==other.arg
20
 
    __hash__=object.__hash__
21
 
 
22
 
class CoreTests(unittest.TestCase):
23
 
    def testConfig(self):
24
 
        self.assertTrue(type(Pyro.config.COMPRESSION) is bool)
25
 
        self.assertTrue(type(Pyro.config.NS_PORT) is int)
26
 
        config=Pyro.config.asDict()
27
 
        self.assertTrue(type(config) is dict)
28
 
        self.assertTrue("COMPRESSION" in config)
29
 
        self.assertEqual(Pyro.config.COMPRESSION, config["COMPRESSION"])
30
 
 
31
 
    def testImports(self):
32
 
        self.assertTrue(Pyro.Proxy is Pyro.core.Proxy)
33
 
        self.assertTrue(Pyro.URI is Pyro.core.URI)
34
 
        self.assertTrue(Pyro.Daemon is Pyro.core.Daemon)
35
 
        self.assertTrue(Pyro.locateNS is Pyro.naming.locateNS)
36
 
        self.assertTrue(Pyro.resolve is Pyro.naming.resolve)
37
 
        
38
 
    def testUriStrAndRepr(self):
39
 
        uri="PYRONAME:some_obj_name"
40
 
        p=Pyro.core.URI(uri)
41
 
        self.assertEqual(uri,str(p))
42
 
        uri="PYRONAME:some_obj_name@host.com"
43
 
        p=Pyro.core.URI(uri)
44
 
        self.assertEqual(uri+":"+str(Pyro.config.NS_PORT),str(p))   # a PYRONAME uri with a hostname gets a port too if omitted
45
 
        uri="PYRONAME:some_obj_name@host.com:8888"
46
 
        p=Pyro.core.URI(uri)
47
 
        self.assertEqual(uri,str(p))
48
 
        self.assertTrue("URI object at" in repr(p))
49
 
        uri="PYRO:12345@host.com:9999"
50
 
        p=Pyro.core.URI(uri)
51
 
        self.assertEqual(uri,str(p))
52
 
        self.assertEqual(uri,p.asString())
53
 
        uri="PYRO:12345@./p:pipename"
54
 
        p=Pyro.core.URI(uri)
55
 
        self.assertEqual(uri,str(p))
56
 
        uri="PYRO:12345@./u:sockname"
57
 
        p=Pyro.core.URI(uri)
58
 
        self.assertEqual(uri,str(p))
59
 
        uri="PYRO:12345@./u:sockname"
60
 
        unicodeuri=unicode(uri)
61
 
        p=Pyro.core.URI(unicodeuri)
62
 
        self.assertEqual(uri,str(p))
63
 
        self.assertEqual(unicodeuri,unicode(p))
64
 
        self.assertTrue(type(p.sockname) is unicode)
65
 
 
66
 
    def testUriParsingPyro(self):
67
 
        p=Pyro.core.URI("PYRONAME:some_obj_name")
68
 
        self.assertEqual("PYRONAME",p.protocol)
69
 
        self.assertEqual("some_obj_name",p.object)
70
 
        self.assertEqual(None,p.host)
71
 
        self.assertEqual(None,p.pipename)
72
 
        self.assertEqual(None,p.sockname)
73
 
        self.assertEqual(None,p.port)
74
 
        p=Pyro.core.URI("PYRONAME:some_obj_name@host.com:9999")
75
 
        self.assertEqual("PYRONAME",p.protocol)
76
 
        self.assertEqual("some_obj_name",p.object)
77
 
        self.assertEqual("host.com",p.host)
78
 
        self.assertEqual(9999,p.port)
79
 
 
80
 
        p=Pyro.core.URI("PYRO:12345@host.com:4444")
81
 
        self.assertEqual("PYRO",p.protocol)
82
 
        self.assertEqual("12345",p.object)
83
 
        self.assertEqual("host.com",p.host)
84
 
        self.assertEqual(None,p.pipename)
85
 
        self.assertEqual(None,p.sockname)
86
 
        self.assertEqual(4444,p.port)
87
 
        p=Pyro.core.URI("PYRO:12345@./p:pipename")
88
 
        self.assertEqual("12345",p.object)
89
 
        self.assertEqual("pipename",p.pipename)
90
 
        p=Pyro.core.URI("PYRO:12345@./u:sockname")
91
 
        self.assertEqual("12345",p.object)
92
 
        self.assertEqual("sockname",p.sockname)
93
 
 
94
 
    def testUriParsingPyroname(self):
95
 
        p=Pyro.core.URI("PYRONAME:objectname")
96
 
        self.assertEqual("PYRONAME",p.protocol)
97
 
        self.assertEqual("objectname",p.object)
98
 
        self.assertEqual(None,p.host)
99
 
        self.assertEqual(None,p.port)
100
 
        p=Pyro.core.URI("PYRONAME:objectname@nameserverhost")
101
 
        self.assertEqual("PYRONAME",p.protocol)
102
 
        self.assertEqual("objectname",p.object)
103
 
        self.assertEqual("nameserverhost",p.host)
104
 
        self.assertEqual(Pyro.config.NS_PORT,p.port)   # Pyroname uri with host gets a port too if not specified
105
 
        p=Pyro.core.URI("PYRONAME:objectname@nameserverhost:4444")
106
 
        self.assertEqual("PYRONAME",p.protocol)
107
 
        self.assertEqual("objectname",p.object)
108
 
        self.assertEqual("nameserverhost",p.host)
109
 
        self.assertEqual(4444,p.port)
110
 
 
111
 
    def testInvalidUris(self):
112
 
        self.assertRaises(TypeError, Pyro.core.URI, None)
113
 
        self.assertRaises(TypeError, Pyro.core.URI, 99999)
114
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "")
115
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "a")
116
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO")
117
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:")
118
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO::")
119
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:a")
120
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:x@")
121
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:x@hostname")
122
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:@hostname:portstr")
123
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:@hostname:7766")
124
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:objid@hostname:7766:bogus")
125
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYROLOC:objname")
126
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYROLOC:objname@host")
127
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYROLOC:objectname@hostname:4444")
128
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRONAME:")
129
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRONAME:objname@nameserver:bogus")
130
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRONAME:objname@nameserver:7766:bogus")
131
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "FOOBAR:")
132
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "FOOBAR:objid@hostname:7766")
133
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:12345@./p:pipename/slash")
134
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:12345@./u:sockname/slash")
135
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:12345@./p:pipename:9999")
136
 
        self.assertRaises(Pyro.errors.PyroError, Pyro.core.URI, "PYRO:12345@./u:sockname:9999")
137
 
 
138
 
    def testUriUnicode(self):
139
 
        p=Pyro.core.URI(u"PYRO:12345@host.com:4444") 
140
 
        self.assertEqual("PYRO",p.protocol)
141
 
        self.assertEqual("12345",p.object)
142
 
        self.assertEqual("host.com",p.host)
143
 
        self.assertTrue(type(p.protocol) is unicode)
144
 
        self.assertTrue(type(p.object) is unicode)
145
 
        self.assertTrue(type(p.host) is unicode)
146
 
        self.assertEqual(None,p.pipename)
147
 
        self.assertEqual(None,p.sockname)
148
 
        self.assertEqual(4444,p.port)
149
 
 
150
 
        uri="PYRO:12345@hostname:9999"
151
 
        p=Pyro.core.URI(uri)
152
 
        pu=Pyro.core.URI(unicode(uri))
153
 
        self.assertEqual("PYRO",pu.protocol)
154
 
        self.assertEqual("hostname",pu.host)
155
 
        self.assertEqual(p,pu)
156
 
        self.assertEqual(str(p), str(pu))
157
 
        unicodeuri=u"PYRO:weirdchars\u20AC@host\u20AC.com:4444"
158
 
        pu=Pyro.core.URI(unicodeuri)
159
 
        self.assertEqual("PYRO",pu.protocol)
160
 
        self.assertEqual(u"host\u20AC.com",pu.host)
161
 
        self.assertEqual(u"weirdchars\u20AC",pu.object)
162
 
        self.assertEqual(pu.asString(), pu.__str__())
163
 
        self.assertEqual(u"PYRO:weirdchars\u20ac@host\u20ac.com:4444", pu.asString())
164
 
        self.assertEqual(u"PYRO:weirdchars\u20ac@host\u20ac.com:4444", unicode(pu))
165
 
        self.assertTrue("URI object at" in repr(pu))
166
 
    
167
 
    def testUriCopy(self):
168
 
        p1=Pyro.core.URI("PYRO:12345@hostname:9999")
169
 
        p2=Pyro.core.URI(p1)
170
 
        p3=copy.copy(p1)
171
 
        self.assertEqual(p1.protocol, p2.protocol)
172
 
        self.assertEqual(p1.host, p2.host)
173
 
        self.assertEqual(p1.port, p2.port)
174
 
        self.assertEqual(p1.object, p2.object)
175
 
        self.assertEqual(p1,p2)
176
 
        self.assertEqual(p1.protocol, p3.protocol)
177
 
        self.assertEqual(p1.host, p3.host)
178
 
        self.assertEqual(p1.port, p3.port)
179
 
        self.assertEqual(p1.object, p3.object)
180
 
        self.assertEqual(p1,p3)
181
 
        
182
 
    def testUriEqual(self):
183
 
        p1=Pyro.core.URI("PYRO:12345@host.com:9999")
184
 
        p2=Pyro.core.URI("PYRO:12345@host.com:9999")
185
 
        p3=Pyro.core.URI("PYRO:99999@host.com:4444")
186
 
        self.assertEqual(p1,p2)
187
 
        self.assertNotEqual(p1,p3)
188
 
        self.assertNotEqual(p2,p3)
189
 
        p2.port=4444
190
 
        p2.object="99999"
191
 
        self.assertNotEqual(p1,p2)
192
 
        self.assertEqual(p2,p3)
193
 
        
194
 
    def testLocation(self):
195
 
        self.assertTrue(Pyro.core.URI.isPipeOrUnixsockLocation("./p:name"))
196
 
        self.assertTrue(Pyro.core.URI.isPipeOrUnixsockLocation("./u:name"))
197
 
        self.assertFalse(Pyro.core.URI.isPipeOrUnixsockLocation("./x:name"))
198
 
        self.assertFalse(Pyro.core.URI.isPipeOrUnixsockLocation("foobar"))
199
 
 
200
 
    def testMsgFactory(self):
201
 
        MF=Pyro.core.MessageFactory
202
 
        MF.createMessage(99, None) # doesn't check msg type here
203
 
        self.assertRaises(Pyro.errors.ProtocolError, MF.parseMessageHeader, "FOOBAR")
204
 
        hdr=MF.createMessage(MF.MSG_CONNECT, "hello")[:-5]
205
 
        msgType,flags,dataLen=MF.parseMessageHeader(hdr)
206
 
        self.assertEqual(MF.MSG_CONNECT, msgType)
207
 
        self.assertEqual(0, flags)
208
 
        self.assertEqual(5, dataLen)
209
 
        hdr=MF.createMessage(MF.MSG_RESULT, None)
210
 
        msgType,flags,dataLen=MF.parseMessageHeader(hdr)
211
 
        self.assertEqual(MF.MSG_RESULT, msgType)
212
 
        self.assertEqual(0, flags)
213
 
        self.assertEqual(0, dataLen)
214
 
        hdr=MF.createMessage(MF.MSG_RESULT, "hello", 42)[:-5]
215
 
        msgType,flags,dataLen=MF.parseMessageHeader(hdr)
216
 
        self.assertEqual(MF.MSG_RESULT, msgType)
217
 
        self.assertEqual(42, flags)
218
 
        self.assertEqual(5, dataLen)
219
 
        msg=MF.createMessage(255,None)
220
 
        self.assertEqual("PYRO\x00"+chr(Pyro.constants.PROTOCOL_VERSION)+"\x00\xff\x00\x00\x00\x00\x00\x00",msg)
221
 
        msg=MF.createMessage(1,None)
222
 
        self.assertEqual("PYRO\x00"+chr(Pyro.constants.PROTOCOL_VERSION)+"\x00\x01\x00\x00\x00\x00\x00\x00",msg)
223
 
        msg=MF.createMessage(1,None,flags=255)
224
 
        self.assertEqual("PYRO\x00"+chr(Pyro.constants.PROTOCOL_VERSION)+"\x00\x01\x00\xff\x00\x00\x00\x00",msg)
225
 
        # compression is a job of the code supplying the data, so the messagefactory should leave it untouched
226
 
        data="x"*1000
227
 
        msg=MF.createMessage(MF.MSG_INVOKE, data, 0)
228
 
        msg2=MF.createMessage(MF.MSG_INVOKE, data, MF.FLAGS_COMPRESSED)
229
 
        self.assertEquals(len(msg),len(msg2))
230
 
 
231
 
    def testProxyOffline(self):
232
 
        # only offline stuff here.
233
 
        # online stuff needs a running daemon, so we do that in another test, to keep this one simple
234
 
        self.assertRaises(TypeError, Pyro.core.Proxy, 999)  # wrong arg
235
 
        p1=Pyro.core.Proxy("PYRO:9999@localhost:15555")
236
 
        p2=Pyro.core.Proxy(Pyro.core.URI("PYRO:9999@localhost:15555"))
237
 
        self.assertEqual(p1._pyroUri, p2._pyroUri)
238
 
        self.assertTrue(p1._pyroConnection is None)
239
 
        p1._pyroRelease()
240
 
        p1._pyroRelease()
241
 
        # try copying a not-connected proxy
242
 
        p3=copy.copy(p1)
243
 
        self.assertTrue(p3._pyroConnection is None)
244
 
        self.assertTrue(p1._pyroConnection is None)
245
 
        self.assertEqual(p3._pyroUri, p1._pyroUri)
246
 
        self.assertFalse(p3._pyroUri is p1._pyroUri)
247
 
        self.assertEqual(p3._pyroSerializer, p1._pyroSerializer)
248
 
        self.assertTrue(p3._pyroSerializer is p1._pyroSerializer)
249
 
 
250
 
    def testProxyStr(self):
251
 
        p=Pyro.core.Proxy("PYRO:9999@localhost:15555")
252
 
        self.assertEqual("<Pyro Proxy for PYRO:9999@localhost:15555>", str(p))
253
 
        self.assertEqual(u"<Pyro Proxy for PYRO:9999@localhost:15555>", unicode(p))
254
 
        self.assertTrue("Proxy object at" in repr(p))
255
 
        
256
 
    def testProxySettings(self):
257
 
        p1=Pyro.core.Proxy("PYRO:9999@localhost:15555")
258
 
        p2=Pyro.core.Proxy("PYRO:9999@localhost:15555")
259
 
        p1._pyroOneway.add("method")
260
 
        self.assertTrue("method" in p1._pyroOneway, "p1 should have oneway method")
261
 
        self.assertFalse("method" in p2._pyroOneway, "p2 should not have the same oneway method")
262
 
        self.assertFalse(p1._pyroOneway is p2._pyroOneway, "p1 and p2 should have different oneway tables")
263
 
        
264
 
    def testProxyWithStmt(self):
265
 
        class ConnectionMock(object):
266
 
            closeCalled=False
267
 
            def close(self):
268
 
                self.closeCalled=True
269
 
 
270
 
        connMock=ConnectionMock()
271
 
        p=Pyro.core.Proxy("PYRO:9999@localhost:15555")
272
 
        p._pyroConnection=connMock
273
 
        self.assertFalse(connMock.closeCalled)
274
 
        p._pyroRelease()
275
 
        self.assertTrue(p._pyroConnection is None)
276
 
        self.assertTrue(connMock.closeCalled)
277
 
        
278
 
        connMock=ConnectionMock()
279
 
        with Pyro.core.Proxy("PYRO:9999@localhost:15555") as p:
280
 
            p._pyroConnection=connMock
281
 
        self.assertTrue(p._pyroConnection is None)
282
 
        self.assertTrue(connMock.closeCalled)
283
 
        connMock=ConnectionMock()
284
 
        try:
285
 
            with Pyro.core.Proxy("PYRO:9999@localhost:15555") as p:
286
 
                p._pyroConnection=connMock
287
 
                print 1//0  # cause an error
288
 
            self.fail("expected error")
289
 
        except ZeroDivisionError:
290
 
            pass
291
 
        self.assertTrue(p._pyroConnection is None)
292
 
        self.assertTrue(connMock.closeCalled)
293
 
        connMock=ConnectionMock()
294
 
        p=Pyro.core.Proxy("PYRO:9999@localhost:15555")
295
 
        with p:
296
 
            self.assertTrue(p._pyroUri is not None)
297
 
        with p:
298
 
            self.assertTrue(p._pyroUri is not None)            
299
 
 
300
 
    def testRemoteMethod(self):
301
 
        class Proxy(object):
302
 
            def invoke(self, name, args, kwargs):
303
 
                return "INVOKED name=%s args=%s kwargs=%s" % (name,args,kwargs)
304
 
            def __getattr__(self, name):
305
 
                return Pyro.core._RemoteMethod(self.invoke, name)
306
 
        o=Proxy()
307
 
        self.assertEqual("INVOKED name=foo args=(1,) kwargs={}", o.foo(1)) #normal
308
 
        self.assertEqual("INVOKED name=foo.bar args=(1,) kwargs={}", o.foo.bar(1)) #dotted
309
 
        self.assertEqual("INVOKED name=foo.bar args=(1, 'hello') kwargs={'a': True}", o.foo.bar(1,"hello",a=True))
310
 
        p=Pyro.core.Proxy("PYRO:obj@host:666")
311
 
        a=p.someattribute
312
 
        self.assertTrue(isinstance(a, Pyro.core._RemoteMethod), "attribute access should just be a RemoteMethod")
313
 
        a2=a.nestedattribute
314
 
        self.assertTrue(isinstance(a2, Pyro.core._RemoteMethod), "nested attribute should just be another RemoteMethod")
315
 
 
316
 
    def testTimeoutGetSet(self):
317
 
        class ConnectionMock(object):
318
 
            def __init__(self):
319
 
                self.timeout=Pyro.config.COMMTIMEOUT
320
 
            def close(self):
321
 
                pass
322
 
        Pyro.config.COMMTIMEOUT=None
323
 
        p=Pyro.core.Proxy("PYRO:obj@host:555")
324
 
        self.assertEqual(None, p._pyroTimeout)
325
 
        p._pyroTimeout=5
326
 
        self.assertEqual(5, p._pyroTimeout)
327
 
        p=Pyro.core.Proxy("PYRO:obj@host:555")
328
 
        p._pyroConnection=ConnectionMock()
329
 
        self.assertEqual(None, p._pyroTimeout)
330
 
        p._pyroTimeout=5
331
 
        self.assertEqual(5, p._pyroTimeout)
332
 
        self.assertEqual(5, p._pyroConnection.timeout)
333
 
        Pyro.config.COMMTIMEOUT=2
334
 
        p=Pyro.core.Proxy("PYRO:obj@host:555")
335
 
        p._pyroConnection=ConnectionMock()
336
 
        self.assertEqual(2, p._pyroTimeout)
337
 
        self.assertEqual(2, p._pyroConnection.timeout)
338
 
        p._pyroTimeout=None
339
 
        self.assertEqual(None, p._pyroTimeout)
340
 
        self.assertEqual(None, p._pyroConnection.timeout)
341
 
        Pyro.config.COMMTIMEOUT=None
342
 
 
343
 
    def testDecorators(self):
344
 
        # just test the decorator itself, testing the callback
345
 
        # exception handling is kinda hard in unit tests. Maybe later.
346
 
        class Test(object):
347
 
            @Pyro.callback
348
 
            def method(self):
349
 
                pass
350
 
            def method2(self):
351
 
                pass
352
 
        t=Test()
353
 
        self.assertEquals(True, getattr(t.method,"_pyroCallback"))
354
 
        self.assertEquals(False, getattr(t.method2,"_pyroCallback", False))
355
 
        
356
 
        
357
 
if __name__ == "__main__":
358
 
    #import sys;sys.argv = ['', 'Test.testName']
359
 
    unittest.main()