Repository navigation
Expand file tree
/
Copy pathtest_schema.py
More file actions
145 lines (121 loc) · 6.3 KB
/
Copy pathtest_schema.py
File metadata and controls
145 lines (121 loc) · 6.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
from __future__ import annotations
import json
import os
import tempfile
import unittest
from typing import List
from babelqueue.exceptions import InvalidPayloadError
from babelqueue.schema import (
DirProvider,
MapProvider,
check,
validate,
validate_schema,
wrap,
)
ORDERS = (
'{"type":"object","required":["order_id"],'
'"properties":{"order_id":{"type":"integer"}},"additionalProperties":false}'
)
class ValidatorTest(unittest.TestCase):
def test_object_required_types_and_additional(self) -> None:
schema = json.loads(
'{"type":"object","required":["order_id"],'
'"properties":{"order_id":{"type":"integer"},"note":{"type":"string","minLength":1}},'
'"additionalProperties":false}'
)
self.assertIsNone(validate_schema(schema, {"order_id": 7}))
self.assertIsNotNone(validate_schema(schema, {}))
self.assertIsNotNone(validate_schema(schema, {"order_id": "x"}))
self.assertIsNotNone(validate_schema(schema, {"order_id": 7, "extra": 1}))
self.assertIsNotNone(validate_schema(schema, {"order_id": 7, "note": ""}))
def test_enum_minimum_and_array_items(self) -> None:
schema = json.loads(
'{"type":"object","properties":{"status":{"enum":["new","paid"]},'
'"qty":{"type":"integer","minimum":1},'
'"tags":{"type":"array","items":{"type":"string"}}}}'
)
self.assertIsNone(validate_schema(schema, {"status": "paid", "qty": 2, "tags": ["a", "b"]}))
self.assertIsNotNone(validate_schema(schema, {"status": "cancelled"}))
self.assertIsNotNone(validate_schema(schema, {"qty": 0}))
self.assertIsNotNone(validate_schema(schema, {"tags": ["a", 1]}))
def test_scalar_types(self) -> None:
cases = [
('{"type":"boolean"}', True, True),
('{"type":"boolean"}', "x", False),
('{"type":"null"}', None, True),
('{"type":"null"}', 1, False),
('{"type":"number","minimum":0.5}', 0.6, True),
('{"type":"number","minimum":0.5}', 0.4, False),
('{"type":"number"}', "x", False),
('{"type":"string"}', 5, False),
('{"type":"integer"}', 1.0, True),
('{"type":"integer"}', 1.5, False),
('{"type":"integer"}', True, False), # bool is not an integer
('{"const":"v1"}', "v1", True),
('{"const":"v1"}', "v2", False),
]
for src, value, valid in cases:
schema = json.loads(src)
violation = validate_schema(schema, value)
self.assertEqual(valid, violation is None, f"{src} / {value!r}: {violation}")
class ProviderTest(unittest.TestCase):
def test_map_provider_from_json(self) -> None:
provider = MapProvider.from_json({"urn:babel:orders:created": ORDERS})
self.assertIsNotNone(provider.schema_for("urn:babel:orders:created"))
self.assertIsNone(provider.schema_for("urn:babel:unknown"))
def test_map_provider_invalid_json(self) -> None:
with self.assertRaises(ValueError):
MapProvider.from_json({"u": "not json"})
def test_dir_provider_reads_registry_lazily(self) -> None:
with tempfile.TemporaryDirectory() as d:
os.makedirs(os.path.join(d, "schemas"))
with open(os.path.join(d, "schemas", "orders.json"), "w", encoding="utf-8") as fh:
fh.write('{"type":"object","required":["order_id"],"properties":{"order_id":{"type":"integer"}}}')
with open(os.path.join(d, "registry.json"), "w", encoding="utf-8") as fh:
fh.write('{"schemas":[{"urn":"urn:babel:orders:created","schema":"schemas/orders.json"},{"urn":"","schema":"x"}]}')
provider = DirProvider(os.path.join(d, "registry.json"))
for _ in range(2): # the second call hits the cache
self.assertIsNotNone(provider.schema_for("urn:babel:orders:created"))
self.assertIsNone(provider.schema_for("urn:babel:unknown"))
def test_dir_provider_missing_manifest(self) -> None:
with self.assertRaises(OSError):
DirProvider(os.path.join(tempfile.gettempdir(), "nope_registry_xyz.json"))
def test_dir_provider_missing_schema_file(self) -> None:
with tempfile.TemporaryDirectory() as d:
with open(os.path.join(d, "registry.json"), "w", encoding="utf-8") as fh:
fh.write('{"schemas":[{"urn":"u","schema":"missing.json"}]}')
provider = DirProvider(os.path.join(d, "registry.json"))
with self.assertRaises(OSError):
provider.schema_for("u")
class FacadeTest(unittest.TestCase):
def _provider(self) -> MapProvider:
return MapProvider.from_json({"urn:babel:orders:created": ORDERS})
def test_check_valid_invalid_unregistered(self) -> None:
provider = self._provider()
self.assertIsNone(check(provider, "urn:babel:orders:created", {"order_id": 1}))
self.assertIsNone(check(provider, "urn:babel:unknown", {"x": 1})) # opt-in
self.assertIsNotNone(check(provider, "urn:babel:orders:created", {}))
def test_validate_raises_on_invalid(self) -> None:
with self.assertRaises(InvalidPayloadError):
validate(self._provider(), "urn:babel:orders:created", {"order_id": "x"})
def test_validate_passes_unregistered(self) -> None:
validate(self._provider(), "urn:babel:unknown", {"anything": True}) # no raise
def test_wrap_runs_handler_on_valid(self) -> None:
calls: List[int] = []
wrapped = wrap(self._provider(), "urn:babel:orders:created", lambda data, meta: calls.append(1))
wrapped({"order_id": 1}, {"id": "m1"})
self.assertEqual(calls, [1])
def test_wrap_raises_and_skips_on_invalid(self) -> None:
calls: List[int] = []
wrapped = wrap(self._provider(), "urn:babel:orders:created", lambda data, meta: calls.append(1))
with self.assertRaises(InvalidPayloadError):
wrapped({}, {"id": "m1"})
self.assertEqual(calls, [])
def test_wrap_runs_handler_for_unregistered_urn(self) -> None:
calls: List[int] = []
wrapped = wrap(self._provider(), "urn:babel:unknown", lambda data, meta: calls.append(1))
wrapped({"anything": True}, {"id": "m1"})
self.assertEqual(calls, [1])
if __name__ == "__main__":
unittest.main()