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 |