1
from datetime import datetime
3
from pytz import UTC, FixedOffset
5
from twisted.trial.unittest import TestCase
7
from txaws.server.exception import APIError
8
from txaws.server.schema import (
9
Arguments, Bool, Date, Enum, Integer, Parameter, RawStr, Schema, Unicode)
12
class ArgumentsTest(TestCase):
14
def test_instantiate_empty(self):
15
"""Creating an L{Arguments} object."""
16
arguments = Arguments({})
17
self.assertEqual({}, arguments.__dict__)
19
def test_instantiate_non_empty(self):
20
"""Creating an L{Arguments} object with some arguments."""
21
arguments = Arguments({"foo": 123, "bar": 456})
22
self.assertEqual(123, arguments.foo)
23
self.assertEqual(456, arguments.bar)
25
def test_iterate(self):
26
"""L{Arguments} returns an iterator with both keys and values."""
27
arguments = Arguments({"foo": 123, "bar": 456})
28
self.assertEqual([("foo", 123), ("bar", 456)], list(arguments))
30
def test_getitem(self):
31
"""Values can be looked up using C{[index]} notation."""
32
arguments = Arguments({1: "a", 2: "b", "foo": "bar"})
33
self.assertEqual("b", arguments[2])
34
self.assertEqual("bar", arguments["foo"])
36
def test_getitem_error(self):
37
"""L{KeyError} is raised when the argument is not found."""
38
arguments = Arguments({})
39
self.assertRaises(KeyError, arguments.__getitem__, 1)
42
"""C{len()} can be used with an L{Arguments} instance."""
43
self.assertEqual(0, len(Arguments({})))
44
self.assertEqual(1, len(Arguments({1: 2})))
46
def test_nested_data(self):
47
"""L{Arguments} can cope fine with nested data structures."""
48
arguments = Arguments({"foo": Arguments({"bar": "egg"})})
49
self.assertEqual("egg", arguments.foo.bar)
51
def test_nested_data_with_numbers(self):
52
"""L{Arguments} can cope fine with list items."""
53
arguments = Arguments({"foo": {1: "egg"}})
54
self.assertEqual("egg", arguments.foo[0])
57
class ParameterTest(TestCase):
59
def test_coerce(self):
61
L{Parameter.coerce} coerces a request argument with a single value.
63
parameter = Parameter("Test")
64
parameter.parse = lambda value: value
65
self.assertEqual("foo", parameter.coerce("foo"))
67
def test_coerce_with_optional(self):
68
"""L{Parameter.coerce} returns C{None} if the parameter is optional."""
69
parameter = Parameter("Test", optional=True)
70
self.assertEqual(None, parameter.coerce(None))
72
def test_coerce_with_required(self):
74
L{Parameter.coerce} raises an L{APIError} if the parameter is
75
required but not present in the request.
77
parameter = Parameter("Test")
78
error = self.assertRaises(APIError, parameter.coerce, None)
79
self.assertEqual(400, error.status)
80
self.assertEqual("MissingParameter", error.code)
81
self.assertEqual("The request must contain the parameter Test",
84
def test_coerce_with_default(self):
86
L{Parameter.coerce} returns F{Parameter.default} if the parameter is
87
optional and not present in the request.
89
parameter = Parameter("Test", optional=True, default=123)
90
self.assertEqual(123, parameter.coerce(None))
92
def test_coerce_with_parameter_error(self):
94
L{Parameter.coerce} raises an L{APIError} if an invalid value is
95
passed as request argument.
97
parameter = Parameter("Test")
98
parameter.parse = lambda value: int(value)
99
parameter.kind = "integer"
100
error = self.assertRaises(APIError, parameter.coerce, "foo")
101
self.assertEqual(400, error.status)
102
self.assertEqual("InvalidParameterValue", error.code)
103
self.assertEqual("Invalid integer value foo", error.message)
105
def test_coerce_with_empty_strings(self):
107
L{Parameter.coerce} returns C{None} if the value is an empty string and
108
C{allow_none} is C{True}.
110
parameter = Parameter("Test", allow_none=True)
111
self.assertEqual(None, parameter.coerce(""))
113
def test_coerce_with_empty_strings_error(self):
115
L{Parameter.coerce} raises an error if the value is an empty string and
116
C{allow_none} is not C{True}.
118
parameter = Parameter("Test")
119
error = self.assertRaises(APIError, parameter.coerce, "")
120
self.assertEqual(400, error.status)
121
self.assertEqual("MissingParameter", error.code)
122
self.assertEqual("The request must contain the parameter Test",
125
def test_coerce_with_min(self):
127
L{Parameter.coerce} raises an error if the given value is lower than
130
parameter = Parameter("Test", min=50)
131
parameter.measure = lambda value: int(value)
132
parameter.lower_than_min_template = "Please give me at least %s"
133
error = self.assertRaises(APIError, parameter.coerce, "4")
134
self.assertEqual(400, error.status)
135
self.assertEqual("InvalidParameterValue", error.code)
136
self.assertEqual("Value (4) for parameter Test is invalid. "
137
"Please give me at least 50", error.message)
139
def test_coerce_with_max(self):
141
L{Parameter.coerce} raises an error if the given value is greater than
144
parameter = Parameter("Test", max=3)
145
parameter.measure = lambda value: len(value)
146
parameter.greater_than_max_template = "%s should be enough for anybody"
147
error = self.assertRaises(APIError, parameter.coerce, "longish")
148
self.assertEqual(400, error.status)
149
self.assertEqual("InvalidParameterValue", error.code)
150
self.assertEqual("Value (longish) for parameter Test is invalid. "
151
"3 should be enough for anybody", error.message)
154
class UnicodeTest(TestCase):
156
def test_parse(self):
157
"""L{Unicode.parse} converts the given raw C{value} to C{unicode}."""
158
parameter = Unicode("Test")
159
self.assertEqual(u"foo", parameter.parse("foo"))
161
def test_format(self):
162
"""L{Unicode.format} encodes the given C{unicode} with utf-8."""
163
parameter = Unicode("Test")
164
value = parameter.format(u"fo\N{TAGBANWA LETTER SA}")
165
self.assertEqual("fo\xe1\x9d\xb0", value)
166
self.assertTrue(isinstance(value, str))
168
def test_min_and_max(self):
169
"""The L{Unicode} parameter properly supports ranges."""
170
parameter = Unicode("Test", min=2, max=4)
172
error = self.assertRaises(APIError, parameter.coerce, "a")
173
self.assertEqual(400, error.status)
174
self.assertEqual("InvalidParameterValue", error.code)
175
self.assertIn("Length must be at least 2.", error.message)
177
error = self.assertRaises(APIError, parameter.coerce, "abcde")
178
self.assertIn("Length exceeds maximum of 4.", error.message)
179
self.assertEqual(400, error.status)
180
self.assertEqual("InvalidParameterValue", error.code)
183
class RawStrTest(TestCase):
185
def test_parse(self):
186
"""L{RawStr.parse} checks that the given raw C{value} is a string."""
187
parameter = RawStr("Test")
188
self.assertEqual("foo", parameter.parse("foo"))
190
def test_format(self):
191
"""L{RawStr.format} simply returns the given string."""
192
parameter = RawStr("Test")
193
value = parameter.format("foo")
194
self.assertEqual("foo", value)
195
self.assertTrue(isinstance(value, str))
198
class IntegerTest(TestCase):
200
def test_parse(self):
201
"""L{Integer.parse} converts the given raw C{value} to C{int}."""
202
parameter = Integer("Test")
203
self.assertEqual(123, parameter.parse("123"))
205
def test_parse_with_negative(self):
206
"""L{Integer.parse} converts the given raw C{value} to C{int}."""
207
parameter = Integer("Test")
208
self.assertRaises(ValueError, parameter.parse, "-1")
210
def test_format(self):
211
"""L{Integer.format} converts the given integer to a string."""
212
parameter = Integer("Test")
213
self.assertEqual("123", parameter.format(123))
216
class BoolTest(TestCase):
218
def test_parse(self):
219
"""L{Bool.parse} converts 'true' to C{True}."""
220
parameter = Bool("Test")
221
self.assertEqual(True, parameter.parse("true"))
223
def test_parse_with_false(self):
224
"""L{Bool.parse} converts 'false' to C{False}."""
225
parameter = Bool("Test")
226
self.assertEqual(False, parameter.parse("false"))
228
def test_parse_with_error(self):
230
L{Bool.parse} raises C{ValueError} if the given value is neither 'true'
233
parameter = Bool("Test")
234
self.assertRaises(ValueError, parameter.parse, "0")
236
def test_format(self):
237
"""L{Bool.format} converts the given boolean to either '0' or '1'."""
238
parameter = Bool("Test")
239
self.assertEqual("true", parameter.format(True))
240
self.assertEqual("false", parameter.format(False))
243
class EnumTest(TestCase):
245
def test_parse(self):
246
"""L{Enum.parse} accepts a map for translating values."""
247
parameter = Enum("Test", {"foo": "bar"})
248
self.assertEqual("bar", parameter.parse("foo"))
250
def test_parse_with_error(self):
252
L{Bool.parse} raises C{ValueError} if the given value is not
253
present in the mapping.
255
parameter = Enum("Test", {})
256
self.assertRaises(ValueError, parameter.parse, "bar")
258
def test_format(self):
259
"""L{Enum.format} converts back the given value to the original map."""
260
parameter = Enum("Test", {"foo": "bar"})
261
self.assertEqual("foo", parameter.format("bar"))
264
class DateTest(TestCase):
266
def test_parse(self):
267
"""L{Date.parse checks that the given raw C{value} is a date/time."""
268
parameter = Date("Test")
269
date = datetime(2010, 9, 15, 23, 59, 59, tzinfo=UTC)
270
self.assertEqual(date, parameter.parse("2010-09-15T23:59:59Z"))
272
def test_format(self):
274
L{Date.format} returns a string representation of the given datetime
277
parameter = Date("Test")
278
date = datetime(2010, 9, 15, 23, 59, 59,
279
tzinfo=FixedOffset(120))
280
self.assertEqual("2010-09-15T21:59:59Z", parameter.format(date))
283
class SchemaTest(TestCase):
285
def test_extract(self):
287
L{Schema.extract} returns an L{Argument} object whose attributes are
288
the arguments extracted from the given C{request}, as specified.
290
schema = Schema(Unicode("name"))
291
arguments, _ = schema.extract({"name": "value"})
292
self.assertEqual("value", arguments.name)
294
def test_extract_with_rest(self):
296
L{Schema.extract} stores unknown parameters in the 'rest' return
300
_, rest = schema.extract({"name": "value"})
301
self.assertEqual(rest, {"name": "value"})
303
def test_extract_with_many_arguments(self):
304
"""L{Schema.extract} can handle multiple parameters."""
305
schema = Schema(Unicode("name"), Integer("count"))
306
arguments, _ = schema.extract({"name": "value", "count": "123"})
307
self.assertEqual(u"value", arguments.name)
308
self.assertEqual(123, arguments.count)
310
def test_extract_with_optional(self):
311
"""L{Schema.extract} can handle optional parameters."""
312
schema = Schema(Unicode("name"), Integer("count", optional=True))
313
arguments, _ = schema.extract({"name": "value"})
314
self.assertEqual(u"value", arguments.name)
315
self.assertEqual(None, arguments.count)
317
def test_extract_with_numbered(self):
319
L{Schema.extract} can handle parameters with numbered values.
321
schema = Schema(Unicode("name.n"))
322
arguments, _ = schema.extract({"name.0": "Joe", "name.1": "Tom"})
323
self.assertEqual("Joe", arguments.name[0])
324
self.assertEqual("Tom", arguments.name[1])
326
def test_extract_with_single_numbered(self):
328
L{Schema.extract} can handle a single parameter with a numbered value.
330
schema = Schema(Unicode("name.n"))
331
arguments, _ = schema.extract({"name.0": "Joe"})
332
self.assertEqual("Joe", arguments.name[0])
334
def test_extract_complex(self):
335
"""L{Schema} can cope with complex schemas."""
337
Unicode("GroupName"),
338
RawStr("IpPermissions.n.IpProtocol"),
339
Integer("IpPermissions.n.FromPort"),
340
Integer("IpPermissions.n.ToPort"),
341
Unicode("IpPermissions.n.Groups.m.UserId", optional=True),
342
Unicode("IpPermissions.n.Groups.m.GroupName", optional=True))
344
arguments, _ = schema.extract(
346
"IpPermissions.1.IpProtocol": "tcp",
347
"IpPermissions.1.FromPort": "1234",
348
"IpPermissions.1.ToPort": "5678",
349
"IpPermissions.1.Groups.1.GroupName": "Bar",
350
"IpPermissions.1.Groups.2.GroupName": "Egg"})
352
self.assertEqual(u"Foo", arguments.GroupName)
353
self.assertEqual(1, len(arguments.IpPermissions))
354
self.assertEqual(1234, arguments.IpPermissions[0].FromPort)
355
self.assertEqual(5678, arguments.IpPermissions[0].ToPort)
356
self.assertEqual(2, len(arguments.IpPermissions[0].Groups))
357
self.assertEqual("Bar", arguments.IpPermissions[0].Groups[0].GroupName)
358
self.assertEqual("Egg", arguments.IpPermissions[0].Groups[1].GroupName)
360
def test_extract_with_multiple_parameters_in_singular_schema(self):
362
If multiple parameters are passed in to a Schema element that is not
363
flagged as supporting multiple values then we should throw an
366
schema = Schema(Unicode("name"))
367
params = {"name.1": "value", "name.2": "value2"}
368
error = self.assertRaises(APIError, schema.extract, params)
369
self.assertEqual(400, error.status)
370
self.assertEqual("InvalidParameterCombination", error.code)
371
self.assertEqual("The parameter 'name' may only be specified once.",
374
def test_extract_with_mixed(self):
376
L{Schema.extract} stores in the rest result all numbered parameters
377
given without an index.
379
schema = Schema(Unicode("name.n"))
380
_, rest = schema.extract({"name": "foo", "name.1": "bar"})
381
self.assertEqual(rest, {"name": "foo"})
383
def test_extract_with_non_numbered_template(self):
385
L{Schema.extract} accepts a single numbered argument even if the
386
associated template is not numbered.
388
schema = Schema(Unicode("name"))
389
arguments, _ = schema.extract({"name.1": "foo"})
390
self.assertEqual("foo", arguments.name)
392
def test_extract_with_non_integer_index(self):
394
L{Schema.extract} raises an error when trying to pass a numbered
395
parameter with a non-integer index.
397
schema = Schema(Unicode("name.n"))
398
params = {"name.one": "foo"}
399
error = self.assertRaises(APIError, schema.extract, params)
400
self.assertEqual(400, error.status)
401
self.assertEqual("UnknownParameter", error.code)
402
self.assertEqual("The parameter name.one is not recognized",
405
def test_extract_with_negative_index(self):
407
L{Schema.extract} raises an error when trying to pass a numbered
408
parameter with a negative index.
410
schema = Schema(Unicode("name.n"))
411
params = {"name.-1": "foo"}
412
error = self.assertRaises(APIError, schema.extract, params)
413
self.assertEqual(400, error.status)
414
self.assertEqual("UnknownParameter", error.code)
415
self.assertEqual("The parameter name.-1 is not recognized",
418
def test_bundle(self):
420
L{Schema.bundle} returns a dictionary of raw parameters that
421
can be used for an EC2-style query.
423
schema = Schema(Unicode("name"))
424
params = schema.bundle(name="foo")
425
self.assertEqual({"name": "foo"}, params)
427
def test_bundle_with_numbered(self):
429
L{Schema.bundle} correctly handles numbered arguments.
431
schema = Schema(Unicode("name.n"))
432
params = schema.bundle(name=["foo", "bar"])
433
self.assertEqual({"name.1": "foo", "name.2": "bar"}, params)
435
def test_bundle_with_none(self):
436
"""L{None} values are discarded in L{Schema.bundle}."""
437
schema = Schema(Unicode("name.n", optional=True))
438
params = schema.bundle(name=None)
439
self.assertEqual({}, params)
441
def test_bundle_with_empty_numbered(self):
443
L{Schema.bundle} correctly handles an empty numbered arguments list.
445
schema = Schema(Unicode("name.n"))
446
params = schema.bundle(names=[])
447
self.assertEqual({}, params)
449
def test_bundle_with_numbered_not_supplied(self):
451
L{Schema.bundle} ignores parameters that are not present.
453
schema = Schema(Unicode("name.n"))
454
params = schema.bundle()
455
self.assertEqual({}, params)
457
def test_bundle_with_multiple(self):
459
L{Schema.bundle} correctly handles multiple arguments.
461
schema = Schema(Unicode("name.n"), Integer("count"))
462
params = schema.bundle(name=["Foo", "Bar"], count=123)
463
self.assertEqual({"name.1": "Foo", "name.2": "Bar", "count": "123"},
466
def test_bundle_with_arguments(self):
467
"""L{Schema.bundle} can bundle L{Arguments} too."""
468
schema = Schema(Unicode("name.n"), Integer("count"))
469
arguments = Arguments({"name": Arguments({1: "Foo", 7: "Bar"}),
471
params = schema.bundle(arguments)
472
self.assertEqual({"name.1": "Foo", "name.7": "Bar", "count": "123"},
475
def test_bundle_with_arguments_and_extra(self):
477
L{Schema.bundle} can bundle L{Arguments} with keyword arguments too.
479
Keyword arguments take precedence.
481
schema = Schema(Unicode("name.n"), Integer("count"))
482
arguments = Arguments({"name": {1: "Foo", 7: "Bar"}, "count": 321})
483
params = schema.bundle(arguments, count=123)
484
self.assertEqual({"name.1": "Foo", "name.2": "Bar", "count": "123"},
487
def test_bundle_with_missing_parameter(self):
489
L{Schema.bundle} raises an exception one of the given parameters
490
doesn't exist in the schema.
492
schema = Schema(Integer("count"))
493
self.assertRaises(RuntimeError, schema.bundle, name="foo")