~tribaal/txaws/xss-hardening

« back to all changes in this revision

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

merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
from json import dumps, loads
1
2
from pytz import UTC
2
3
from cStringIO import StringIO
3
4
from datetime import datetime
7
8
from txaws.credentials import AWSCredentials
8
9
from txaws.service import AWSServiceEndpoint
9
10
from txaws.ec2.client import Query
 
11
from txaws.server.method import Method
 
12
from txaws.server.registry import Registry
10
13
from txaws.server.resource import QueryAPI
11
14
 
12
15
 
55
58
        return self.written.getvalue()
56
59
 
57
60
 
 
61
class TestMethod(Method):
 
62
 
 
63
    def invoke(self, call):
 
64
        return "data"
 
65
 
 
66
 
58
67
class TestPrincipal(object):
59
68
 
60
69
    def __init__(self, creds):
71
80
 
72
81
class TestQueryAPI(QueryAPI):
73
82
 
74
 
    actions = ["SomeAction"]
75
83
    signature_versions = (1, 2)
76
84
    content_type = "text/plain"
77
85
 
79
87
        QueryAPI.__init__(self, *args, **kwargs)
80
88
        self.principal = None
81
89
 
82
 
    def execute(self, call):
83
 
        return "data"
84
 
 
85
90
    def get_principal(self, access_key):
86
91
        if self.principal and self.principal.access_key == access_key:
87
92
            return self.principal
94
99
 
95
100
    def setUp(self):
96
101
        super(QueryAPITest, self).setUp()
97
 
        self.api = TestQueryAPI()
 
102
        self.registry = Registry()
 
103
        self.registry.add(TestMethod, action="SomeAction", version=None)
 
104
        self.api = TestQueryAPI(registry=self.registry)
98
105
 
99
106
    def test_handle(self):
100
107
        """
116
123
        self.api.principal = TestPrincipal(creds)
117
124
        return self.api.handle(request).addCallback(check)
118
125
 
 
126
    def test_handle_with_dump_result(self):
 
127
        """
 
128
        L{QueryAPI.handle} serializes the action result with C{dump_result}.
 
129
        """
 
130
        creds = AWSCredentials("access", "secret")
 
131
        endpoint = AWSServiceEndpoint("http://uri")
 
132
        query = Query(action="SomeAction", creds=creds, endpoint=endpoint)
 
133
        query.sign()
 
134
        request = FakeRequest(query.params, endpoint)
 
135
 
 
136
        def check(ignored):
 
137
            self.assertEqual("data", loads(request.response))
 
138
 
 
139
        self.api.dump_result = dumps
 
140
        self.api.principal = TestPrincipal(creds)
 
141
        return self.api.handle(request).addCallback(check)
 
142
 
 
143
    def test_handle_with_deprecated_actions(self):
 
144
        """
 
145
        L{QueryAPI.handle} supports the legacy 'actions' attribute.
 
146
        """
 
147
        self.api.actions = ["SomeAction"]
 
148
        creds = AWSCredentials("access", "secret")
 
149
        endpoint = AWSServiceEndpoint("http://uri")
 
150
        query = Query(action="SomeAction", creds=creds, endpoint=endpoint)
 
151
        query.sign()
 
152
        request = FakeRequest(query.params, endpoint)
 
153
 
 
154
        def check(ignored):
 
155
            self.assertEqual("data", request.response)
 
156
 
 
157
        self.api.principal = TestPrincipal(creds)
 
158
        return self.api.handle(request).addCallback(check)
 
159
 
119
160
    def test_handle_pass_params_to_call(self):
120
161
        """
121
162
        L{QueryAPI.handle} creates a L{Call} object with the correct
122
163
        parameters.
123
164
        """
 
165
        self.registry.add(TestMethod, "SomeAction", "1.2.3")
124
166
        creds = AWSCredentials("access", "secret")
125
167
        endpoint = AWSServiceEndpoint("http://uri")
126
168
        query = Query(action="SomeAction", creds=creds, endpoint=endpoint,
250
292
        return self.api.handle(request).addCallback(check)
251
293
 
252
294
    def test_handle_with_unsupported_action(self):
253
 
        """Only actions listed in L{QueryAPI.actions} are supported."""
 
295
        """Only actions registered in the L{Registry} are supported."""
 
296
        creds = AWSCredentials("access", "secret")
 
297
        endpoint = AWSServiceEndpoint("http://uri")
 
298
        query = Query(action="FooBar", creds=creds, endpoint=endpoint)
 
299
        query.sign()
 
300
        request = FakeRequest(query.params, endpoint)
 
301
 
 
302
        def check(ignored):
 
303
            self.flushLoggedErrors()
 
304
            self.assertEqual("InvalidAction - The action FooBar is not valid"
 
305
                             " for this web service.", request.response)
 
306
            self.assertEqual(400, request.code)
 
307
 
 
308
        return self.api.handle(request).addCallback(check)
 
309
 
 
310
    def test_handle_with_deprecated_actions_and_unsupported_action(self):
 
311
        """
 
312
        If the deprecated L{QueryAPI.actions} attribute is set, it will be
 
313
        used for looking up supported actions.
 
314
        """
 
315
        self.api.actions = ["SomeAction"]
254
316
        creds = AWSCredentials("access", "secret")
255
317
        endpoint = AWSServiceEndpoint("http://uri")
256
318
        query = Query(action="FooBar", creds=creds, endpoint=endpoint)