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
55
58
return self.written.getvalue()
61
class TestMethod(Method):
63
def invoke(self, call):
58
67
class TestPrincipal(object):
60
69
def __init__(self, creds):
79
87
QueryAPI.__init__(self, *args, **kwargs)
80
88
self.principal = None
82
def execute(self, call):
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
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)
99
106
def test_handle(self):
116
123
self.api.principal = TestPrincipal(creds)
117
124
return self.api.handle(request).addCallback(check)
126
def test_handle_with_dump_result(self):
128
L{QueryAPI.handle} serializes the action result with C{dump_result}.
130
creds = AWSCredentials("access", "secret")
131
endpoint = AWSServiceEndpoint("http://uri")
132
query = Query(action="SomeAction", creds=creds, endpoint=endpoint)
134
request = FakeRequest(query.params, endpoint)
137
self.assertEqual("data", loads(request.response))
139
self.api.dump_result = dumps
140
self.api.principal = TestPrincipal(creds)
141
return self.api.handle(request).addCallback(check)
143
def test_handle_with_deprecated_actions(self):
145
L{QueryAPI.handle} supports the legacy 'actions' attribute.
147
self.api.actions = ["SomeAction"]
148
creds = AWSCredentials("access", "secret")
149
endpoint = AWSServiceEndpoint("http://uri")
150
query = Query(action="SomeAction", creds=creds, endpoint=endpoint)
152
request = FakeRequest(query.params, endpoint)
155
self.assertEqual("data", request.response)
157
self.api.principal = TestPrincipal(creds)
158
return self.api.handle(request).addCallback(check)
119
160
def test_handle_pass_params_to_call(self):
121
162
L{QueryAPI.handle} creates a L{Call} object with the correct
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)
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)
300
request = FakeRequest(query.params, endpoint)
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)
308
return self.api.handle(request).addCallback(check)
310
def test_handle_with_deprecated_actions_and_unsupported_action(self):
312
If the deprecated L{QueryAPI.actions} attribute is set, it will be
313
used for looking up supported actions.
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)