Ticket #1879: testnis.py

File testnis.py, 2.1 KB (added by pgstudy, 16 years ago)

test for patch

Line 
1import unittest
2import sys
3#sys.path.insert(1,"/home/linlove/workspace/twistedcheckout/trunk/")
4from twisted.spread import pb, jelly
5from twisted.internet import reactor
6
7class Pond:
8  def setRemote(self,remote):
9    self.remote=remote
10    pass
11
12class Receiver(pb.Root):
13    def remote_takePond(self, pond):
14        print " got pond:", pond
15        pond.remote.callRemote("takePond2",Pond()).addCallback(self.ok).addErrback(self.notOk)
16        return "safe and sound" # positive acknowledgement
17   
18    def ok(self, response):
19        print "pond arrived2", response
20        reactor.stop()
21    def notOk(self, failure):
22        print "error during takePond2:"
23        if failure.type == jelly.InsecureJelly:
24            print " InsecureJelly2"
25        else:
26            print failure
27        return None
28        reactor.stop()
29 
30class Sender(pb.Referenceable):
31    def __init__(self, pond):
32        self.pond = pond
33    def remote_takePond2(self, pond):
34        print " got pond2:", pond
35    def got_obj(self, obj):
36        d = obj.callRemote("takePond", self.pond)
37        d.addCallback(self.ok).addErrback(self.notOk)
38    def ok(self, response):
39        print "pond arrived", response
40    def notOk(self, failure):
41        print "error during takePond:"
42        if failure.type == jelly.InsecureJelly:
43            print " InsecureJelly"
44        else:
45            print failure
46        return None
47        reactor.stop()
48
49
50
51class PBWithSecurityOptionsTest(unittest.TestCase):
52
53    def testSecOptions(self):
54      import copy
55      gs=copy.copy(jelly.globalSecurity)
56      gs.isModuleAllowed=jelly.DummySecurityOptions().isModuleAllowed
57      gs.isClassAllowed=jelly.DummySecurityOptions().isClassAllowed
58      gs.isTypeAllowed=jelly.DummySecurityOptions().isTypeAllowed
59
60      factory = pb.PBClientFactory(gs)
61      serverFactory=pb.PBServerFactory(Receiver(),security=gs)
62      reactor.connectTCP("localhost", 8801, factory)
63      deferred = factory.getRootObject()
64      p=Pond()
65      sender = Sender(p)
66      p.setRemote(sender)
67      deferred.addCallback(sender.got_obj)
68      reactor.listenTCP(8801, serverFactory)
69      reactor.run()
70
71if __name__ == "__main__":
72  unittest.main()