~tribaal/txaws/xss-hardening

« back to all changes in this revision

Viewing changes to txaws/server/tests/test_registry.py

merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
from twisted.trial.unittest import TestCase
 
2
 
 
3
from txaws.server.method import Method
 
4
from txaws.server.registry import Registry
 
5
from txaws.server.exception import APIError
 
6
 
 
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)
 
12
 
 
13
 
 
14
class RegistryTest(TestCase):
 
15
 
 
16
    def setUp(self):
 
17
        super(RegistryTest, self).setUp()
 
18
        self.registry = Registry()
 
19
 
 
20
    def test_add(self):
 
21
        """
 
22
        L{MehtodRegistry.add} registers a method class for the given action
 
23
        and version.
 
24
        """
 
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"))
 
31
 
 
32
    def test_add_duplicate_method(self):
 
33
        """
 
34
        L{MehtodRegistry.add} fails if a method class for the given action
 
35
        and version was already registered.
 
36
        """
 
37
 
 
38
        class TestMethod2(Method):
 
39
            pass
 
40
 
 
41
        self.registry.add(TestMethod, "test", "1.0")
 
42
        self.assertRaises(RuntimeError, self.registry.add, TestMethod2,
 
43
                          "test", "1.0")
 
44
 
 
45
    def test_get(self):
 
46
        """
 
47
        L{MehtodRegistry.get} returns the method class registered for the
 
48
        given action and version.
 
49
        """
 
50
 
 
51
        class TestMethod2(Method):
 
52
            pass
 
53
 
 
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"))
 
60
 
 
61
    def test_check_with_missing_action(self):
 
62
        """
 
63
        L{MehtodRegistry.get} fails if the given action is not registered.
 
64
        """
 
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.",
 
69
                         error.message)
 
70
 
 
71
    def test_check_with_missing_version(self):
 
72
        """
 
73
        L{MehtodRegistry.get} fails if the given action is not registered.
 
74
        """
 
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)
 
80
 
 
81
    def test_scan(self):
 
82
        """
 
83
        L{MehtodRegistry.scan} registers the L{Method}s decorated with L{api}.
 
84
        """
 
85
        self.registry.scan(amodule)
 
86
        self.assertIdentical(TestMethod, self.registry.get("TestMethod", None))
 
87
 
 
88
    def test_scan_raises_error_on_importerror(self):
 
89
        """
 
90
        L{MethodRegistry.scan} raises an error by default when an error happens
 
91
        and there is no onerror callback is passed.
 
92
        """
 
93
        self.assertRaises(ImportError, self.registry.scan, importerror)
 
94
 
 
95
    def test_scan_swallows_with_onerror(self):
 
96
        """
 
97
        L{MethodRegistry.scan} accepts an onerror callback that can be used to
 
98
        deal with scanning errors.
 
99
        """
 
100
        swallowed = []
 
101
        def swallow(error):
 
102
            swallowed.append(error)
 
103
 
 
104
        self.registry.scan(importerror, onerror=swallow)
 
105
        self.assertEqual(1, len(swallowed))
 
106
        self.assertEqual(testmethod, self.registry.get("TestMethod"))
 
107
 
 
108
    if not has_venusian:
 
109
        test_scan.skip = "venusian module not available"
 
110
        test_scan_raises_error_on_importerror.skip = "venusian module not "
 
111
        "available"
 
112
        test_scan_swallows_with_onerror.skip = "venusian module not available"