1
from twisted.trial.unittest import TestCase
3
from txaws.server.method import Method
4
from txaws.server.registry import Registry
5
from txaws.server.exception import APIError
7
from txaws.server.tests.fixtures import (
8
has_venusian, importerror, amodule)
9
from txaws.server.tests.fixtures.amodule import TestMethod
10
from txaws.server.tests.fixtures.importerror.amodule import (
11
TestMethod as testmethod)
14
class RegistryTest(TestCase):
17
super(RegistryTest, self).setUp()
18
self.registry = Registry()
22
L{MehtodRegistry.add} registers a method class for the given action
25
self.registry.add(TestMethod, "test", "1.0")
26
self.registry.add(TestMethod, "test", "2.0")
27
self.registry.check("test", "1.0")
28
self.registry.check("test", "2.0")
29
self.assertIdentical(TestMethod, self.registry.get("test", "1.0"))
30
self.assertIdentical(TestMethod, self.registry.get("test", "2.0"))
32
def test_add_duplicate_method(self):
34
L{MehtodRegistry.add} fails if a method class for the given action
35
and version was already registered.
38
class TestMethod2(Method):
41
self.registry.add(TestMethod, "test", "1.0")
42
self.assertRaises(RuntimeError, self.registry.add, TestMethod2,
47
L{MehtodRegistry.get} returns the method class registered for the
48
given action and version.
51
class TestMethod2(Method):
54
self.registry.add(TestMethod, "test", "1.0")
55
self.registry.add(TestMethod, "test", "2.0")
56
self.registry.add(TestMethod2, "test", "3.0")
57
self.assertIdentical(TestMethod, self.registry.get("test", "1.0"))
58
self.assertIdentical(TestMethod, self.registry.get("test", "2.0"))
59
self.assertIdentical(TestMethod2, self.registry.get("test", "3.0"))
61
def test_check_with_missing_action(self):
63
L{MehtodRegistry.get} fails if the given action is not registered.
65
error = self.assertRaises(APIError, self.registry.check, "boom", "1.0")
66
self.assertEqual(400, error.status)
67
self.assertEqual("InvalidAction", error.code)
68
self.assertEqual("The action boom is not valid for this web service.",
71
def test_check_with_missing_version(self):
73
L{MehtodRegistry.get} fails if the given action is not registered.
75
self.registry.add(TestMethod, "test", "1.0")
76
error = self.assertRaises(APIError, self.registry.check, "test", "2.0")
77
self.assertEqual(400, error.status)
78
self.assertEqual("InvalidVersion", error.code)
79
self.assertEqual("Invalid API version.", error.message)
83
L{MehtodRegistry.scan} registers the L{Method}s decorated with L{api}.
85
self.registry.scan(amodule)
86
self.assertIdentical(TestMethod, self.registry.get("TestMethod", None))
88
def test_scan_raises_error_on_importerror(self):
90
L{MethodRegistry.scan} raises an error by default when an error happens
91
and there is no onerror callback is passed.
93
self.assertRaises(ImportError, self.registry.scan, importerror)
95
def test_scan_swallows_with_onerror(self):
97
L{MethodRegistry.scan} accepts an onerror callback that can be used to
98
deal with scanning errors.
102
swallowed.append(error)
104
self.registry.scan(importerror, onerror=swallow)
105
self.assertEqual(1, len(swallowed))
106
self.assertEqual(testmethod, self.registry.get("TestMethod"))
109
test_scan.skip = "venusian module not available"
110
test_scan_raises_error_on_importerror.skip = "venusian module not "
112
test_scan_swallows_with_onerror.skip = "venusian module not available"