~tribaal/txaws/xss-hardening
» Revision
50.1.1
(compared to revision 154)
: txaws/server/tests/test_schema.py
« back to all changes in this revision
Viewing changes to txaws/server/tests/test_schema.py
- browse files at revision 50.1.1
- compare with another revision
- download diff
- download tarball
- view history from revision 50.1.1
- Committer: Duncan McGreggor
- Date: 2009-11-22 02:20:42 UTC
- mto: (44.3.2 484858-s3-scripts)
- mto: This revision was merged to the branch mainline in revision 52.
- Revision ID: duncan@canonical.com-20091122022042-4zi231hxni1z53xd
* Updated the LICENSE file with copyright information.
* Updated the README with license information.
* Updated the README with license information.
- files removed:
- .testr.conf
- admin
- txaws/client/discover
- txaws/client/discover/tests
- txaws/client/tests
- txaws/server
- txaws/server/tests
- txaws/server/tests/fixtures
- txaws/server/tests/fixtures/importerror
- txaws/server/tests/fixtures/importerror/submodule
- wsdl
- files renamed:
- txaws/s3/ => txaws/storage/
- files modified:
- .bzrignore
added
removed
txaws/server/tests/test_schema.py
1
# -*- coding: utf-8 -*-
2
3
from datetime import datetime
4
5
from dateutil.tz import tzutc, tzoffset
6
7
from twisted.trial.unittest import TestCase
8
9
from txaws.server.exception import APIError
10
from txaws.server.schema import (
11
Arguments, Bool, Date, Enum, Integer, Parameter, RawStr, Schema, Unicode,
12
UnicodeLine, List, Structure, InconsistentParameterError)
13
14
15
class ArgumentsTestCase(TestCase):
16
17
def test_instantiate_empty(self):
18
"""Creating an L{Arguments} object."""
19
arguments = Arguments({})
20
self.assertEqual({}, arguments.__dict__)
21
22
def test_instantiate_non_empty(self):
23
"""Creating an L{Arguments} object with some arguments."""
24
arguments = Arguments({"foo": 123, "bar": 456})
25
self.assertEqual(123, arguments.foo)
26
self.assertEqual(456, arguments.bar)
27
28
def test_iterate(self):
29
"""L{Arguments} returns an iterator with both keys and values."""
30
arguments = Arguments({"foo": 123, "bar": 456})
31
self.assertEqual([("foo", 123), ("bar", 456)], list(arguments))
32
33
def test_getitem(self):
34
"""Values can be looked up using C{[index]} notation."""
35
arguments = Arguments({1: "a", 2: "b", "foo": "bar"})
36
self.assertEqual("b", arguments[2])
37
self.assertEqual("bar", arguments["foo"])
38
39
def test_getitem_error(self):
40
"""L{KeyError} is raised when the argument is not found."""
41
arguments = Arguments({})
42
self.assertRaises(KeyError, arguments.__getitem__, 1)
43
44
def test_contains(self):
45
"""
46
The presence of a certain argument can be inspected using the 'in'
47
keyword.
48
."""
49
arguments = Arguments({"foo": 1})
50
self.assertIn("foo", arguments)
51
self.assertNotIn("bar", arguments)
52
53
def test_len(self):
54
"""C{len()} can be used with an L{Arguments} instance."""
55
self.assertEqual(0, len(Arguments({})))
56
self.assertEqual(1, len(Arguments({1: 2})))
57
58
def test_nested_data(self):
59
"""L{Arguments} can cope fine with nested data structures."""
60
arguments = Arguments({"foo": Arguments({"bar": "egg"})})
61
self.assertEqual("egg", arguments.foo.bar)
62
63
def test_nested_data_with_numbers(self):
64
"""L{Arguments} can cope fine with list items."""
65
arguments = Arguments({"foo": {1: "egg"}})
66
self.assertEqual("egg", arguments.foo[0])
67
68
69
class ParameterTestCase(TestCase):
70
71
def test_coerce(self):
72
"""
73
L{Parameter.coerce} coerces a request argument with a single value.
74
"""
75
parameter = Parameter("Test")
76
parameter.parse = lambda value: value
77
self.assertEqual("foo", parameter.coerce("foo"))
78
79
def test_coerce_with_optional(self):
80
"""L{Parameter.coerce} returns C{None} if the parameter is optional."""
81
parameter = Parameter("Test", optional=True)
82
self.assertEqual(None, parameter.coerce(None))
83
84
def test_coerce_with_required(self):
85
"""
86
L{Parameter.coerce} raises an L{APIError} if the parameter is
87
required but not present in the request.
88
"""
89
parameter = Parameter("Test")
90
parameter.kind = "testy kind"
91
error = self.assertRaises(APIError, parameter.coerce, None)
92
self.assertEqual(400, error.status)
93
self.assertEqual("MissingParameter", error.code)
94
self.assertEqual("The request must contain the parameter Test "
95
"(testy kind)",
96
error.message)
97
98
def test_coerce_with_default(self):
99
"""
100
L{Parameter.coerce} returns F{Parameter.default} if the parameter is
101
optional and not present in the request.
102
"""
103
parameter = Parameter("Test", optional=True, default=123)
104
self.assertEqual(123, parameter.coerce(None))
105
106
def test_coerce_with_parameter_error(self):
107
"""
108
L{Parameter.coerce} raises an L{APIError} if an invalid value is
109
passed as request argument.
110
"""
111
parameter = Parameter("Test")
112
parameter.parse = lambda value: int(value)
113
parameter.kind = "integer"
114
error = self.assertRaises(APIError, parameter.coerce, "foo")
115
self.assertEqual(400, error.status)
116
self.assertEqual("InvalidParameterValue", error.code)
117
self.assertEqual("Invalid integer value foo", error.message)
118
119
def test_coerce_with_parameter_error_unicode(self):
120
"""
121
L{Parameter.coerce} raises an L{APIError} if an invalid value is
122
passed as request argument and parameter value is unicode.
123
"""
124
parameter = Parameter("Test")
125
parameter.parse = lambda value: int(value)
126
parameter.kind = "integer"
127
error = self.assertRaises(APIError, parameter.coerce, "citt\xc3\xa1")
128
self.assertEqual(400, error.status)
129
self.assertEqual("InvalidParameterValue", error.code)
130
self.assertEqual(u"Invalid integer value cittá", error.message)
131
132
def test_coerce_with_empty_strings(self):
133
"""
134
L{Parameter.coerce} returns C{None} if the value is an empty string and
135
C{allow_none} is C{True}.
136
"""
137
parameter = Parameter("Test", allow_none=True)
138
self.assertEqual(None, parameter.coerce(""))
139
140
def test_coerce_with_empty_strings_error(self):
141
"""
142
L{Parameter.coerce} raises an error if the value is an empty string and
143
C{allow_none} is not C{True}.
144
"""
145
parameter = Parameter("Test")
146
error = self.assertRaises(APIError, parameter.coerce, "")
147
self.assertEqual(400, error.status)
148
self.assertEqual("MissingParameter", error.code)
149
self.assertEqual("The request must contain the parameter Test",
150
error.message)
151
152
def test_coerce_with_min(self):
153
"""
154
L{Parameter.coerce} raises an error if the given value is lower than
155
the lower bound.
156
"""
157
parameter = Parameter("Test", min=50)
158
parameter.measure = lambda value: int(value)
159
parameter.lower_than_min_template = "Please give me at least %s"
160
error = self.assertRaises(APIError, parameter.coerce, "4")
161
self.assertEqual(400, error.status)
162
self.assertEqual("InvalidParameterValue", error.code)
163
self.assertEqual("Value (4) for parameter Test is invalid. "
164
"Please give me at least 50", error.message)
165
166
def test_coerce_with_max(self):
167
"""
168
L{Parameter.coerce} raises an error if the given value is greater than
169
the upper bound.
170
"""
171
parameter = Parameter("Test", max=3)
172
parameter.measure = lambda value: len(value)
173
parameter.greater_than_max_template = "%s should be enough for anybody"
174
error = self.assertRaises(APIError, parameter.coerce, "longish")
175
self.assertEqual(400, error.status)
176
self.assertEqual("InvalidParameterValue", error.code)
177
self.assertEqual("Value (longish) for parameter Test is invalid. "
178
"3 should be enough for anybody", error.message)
179
180
def test_validator_invalid(self):
181
"""
182
L{Parameter.coerce} raises an error if the validator returns False.
183
"""
184
parameter = Parameter("Test", validator=lambda _: False)
185
parameter.parse = lambda value: value
186
parameter.kind = "test_parameter"
187
error = self.assertRaises(APIError, parameter.coerce, "foo")
188
self.assertEqual(400, error.status)
189
self.assertEqual("InvalidParameterValue", error.code)
190
self.assertEqual("Invalid test_parameter value foo", error.message)
191
192
def test_validator_valid(self):
193
"""
194
L{Parameter.coerce} returns the correct value if validator returns
195
True.
196
"""
197
parameter = Parameter("Test", validator=lambda _: True)
198
parameter.parse = lambda value: value
199
parameter.kind = "test_parameter"
200
self.assertEqual("foo", parameter.coerce("foo"))
201
202
def test_parameter_doc(self):
203
"""
204
All L{Parameter} subclasses accept a 'doc' keyword argument.
205
"""
206
parameters = [
207
Unicode(doc="foo"),
208
RawStr(doc="foo"),
209
Integer(doc="foo"),
210
Bool(doc="foo"),
211
Enum(mapping={"hey": 1}, doc="foo"),
212
Date(doc="foo"),
213
List(item=Integer(), doc="foo"),
214
Structure(fields={}, doc="foo")]
215
for parameter in parameters:
216
self.assertEqual("foo", parameter.doc)
217
218
219
class UnicodeTestCase(TestCase):
220
221
def test_parse(self):
222
"""L{Unicode.parse} converts the given raw C{value} to C{unicode}."""
223
parameter = Unicode("Test")
224
self.assertEqual(u"foo", parameter.parse("foo"))
225
226
def test_parse_unicode(self):
227
"""L{Unicode.parse} works with unicode input."""
228
parameter = Unicode("Test")
229
self.assertEqual(u"cittá", parameter.parse("citt\xc3\xa1"))
230
231
def test_format(self):
232
"""L{Unicode.format} encodes the given C{unicode} with utf-8."""
233
parameter = Unicode("Test")
234
value = parameter.format(u"fo\N{TAGBANWA LETTER SA}")
235
self.assertEqual("fo\xe1\x9d\xb0", value)
236
self.assertTrue(isinstance(value, str))
237
238
def test_min_and_max(self):
239
"""The L{Unicode} parameter properly supports ranges."""
240
parameter = Unicode("Test", min=2, max=4)
241
242
error = self.assertRaises(APIError, parameter.coerce, "a")
243
self.assertEqual(400, error.status)
244
self.assertEqual("InvalidParameterValue", error.code)
245
self.assertIn("Length must be at least 2.", error.message)
246
247
error = self.assertRaises(APIError, parameter.coerce, "abcde")
248
self.assertIn("Length exceeds maximum of 4.", error.message)
249
self.assertEqual(400, error.status)
250
self.assertEqual("InvalidParameterValue", error.code)
251
252
def test_invalid_unicode(self):
253
"""
254
The L{Unicode} parameter returns an error with invalid unicode data.
255
"""
256
parameter = Unicode("Test")
257
error = self.assertRaises(APIError, parameter.coerce, "Test\x95Error")
258
self.assertIn(u"Invalid unicode value", error.message)
259
self.assertEqual(400, error.status)
260
self.assertEqual("InvalidParameterValue", error.code)
261
262
263
class UnicodeLineTestCase(TestCase):
264
265
def test_parse(self):
266
"""L{UnicodeLine.parse} converts the given raw C{value} to
267
C{unicode}."""
268
parameter = UnicodeLine("Test")
269
self.assertEqual(u"foo", parameter.parse("foo"))
270
271
def test_newlines_in_text(self):
272
"""
273
The L{UnicodeLine} parameter returns an error if text contains
274
newlines.
275
"""
276
parameter = UnicodeLine("Test")
277
error = self.assertRaises(APIError, parameter.coerce, "Test\nError")
278
self.assertIn(u"Can't contain newlines", error.message)
279
self.assertEqual(400, error.status)
280
281
282
class RawStrTestCase(TestCase):
283
284
def test_parse(self):
285
"""L{RawStr.parse} checks that the given raw C{value} is a string."""
286
parameter = RawStr("Test")
287
self.assertEqual("foo", parameter.parse("foo"))
288
289
def test_format(self):
290
"""L{RawStr.format} simply returns the given string."""
291
parameter = RawStr("Test")
292
value = parameter.format("foo")
293
self.assertEqual("foo", value)
294
self.assertTrue(isinstance(value, str))
295
296
297
class IntegerTestCase(TestCase):
298
299
def test_parse(self):
300
"""L{Integer.parse} converts the given raw C{value} to C{int}."""
301
parameter = Integer("Test")
302
self.assertEqual(123, parameter.parse("123"))
303
304
def test_parse_with_negative(self):
305
"""L{Integer.parse} converts the given raw C{value} to C{int}."""
306
parameter = Integer("Test")
307
error = self.assertRaises(APIError, parameter.coerce, "-1")
308
self.assertEqual(400, error.status)
309
self.assertEqual("InvalidParameterValue", error.code)
310
self.assertIn("Value must be at least 0.", error.message)
311
312
def test_format(self):
313
"""L{Integer.format} converts the given integer to a string."""
314
parameter = Integer("Test")
315
self.assertEqual("123", parameter.format(123))
316
317
def test_min_and_max(self):
318
"""The L{Integer} parameter properly supports ranges."""
319
parameter = Integer("Test", min=2, max=4)
320
321
error = self.assertRaises(APIError, parameter.coerce, "1")
322
self.assertEqual(400, error.status)
323
self.assertEqual("InvalidParameterValue", error.code)
324
self.assertIn("Value must be at least 2.", error.message)
325
326
error = self.assertRaises(APIError, parameter.coerce, "5")
327
self.assertIn("Value exceeds maximum of 4.", error.message)
328
self.assertEqual(400, error.status)
329
self.assertEqual("InvalidParameterValue", error.code)
330
331
def test_non_integer_string(self):
332
"""
333
The L{Integer} parameter raises an L{APIError} when passed non-int
334
values (in this case, a string).
335
"""
336
garbage = "blah"
337
parameter = Integer("Test")
338
error = self.assertRaises(APIError, parameter.coerce, garbage)
339
self.assertEqual(400, error.status)
340
self.assertEqual("InvalidParameterValue", error.code)
341
self.assertIn("Invalid integer value %s" % garbage, error.message)
342
343
344
class BoolTestCase(TestCase):
345
346
def test_parse(self):
347
"""L{Bool.parse} converts 'true' to C{True}."""
348
parameter = Bool("Test")
349
self.assertEqual(True, parameter.parse("true"))
350
351
def test_parse_with_false(self):
352
"""L{Bool.parse} converts 'false' to C{False}."""
353
parameter = Bool("Test")
354
self.assertEqual(False, parameter.parse("false"))
355
356
def test_parse_with_error(self):
357
"""
358
L{Bool.parse} raises C{ValueError} if the given value is neither 'true'
359
or 'false'.
360
"""
361
parameter = Bool("Test")
362
self.assertRaises(ValueError, parameter.parse, "0")
363
364
def test_format(self):
365
"""L{Bool.format} converts the given boolean to either '0' or '1'."""
366
parameter = Bool("Test")
367
self.assertEqual("true", parameter.format(True))
368
self.assertEqual("false", parameter.format(False))
369
370
371
class EnumTestCase(TestCase):
372
373
def test_parse(self):
374
"""L{Enum.parse} accepts a map for translating values."""
375
parameter = Enum("Test", {"foo": "bar"})
376
self.assertEqual("bar", parameter.parse("foo"))
377
378
def test_parse_with_error(self):
379
"""
380
L{Bool.parse} raises C{ValueError} if the given value is not
381
present in the mapping.
382
"""
383
parameter = Enum("Test", {})
384
self.assertRaises(ValueError, parameter.parse, "bar")
385
386
def test_format(self):
387
"""L{Enum.format} converts back the given value to the original map."""
388
parameter = Enum("Test", {"foo": "bar"})
389
self.assertEqual("foo", parameter.format("bar"))
390
391
392
class DateTestCase(TestCase):
393
394
def test_parse(self):
395
"""L{Date.parse checks that the given raw C{value} is a date/time."""
396
parameter = Date("Test")
397
date = datetime(2010, 9, 15, 23, 59, 59, tzinfo=tzutc())
398
self.assertEqual(date, parameter.parse("2010-09-15T23:59:59Z"))
399
400
def test_format(self):
401
"""
402
L{Date.format} returns a string representation of the given datetime
403
instance.
404
"""
405
parameter = Date("Test")
406
date = datetime(2010, 9, 15, 23, 59, 59,
407
tzinfo=tzoffset('UTC', 120 * 60))
408
self.assertEqual("2010-09-15T21:59:59Z", parameter.format(date))
409
410
411
class SchemaTestCase(TestCase):
412
413
def test_get_parameters(self):
414
"""
415
L{Schema.get_parameters} returns the original list of parameters.
416
"""
417
schema = Schema(parameters=[
418
Unicode("name"),
419
List("scores", Integer())])
420
parameters = schema.get_parameters()
421
self.assertEqual("name", parameters[0].name)
422
self.assertEqual("scores", parameters[1].name)
423
424
def test_get_parameters_order_on_parameter_only_construction(self):
425
"""
426
L{Schema.get_parameters} returns the original list of L{Parameter}s
427
even when they are passed as positional arguments to L{Schema}.
428
"""
429
schema = Schema(
430
Unicode("name"),
431
List("scores", Integer()),
432
Integer("index", Integer()))
433
self.assertEqual(["name", "scores", "index"],
434
[p.name for p in schema.get_parameters()])
435
436
def test_extract(self):
437
"""
438
L{Schema.extract} returns an L{Argument} object whose attributes are
439
the arguments extracted from the given C{request}, as specified.
440
"""
441
schema = Schema(Unicode("name"))
442
arguments, _ = schema.extract({"name": "value"})
443
self.assertEqual("value", arguments.name)
444
445
def test_extract_with_rest(self):
446
"""
447
L{Schema.extract} stores unknown parameters in the 'rest' return
448
dictionary.
449
"""
450
schema = Schema()
451
_, rest = schema.extract({"name": "value"})
452
self.assertEqual(rest, {"name": "value"})
453
454
def test_extract_with_nested_rest(self):
455
schema = Schema()
456
_, rest = schema.extract({"foo.1.bar": "hey", "foo.2.baz": "there"})
457
self.assertEqual({"foo.1.bar": "hey", "foo.2.baz": "there"}, rest)
458
459
def test_extract_with_many_arguments(self):
460
"""L{Schema.extract} can handle multiple parameters."""
461
schema = Schema(Unicode("name"), Integer("count"))
462
arguments, _ = schema.extract({"name": "value", "count": "123"})
463
self.assertEqual(u"value", arguments.name)
464
self.assertEqual(123, arguments.count)
465
466
def test_extract_with_optional(self):
467
"""L{Schema.extract} can handle optional parameters."""
468
schema = Schema(Unicode("name"), Integer("count", optional=True))
469
arguments, _ = schema.extract({"name": "value"})
470
self.assertEqual(u"value", arguments.name)
471
self.assertEqual(None, arguments.count)
472
473
def test_extract_with_optional_default(self):
474
"""
475
The value of C{default} on a parameter is used as the value when it is
476
not provided as an argument and the parameter is C{optional}.
477
"""
478
schema = Schema(Unicode("name"),
479
Integer("count", optional=True, default=5))
480
arguments, _ = schema.extract({"name": "value"})
481
self.assertEqual(u"value", arguments.name)
482
self.assertEqual(5, arguments.count)
483
484
def test_extract_structure_with_optional(self):
485
"""L{Schema.extract} can handle optional parameters."""
486
schema = Schema(
487
Structure(
488
"struct",
489
fields={"name": Unicode(optional=True, default="radix")}))
490
arguments, _ = schema.extract({"struct": {}})
491
self.assertEqual(u"radix", arguments.struct.name)
492
493
def test_extract_with_numbered(self):
494
"""
495
L{Schema.extract} can handle parameters with numbered values.
496
"""
497
schema = Schema(Unicode("name.n"))
498
arguments, _ = schema.extract({"name.0": "Joe", "name.1": "Tom"})
499
self.assertEqual("Joe", arguments.name[0])
500
self.assertEqual("Tom", arguments.name[1])
501
502
def test_extract_with_goofy_numbered(self):
503
"""
504
L{Schema.extract} only uses the relative values of indices to determine
505
the index in the resultant list.
506
"""
507
schema = Schema(Unicode("name.n"))
508
arguments, _ = schema.extract({"name.5": "Joe", "name.10": "Tom"})
509
self.assertEqual("Joe", arguments.name[0])
510
self.assertEqual("Tom", arguments.name[1])
511
512
def test_extract_with_single_numbered(self):
513
"""
514
L{Schema.extract} can handle an un-numbered argument passed in to a
515
numbered parameter.
516
"""
517
schema = Schema(Unicode("name.n"))
518
arguments, _ = schema.extract({"name": "Joe"})
519
self.assertEqual("Joe", arguments.name[0])
520
521
def test_extract_complex(self):
522
"""L{Schema} can cope with complex schemas."""
523
schema = Schema(
524
Unicode("GroupName"),
525
RawStr("IpPermissions.n.IpProtocol"),
526
Integer("IpPermissions.n.FromPort"),
527
Integer("IpPermissions.n.ToPort"),
528
Unicode("IpPermissions.n.Groups.m.UserId", optional=True),
529
Unicode("IpPermissions.n.Groups.m.GroupName", optional=True))
530
531
arguments, _ = schema.extract(
532
{"GroupName": "Foo",
533
"IpPermissions.1.IpProtocol": "tcp",
534
"IpPermissions.1.FromPort": "1234",
535
"IpPermissions.1.ToPort": "5678",
536
"IpPermissions.1.Groups.1.GroupName": "Bar",
537
"IpPermissions.1.Groups.2.GroupName": "Egg"})
538
539
self.assertEqual(u"Foo", arguments.GroupName)
540
self.assertEqual(1, len(arguments.IpPermissions))
541
self.assertEqual(1234, arguments.IpPermissions[0].FromPort)
542
self.assertEqual(5678, arguments.IpPermissions[0].ToPort)
543
self.assertEqual(2, len(arguments.IpPermissions[0].Groups))
544
self.assertEqual("Bar", arguments.IpPermissions[0].Groups[0].GroupName)
545
self.assertEqual("Egg", arguments.IpPermissions[0].Groups[1].GroupName)
546
547
def test_extract_with_multiple_parameters_in_singular_schema(self):
548
"""
549
If multiple parameters are passed in to a Schema element that is not
550
flagged as supporting multiple values then we should throw an
551
C{APIError}.
552
"""
553
schema = Schema(Unicode("name"))
554
params = {"name.1": "value", "name.2": "value2"}
555
error = self.assertRaises(APIError, schema.extract, params)
556
self.assertEqual(400, error.status)
557
self.assertEqual("InvalidParameterCombination", error.code)
558
self.assertEqual("The parameter 'name' may only be specified once.",
559
error.message)
560
561
def test_extract_with_mixed(self):
562
"""
563
L{Schema.extract} stores in the rest result all numbered parameters
564
given without an index.
565
"""
566
schema = Schema(Unicode("name.n"))
567
self.assertRaises(
568
InconsistentParameterError,
569
schema.extract, {"nameFOOO": "foo", "nameFOOO.1": "bar"})
570
571
def test_extract_with_non_numbered_template(self):
572
"""
573
L{Schema.extract} accepts a single numbered argument even if the
574
associated template is not numbered.
575
"""
576
schema = Schema(Unicode("name"))
577
arguments, _ = schema.extract({"name.1": "foo"})
578
self.assertEqual("foo", arguments.name)
579
580
def test_extract_with_non_integer_index(self):
581
"""
582
L{Schema.extract} raises an error when trying to pass a numbered
583
parameter with a non-integer index.
584
"""
585
schema = Schema(Unicode("name.n"))
586
params = {"name.one": "foo"}
587
error = self.assertRaises(APIError, schema.extract, params)
588
self.assertEqual(400, error.status)
589
self.assertEqual("UnknownParameter", error.code)
590
self.assertEqual("The parameter one is not recognized",
591
error.message)
592
593
def test_extract_with_negative_index(self):
594
"""
595
L{Schema.extract} raises an error when trying to pass a numbered
596
parameter with a negative index.
597
"""
598
schema = Schema(Unicode("name.n"))
599
params = {"name.-1": "foo"}
600
error = self.assertRaises(APIError, schema.extract, params)
601
self.assertEqual(400, error.status)
602
self.assertEqual("UnknownParameter", error.code)
603
self.assertEqual("The parameter -1 is not recognized",
604
error.message)
605
606
def test_bundle(self):
607
"""
608
L{Schema.bundle} returns a dictionary of raw parameters that
609
can be used for an EC2-style query.
610
"""
611
schema = Schema(Unicode("name"))
612
params = schema.bundle(name="foo")
613
self.assertEqual({"name": "foo"}, params)
614
615
def test_bundle_with_numbered(self):
616
"""
617
L{Schema.bundle} correctly handles numbered arguments.
618
"""
619
schema = Schema(Unicode("name.n"))
620
params = schema.bundle(name=["foo", "bar"])
621
self.assertEqual({"name.1": "foo", "name.2": "bar"}, params)
622
623
def test_bundle_with_two_numbered(self):
624
"""
625
L{Schema.bundle} can bundle multiple numbered lists.
626
"""
627
schema = Schema(Unicode("names.n"), Unicode("things.n"))
628
params = schema.bundle(names=["foo", "bar"], things=["baz", "quux"])
629
self.assertEqual({"names.1": "foo", "names.2": "bar",
630
"things.1": "baz", "things.2": "quux"},
631
params)
632
633
def test_bundle_with_none(self):
634
"""L{None} values are discarded in L{Schema.bundle}."""
635
schema = Schema(Unicode("name.n", optional=True))
636
params = schema.bundle(name=None)
637
self.assertEqual({}, params)
638
639
def test_bundle_with_empty_numbered(self):
640
"""
641
L{Schema.bundle} correctly handles an empty numbered arguments list.
642
"""
643
schema = Schema(Unicode("name.n"))
644
params = schema.bundle(name=[])
645
self.assertEqual({}, params)
646
647
def test_bundle_with_numbered_not_supplied(self):
648
"""
649
L{Schema.bundle} ignores parameters that are not present.
650
"""
651
schema = Schema(Unicode("name.n"))
652
params = schema.bundle()
653
self.assertEqual({}, params)
654
655
def test_bundle_with_multiple(self):
656
"""
657
L{Schema.bundle} correctly handles multiple arguments.
658
"""
659
schema = Schema(Unicode("name.n"), Integer("count"))
660
params = schema.bundle(name=["Foo", "Bar"], count=123)
661
self.assertEqual({"name.1": "Foo", "name.2": "Bar", "count": "123"},
662
params)
663
664
def test_bundle_with_structure(self):
665
"""L{Schema.bundle} can bundle L{Structure}s."""
666
schema = Schema(
667
parameters=[
668
Structure("struct", fields={"field1": Unicode(),
669
"field2": Integer()})])
670
params = schema.bundle(struct={"field1": "hi", "field2": 59})
671
self.assertEqual({"struct.field1": "hi", "struct.field2": "59"},
672
params)
673
674
def test_bundle_with_list(self):
675
"""L{Schema.bundle} can bundle L{List}s."""
676
schema = Schema(parameters=[List("things", item=Unicode())])
677
params = schema.bundle(things=["foo", "bar"])
678
self.assertEqual({"things.1": "foo", "things.2": "bar"}, params)
679
680
def test_bundle_with_structure_with_arguments(self):
681
"""
682
L{Schema.bundle} can bundle L{Structure}s (specified as L{Arguments}).
683
"""
684
schema = Schema(
685
parameters=[
686
Structure("struct", fields={"field1": Unicode(),
687
"field2": Integer()})])
688
params = schema.bundle(struct=Arguments({"field1": "hi",
689
"field2": 59}))
690
self.assertEqual({"struct.field1": "hi", "struct.field2": "59"},
691
params)
692
693
def test_bundle_with_list_with_arguments(self):
694
"""L{Schema.bundle} can bundle L{List}s (specified as L{Arguments})."""
695
schema = Schema(parameters=[List("things", item=Unicode())])
696
params = schema.bundle(things=Arguments({1: "foo", 2: "bar"}))
697
self.assertEqual({"things.1": "foo", "things.2": "bar"}, params)
698
699
def test_bundle_with_arguments(self):
700
"""L{Schema.bundle} can bundle L{Arguments} too."""
701
schema = Schema(Unicode("name.n"), Integer("count"))
702
arguments = Arguments({"name": Arguments({1: "Foo", 7: "Bar"}),
703
"count": 123})
704
params = schema.bundle(arguments)
705
self.assertEqual({"name.1": "Foo", "name.7": "Bar", "count": "123"},
706
params)
707
708
def test_bundle_with_arguments_and_extra(self):
709
"""
710
L{Schema.bundle} can bundle L{Arguments} with keyword arguments too.
711
712
Keyword arguments take precedence.
713
"""
714
schema = Schema(Unicode("name.n"), Integer("count"))
715
arguments = Arguments({"name": {1: "Foo", 7: "Bar"}, "count": 321})
716
params = schema.bundle(arguments, count=123)
717
self.assertEqual({"name.1": "Foo", "name.2": "Bar", "count": "123"},
718
params)
719
720
def test_bundle_with_missing_parameter(self):
721
"""
722
L{Schema.bundle} raises an exception one of the given parameters
723
doesn't exist in the schema.
724
"""
725
schema = Schema(Integer("count"))
726
self.assertRaises(RuntimeError, schema.bundle, name="foo")
727
728
def test_add_single_extra_schema_item(self):
729
"""New Parameters can be added to the Schema."""
730
schema = Schema(Unicode("name"))
731
schema = schema.extend(Unicode("computer"))
732
arguments, _ = schema.extract({"name": "value", "computer": "testing"})
733
self.assertEqual(u"value", arguments.name)
734
self.assertEqual("testing", arguments.computer)
735
736
def test_add_extra_schema_items(self):
737
"""A list of new Parameters can be added to the Schema."""
738
schema = Schema(Unicode("name"))
739
schema = schema.extend(Unicode("computer"), Integer("count"))
740
arguments, _ = schema.extract({"name": "value", "computer": "testing",
741
"count": "5"})
742
self.assertEqual(u"value", arguments.name)
743
self.assertEqual("testing", arguments.computer)
744
self.assertEqual(5, arguments.count)
745
746
def test_list(self):
747
"""L{List}s can be extracted."""
748
schema = Schema(List("foo", Integer()))
749
arguments, _ = schema.extract({"foo.1": "1", "foo.2": "2"})
750
self.assertEqual([1, 2], arguments.foo)
751
752
def test_optional_list(self):
753
"""
754
The default value of an optional L{List} is C{[]}.
755
"""
756
schema = Schema(List("names", Unicode(), optional=True))
757
arguments, _ = schema.extract({})
758
self.assertEqual([], arguments.names)
759
760
def test_default_list(self):
761
"""
762
The default of a L{List} can be specified as a list.
763
"""
764
schema = Schema(List("names", Unicode(), optional=True,
765
default=[u"foo", u"bar"]))
766
arguments, _ = schema.extract({})
767
self.assertEqual([u"foo", u"bar"], arguments.names)
768
769
def test_list_of_list(self):
770
"""L{List}s can be nested."""
771
schema = Schema(List("foo", List(item=Unicode())))
772
arguments, _ = schema.extract(
773
{"foo.1.1": "first-first", "foo.1.2": "first-second",
774
"foo.2.1": "second-first", "foo.2.2": "second-second"})
775
self.assertEqual([["first-first", "first-second"],
776
["second-first", "second-second"]],
777
arguments.foo)
778
779
def test_structure(self):
780
"""
781
L{Schema}s with L{Structure} parameters can have arguments extracted.
782
"""
783
schema = Schema(Structure("foo", {"a": Integer(), "b": Integer()}))
784
arguments, _ = schema.extract({"foo.a": "1", "foo.b": "2"})
785
self.assertEqual(1, arguments.foo.a)
786
self.assertEqual(2, arguments.foo.b)
787
788
def test_structure_of_structures(self):
789
"""L{Structure}s can be nested."""
790
sub_struct = Structure(fields={"a": Unicode(), "b": Unicode()})
791
schema = Schema(Structure("foo", fields={"a": sub_struct,
792
"b": sub_struct}))
793
arguments, _ = schema.extract({"foo.a.a": "a-a", "foo.a.b": "a-b",
794
"foo.b.a": "b-a", "foo.b.b": "b-b"})
795
self.assertEqual("a-a", arguments.foo.a.a)
796
self.assertEqual("a-b", arguments.foo.a.b)
797
self.assertEqual("b-a", arguments.foo.b.a)
798
self.assertEqual("b-b", arguments.foo.b.b)
799
800
def test_list_of_structures(self):
801
"""L{List}s of L{Structure}s are extracted properly."""
802
schema = Schema(
803
List("foo", Structure(fields={"a": Integer(), "b": Integer()})))
804
arguments, _ = schema.extract({"foo.1.a": "1", "foo.1.b": "2",
805
"foo.2.a": "3", "foo.2.b": "4"})
806
self.assertEqual(1, arguments.foo[0]['a'])
807
self.assertEqual(2, arguments.foo[0]['b'])
808
self.assertEqual(3, arguments.foo[1]['a'])
809
self.assertEqual(4, arguments.foo[1]['b'])
810
811
def test_structure_of_list(self):
812
"""L{Structure}s of L{List}s are extracted properly."""
813
schema = Schema(Structure("foo", fields={"l": List(item=Integer())}))
814
arguments, _ = schema.extract({"foo.l.1": "1", "foo.l.2": "2"})
815
self.assertEqual([1, 2], arguments.foo.l)
816
817
def test_new_parameters(self):
818
"""
819
L{Schema} accepts a C{parameters} parameter to specify parameters in a
820
{name: field} format.
821
"""
822
schema = Schema(
823
parameters=[Structure("foo",
824
fields={"l": List(item=Integer())})])
825
arguments, _ = schema.extract({"foo.l.1": "1", "foo.l.2": "2"})
826
self.assertEqual([1, 2], arguments.foo.l)
827
828
def test_schema_conversion_list(self):
829
"""
830
Backwards-compatibility conversion maintains the name of lists.
831
"""
832
schema = Schema(Unicode("foos.N"))
833
parameters = schema.get_parameters()
834
self.assertEqual(1, len(parameters))
835
self.assertTrue(isinstance(parameters[0], List))
836
self.assertEqual("foos", parameters[0].name)
837
838
def test_coerce_list(self):
839
"""
840
When a L{List} coerces the value of one of its item, it uses the the
841
proper name in the C{MissingParameter} error raised.
842
"""
843
parameter = List("foo", Unicode())
844
error = self.assertRaises(APIError, parameter.item.coerce, "")
845
self.assertEqual(400, error.status)
846
self.assertEqual("MissingParameter", error.code)
847
self.assertEqual("The request must contain the parameter foo "
848
"(unicode)",
849
error.message)
850
851
def test_schema_conversion_structure_name(self):
852
"""
853
Backwards-compatibility conversion maintains the names of fields in
854
structures.
855
"""
856
schema = Schema(Unicode("foos.N.field"),
857
Unicode("foos.N.field2"))
858
parameters = schema.get_parameters()
859
self.assertEqual(1, len(parameters))
860
self.assertTrue(isinstance(parameters[0], List))
861
self.assertEqual("foos", parameters[0].name)
862
self.assertEqual("N",
863
parameters[0].item.name)
864
self.assertEqual("field",
865
parameters[0].item.fields["field"].name)
866
self.assertEqual("field2",
867
parameters[0].item.fields["field2"].name)
868
869
def test_schema_conversion_optional_list(self):
870
"""
871
Backwards-compatibility conversions maintains optional-ness of lists.
872
"""
873
schema = Schema(Unicode("foos.N", optional=True))
874
arguments, _ = schema.extract({})
875
self.assertEqual([], arguments.foos)
876
877
def test_schema_conversion_optional_structure_field(self):
878
"""
879
Backwards-compatibility conversion maintains optional-ness of structure
880
fields.
881
"""
882
schema = Schema(Unicode("foos.N.field"),
883
Unicode("foos.N.field2", optional=True, default=u"hi"))
884
arguments, _ = schema.extract({"foos.0.field": u"existent"})
885
self.assertEqual(u"existent", arguments.foos[0].field)
886
self.assertEqual(u"hi", arguments.foos[0].field2)
887
888
def test_additional_schema_attributes(self):
889
"""
890
Additional data can be specified on the Schema class for specifying a
891
more rich schema.
892
"""
893
result = {'id': Integer(), 'name': Unicode(), 'data': RawStr()}
894
errors = [APIError]
895
896
schema = Schema(
897
name="GetStuff",
898
doc="""Get the stuff.""",
899
parameters=[
900
Integer("id"),
901
Unicode("scope")],
902
result=result,
903
errors=errors)
904
905
self.assertEqual("GetStuff", schema.name)
906
self.assertEqual("Get the stuff.", schema.doc)
907
self.assertEqual(result, schema.result)
908
self.assertEqual(set(errors), schema.errors)
909
910
def test_extend_with_additional_schema_attributes(self):
911
"""
912
The additional schema attributes can be passed to L{Schema.extend}.
913
"""
914
result = {'id': Integer(), 'name': Unicode(), 'data': RawStr()}
915
errors = [APIError]
916
917
schema = Schema(
918
name="GetStuff",
919
parameters=[Integer("id")])
920
921
schema2 = schema.extend(
922
name="GetStuff2",
923
doc="Get stuff 2",
924
parameters=[Unicode("scope")],
925
result=result,
926
errors=errors)
927
928
self.assertEqual("GetStuff2", schema2.name)
929
self.assertEqual("Get stuff 2", schema2.doc)
930
self.assertEqual(result, schema2.result)
931
self.assertEqual(set(errors), schema2.errors)
932
933
arguments, _ = schema2.extract({'id': '5', 'scope': u'foo'})
934
self.assertEqual(5, arguments.id)
935
self.assertEqual(u'foo', arguments.scope)
936
937
def test_extend_maintains_existing_attributes(self):
938
"""
939
If additional schema attributes aren't passed to L{Schema.extend}, they
940
stay the same.
941
"""
942
result = {'id': Integer(), 'name': Unicode(), 'data': RawStr()}
943
errors = [APIError]
944
945
schema = Schema(
946
name="GetStuff",
947
doc="""Get the stuff.""",
948
parameters=[Integer("id")],
949
result=result,
950
errors=errors)
951
952
schema2 = schema.extend(parameters=[Unicode("scope")])
953
954
self.assertEqual("GetStuff", schema2.name)
955
self.assertEqual("Get the stuff.", schema2.doc)
956
self.assertEqual(result, schema2.result)
957
self.assertEqual(set(errors), schema2.errors)
958
959
arguments, _ = schema2.extract({'id': '5', 'scope': u'foo'})
960
self.assertEqual(5, arguments.id)
961
self.assertEqual(u'foo', arguments.scope)
962
963
def test_extend_result(self):
964
"""
965
Result fields can also be extended with L{Schema.extend}.
966
"""
967
schema = Schema(result={'name': Unicode()})
968
schema2 = schema.extend(
969
result={'id': Integer()})
970
result_structure = Structure(fields=schema2.result)
971
self.assertEqual(
972
{'name': u'foo', 'id': 5},
973
result_structure.coerce({'name': u'foo', 'id': '5'}))
974
975
def test_extend_errors(self):
976
"""
977
Errors can be extended with L{Schema.extend}.
978
"""
979
schema = Schema(parameters=[], errors=[APIError])
980
schema2 = schema.extend(errors=[ZeroDivisionError])
981
self.assertEqual(set([APIError, ZeroDivisionError]), schema2.errors)
982
983
def test_extend_maintains_parameter_order(self):
984
"""
985
Extending a schema with additional parameters puts the new parameters
986
at the end.
987
"""
988
schema = Schema(parameters=[Unicode("name"), Unicode("value")])
989
schema2 = schema.extend(parameters=[Integer("foo"), Unicode("index")])
990
self.assertEqual(["name", "value", "foo", "index"],
991
[p.name for p in schema2.get_parameters()])
992
993
def test_schema_field_names(self):
994
structure = Structure(fields={"foo": Integer()})
995
self.assertEqual("foo", structure.fields["foo"].name)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1