import copy import unittest from odexa_ref import contracts as c, evidence as e ORIGIN = "https://publisher.example" TIME = "2026-09-16T12:00:00Z" RID = "11111111-1111-4111-8111-111111111111" def authority(): return dict(origin=ORIGIN, service_id=ORIGIN + "/odexa/services/local", revision=1, url=ORIGIN + "/odexa-service/history/1.json", digest=c.digest(b"authority")) def manifest(): return dict(protocol_version=c.VERSION, origin=ORIGIN, asset_id=ORIGIN + "/assets/article", version_id=ORIGIN + "/assets/article/versions/1", published_at=TIME, authority=authority(), representations=[dict(representation_id=RID, media_type="text/plain", media_parameters={"charset": "utf-8"}, languages=["en"], decoded_length=5, decoded_digest=c.digest(b"hello"))]) def event(method="GET", status=200, body=b"hello", **kwargs): m = manifest() ref = dict(asset_id=m["asset_id"], version_id=m["version_id"], manifest_digest=c.digest(c.json_bytes(m)), representation_id=RID) return e.make_delivery_event(origin=ORIGIN, reporter_id=ORIGIN + "/gateway", resource_url=ORIGIN + "/licensed/article", policy_id=ORIGIN + "/policies/1", policy_revision=1, occurred_at=TIME, method=method, status=status, body=body, asset_ref=ref, media_type="text/plain", media_parameters={"charset": "utf-8"}, languages=["en"], **kwargs) class JSONContractTests(unittest.TestCase): def test_lexical_invalid_json(self): for raw in (b'{"a":1,"a":2}', b'{"a":1.0}', b'{"a":1e0}', b'{"a":-0}', b'{"a":NaN}', b'{"a":9007199254740992}', b'[]', b'{"a":"\\ud800"}', b'\xff'): with self.subTest(raw=raw), self.assertRaises(c.ValidationError): c.load_json(raw) def test_depth_and_size(self): with self.assertRaises(c.ValidationError): c.load_json(b'{"a":' + b'[' * 33 + b'0' + b']' * 33 + b'}') with self.assertRaises(c.ValidationError): c.load_json(b'{"a":"' + b'x' * c.MAX_BYTES + b'"}') def test_valid_json_and_calendar(self): self.assertEqual(c.load_json(b'{"a":0,"b":false,"c":null}'), {"a": 0, "b": False, "c": None}) m = manifest() m["published_at"] = "2026-02-30T12:00:00Z" with self.assertRaises(c.ValidationError): c.validate_manifest(m) class AssetContractTests(unittest.TestCase): def test_manifest_valid(self): self.assertIs(c.validate_manifest(manifest())["representations"][0]["decoded_length"], 5) def test_closed_and_origin_fields(self): for key, value in (("unknown", 1), ("version_id", "https://elsewhere.example/v/1"), ("asset_id", ORIGIN + "/a?x"), ("protocol_version", "1.2.0")): m = manifest() m[key] = value with self.subTest(key=key), self.assertRaises(c.ValidationError): c.validate_manifest(m) def test_duplicate_representation(self): m = manifest() m["representations"].append(copy.deepcopy(m["representations"][0])) with self.assertRaises(c.ValidationError): c.validate_manifest(m) def test_media_metadata(self): for key, value in (("media_type", "Text/plain"), ("media_parameters", {"Charset": "utf-8"}), ("languages", ["en", "en"]), ("decoded_length", True), ("decoded_digest", "sha256:no")): m = manifest() m["representations"][0][key] = value with self.subTest(key=key), self.assertRaises(c.ValidationError): c.validate_manifest(m) def test_authority_not_arbitrary_history_url(self): m = manifest() m["authority"]["url"] = ORIGIN + "/latest.json" with self.assertRaises(c.ValidationError): c.validate_manifest(m) class DeliveryContractTests(unittest.TestCase): def test_http_matrix(self): cases = [("GET", 200, "delivery.completed", "full"), ("HEAD", 200, "response.completed", "head"), ("GET", 204, "response.completed", "no_content"), ("GET", 205, "response.completed", "no_content"), ("GET", 304, "response.completed", "not_modified"), ("HEAD", 304, "response.completed", "not_modified"), ("GET", 301, "response.completed", "redirect"), ("GET", 404, "response.completed", "error"), ("GET", 503, "response.completed", "error"), ("GET", 206, "response.completed", "unsupported_partial"), ("HEAD", 206, "response.completed", "unsupported_status"), ("GET", 203, "response.completed", "unsupported_status"), ("GET", 101, "response.completed", "unsupported_status")] for method, status, typ, kind in cases: with self.subTest(method=method, status=status): self.assertEqual(c.classify_http(method, status), (typ, kind)) def test_failed_prefix_is_never_complete(self): v = event(body=b"he", complete=False) self.assertEqual(v["event_type"], "delivery.failed") self.assertEqual(e.validate_binding(v, manifest()), "referenced_only") def test_status_and_source_spoof(self): v = event() with self.assertRaises(c.ValidationError): c.validate_event(v, role="agent") v["http"]["status"] = 304 with self.assertRaises(c.ValidationError): c.validate_event(v) def test_head_and_no_content_bytes(self): for method, status in (("HEAD", 200), ("GET", 204), ("GET", 304)): with self.subTest(method=method, status=status): v = event(method, status, b"") self.assertIsNone(v["http"]["content_digest"]) with self.assertRaises(c.ValidationError): event(method, status, b"illegal") def test_supported_partial(self): v = event(status=206, body=b"ell", byte_range=dict(first=1, last=3, complete_length=5)) self.assertEqual(v["http"]["kind"], "partial") self.assertEqual(e.validate_binding(v, manifest()), "referenced_only") def test_partial_rejects_coding_and_counter_mismatch(self): for args in (dict(content_codings=["gzip"]), dict(body=b"el")): values = dict(status=206, body=b"ell", byte_range=dict(first=1, last=3, complete_length=5)) values.update(args) with self.assertRaises(c.ValidationError): event(**values) def test_unsupported_coded_partial_keeps_encoded_bytes(self): v = event(status=206, body=b"encoded fragment", content_codings=["gzip"]) self.assertEqual(v["http"]["kind"], "unsupported_partial") self.assertIsNone(v["http"]["decoded_digest"]) def test_terminal_interim_and_bad_status_rejected(self): for status in (100, 103, 600, True, "200"): with self.subTest(status=status), self.assertRaises(c.ValidationError): c.classify_http("GET", status) def test_exact_origin_and_reporter_binding(self): for kwargs in (dict(origin="https://other.example"), dict(reporter_id=ORIGIN + "/other")): with self.assertRaises(c.ValidationError): c.validate_event(event(), **kwargs) def test_intent_arrays_and_policy_pair(self): v = event() v["actions"] = ["retrieve"] with self.assertRaises(c.ValidationError): c.validate_event(v) v = event() v["policy_id"] = None with self.assertRaises(c.ValidationError): c.validate_event(v) if __name__ == "__main__": unittest.main()