| OLD | NEW |
| (Empty) |
| 1 # | |
| 2 # Copyright (c) 2001-2004 Twisted Matrix Laboratories. | |
| 3 # See LICENSE for details. | |
| 4 # | |
| 5 | |
| 6 """Test SOAP support.""" | |
| 7 | |
| 8 try: | |
| 9 import SOAPpy | |
| 10 except ImportError: | |
| 11 SOAPpy = None | |
| 12 class SOAPPublisher: pass | |
| 13 else: | |
| 14 from twisted.web import soap | |
| 15 SOAPPublisher = soap.SOAPPublisher | |
| 16 | |
| 17 from twisted.trial import unittest | |
| 18 from twisted.web import server, error | |
| 19 from twisted.internet import reactor, defer | |
| 20 | |
| 21 | |
| 22 class Test(SOAPPublisher): | |
| 23 | |
| 24 def soap_add(self, a, b): | |
| 25 return a + b | |
| 26 | |
| 27 def soap_kwargs(self, a=1, b=2): | |
| 28 return a + b | |
| 29 soap_kwargs.useKeywords=True | |
| 30 | |
| 31 def soap_triple(self, string, num): | |
| 32 return [string, num, None] | |
| 33 | |
| 34 def soap_struct(self): | |
| 35 return SOAPpy.structType({"a": "c"}) | |
| 36 | |
| 37 def soap_defer(self, x): | |
| 38 return defer.succeed(x) | |
| 39 | |
| 40 def soap_deferFail(self): | |
| 41 return defer.fail(ValueError()) | |
| 42 | |
| 43 def soap_fail(self): | |
| 44 raise RuntimeError | |
| 45 | |
| 46 def soap_deferFault(self): | |
| 47 return defer.fail(ValueError()) | |
| 48 | |
| 49 def soap_complex(self): | |
| 50 return {"a": ["b", "c", 12, []], "D": "foo"} | |
| 51 | |
| 52 def soap_dict(self, map, key): | |
| 53 return map[key] | |
| 54 | |
| 55 | |
| 56 class SOAPTestCase(unittest.TestCase): | |
| 57 | |
| 58 def setUp(self): | |
| 59 self.publisher = Test() | |
| 60 self.p = reactor.listenTCP(0, server.Site(self.publisher), | |
| 61 interface="127.0.0.1") | |
| 62 self.port = self.p.getHost().port | |
| 63 | |
| 64 def tearDown(self): | |
| 65 return self.p.stopListening() | |
| 66 | |
| 67 def proxy(self): | |
| 68 return soap.Proxy("http://127.0.0.1:%d/" % self.port) | |
| 69 | |
| 70 def testResults(self): | |
| 71 inputOutput = [ | |
| 72 ("add", (2, 3), 5), | |
| 73 ("defer", ("a",), "a"), | |
| 74 ("dict", ({"a": 1}, "a"), 1), | |
| 75 ("triple", ("a", 1), ["a", 1, None])] | |
| 76 | |
| 77 dl = [] | |
| 78 for meth, args, outp in inputOutput: | |
| 79 d = self.proxy().callRemote(meth, *args) | |
| 80 d.addCallback(self.assertEquals, outp) | |
| 81 dl.append(d) | |
| 82 | |
| 83 # SOAPpy kinda blows. | |
| 84 d = self.proxy().callRemote('complex') | |
| 85 d.addCallback(lambda result: result._asdict()) | |
| 86 d.addCallback(self.assertEquals, {"a": ["b", "c", 12, []], "D": "foo"}) | |
| 87 dl.append(d) | |
| 88 | |
| 89 # We now return to our regularly scheduled program, already in progress. | |
| 90 return defer.DeferredList(dl, fireOnOneErrback=True) | |
| 91 | |
| 92 def testMethodNotFound(self): | |
| 93 """ | |
| 94 Check that a non existing method return error 500. | |
| 95 """ | |
| 96 d = self.proxy().callRemote('doesntexist') | |
| 97 self.assertFailure(d, error.Error) | |
| 98 def cb(err): | |
| 99 self.assertEquals(int(err.status), 500) | |
| 100 d.addCallback(cb) | |
| 101 return d | |
| 102 | |
| 103 def testLookupFunction(self): | |
| 104 """ | |
| 105 Test lookupFunction method on publisher, to see available remote | |
| 106 methods. | |
| 107 """ | |
| 108 self.assertTrue(self.publisher.lookupFunction("add")) | |
| 109 self.assertTrue(self.publisher.lookupFunction("fail")) | |
| 110 self.assertFalse(self.publisher.lookupFunction("foobar")) | |
| 111 | |
| 112 if not SOAPpy: | |
| 113 SOAPTestCase.skip = "SOAPpy not installed" | |
| 114 | |
| OLD | NEW |