Spaces:
Running
Running
File size: 6,812 Bytes
917b35d | 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 146 147 148 | from __future__ import annotations
import copy
import json
import os
import sys
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
import server # noqa: E402
class AnatomyContractTest(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.httpd = server.make_server("127.0.0.1", 0)
cls.port = cls.httpd.server_address[1]
cls.base = f"http://127.0.0.1:{cls.port}"
cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
cls.thread.start()
@classmethod
def tearDownClass(cls) -> None:
cls.httpd.shutdown()
cls.httpd.server_close()
cls.thread.join(timeout=3)
def request(self, path: str, *, method: str = "GET", body=None):
data = None
headers = {}
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(self.base + path, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=5) as response:
return response.status, dict(response.headers), json.load(response)
except urllib.error.HTTPError as exc:
return exc.code, dict(exc.headers), json.load(exc)
def test_health_is_transport_only(self):
status, headers, body = self.request("/healthz")
self.assertEqual(200, status)
self.assertEqual("REACHABLE", body["transport_state"])
self.assertEqual("SNAPSHOT", body["evidence_state"])
self.assertEqual("STRUCTURAL_ONLY", body["verification_state"])
self.assertEqual("READ_ONLY", body["authority_state"])
self.assertIn("quality", body["note"])
self.assertEqual("REACHABLE", headers["X-SZL-Transport-State"])
self.assertEqual("nosniff", headers["X-Content-Type-Options"])
def test_static_bundle_is_served_from_the_anatomy_root(self):
for path, marker in (("/", "SZL Living Anatomy"), ("/frontier_anatomy.js", "Evidence Bay")):
with urllib.request.urlopen(self.base + path, timeout=5) as response:
body = response.read().decode("utf-8")
self.assertEqual(200, response.status)
self.assertIn(marker, body)
def test_manifest_separates_state_dimensions(self):
status, headers, body = self.request("/api/anatomy/v1/manifest")
self.assertEqual(200, status)
self.assertEqual("szl.anatomy-manifest/v1", body["schema"])
self.assertEqual(
{"transport_state", "evidence_state", "verification_state", "authority_state"},
set(body["state_dimensions"]),
)
self.assertEqual("CONJECTURE_1", body["doctrine"]["lambda"])
self.assertEqual(8, len(body["doctrine"]["locked_proven_declared"]))
self.assertEqual("*", headers["Access-Control-Allow-Origin"])
def test_every_capability_has_five_part_shell(self):
status, _, body = self.request("/api/anatomy/v1/capabilities")
self.assertEqual(200, status)
self.assertGreaterEqual(body["count"], 5)
for capability in body["capabilities"]:
for key in ("purpose", "try", "evidence", "limits", "reproduce"):
self.assertIn(key, capability, f"{capability['id']} lacks {key}")
self.assertEqual("READ_ONLY", capability["authority_state"])
formula = next(item for item in body["capabilities"] if item["id"] == "anatomy.formula-spine")
self.assertEqual("SNAPSHOT", formula["evidence"]["state"])
self.assertTrue(all(url.startswith("https://github.com/") for url in formula["provenance"]))
def test_receipt_replays_as_structural_only(self):
status, receipt_headers, receipt = self.request("/api/anatomy/v1/receipt")
self.assertEqual(200, status)
self.assertEqual("STRUCTURAL_ONLY", receipt["verification_state"])
self.assertEqual(64, len(receipt["receipt_id"]))
self.assertEqual("STRUCTURAL_ONLY", receipt_headers["X-SZL-Verification-State"])
verify_status, verify_headers, verified = self.request(
"/api/anatomy/v1/verify/receipt", method="POST", body=receipt
)
self.assertEqual(200, verify_status)
self.assertEqual("STRUCTURAL-ONLY", verified["verdict"])
self.assertEqual("STRUCTURAL_ONLY", verify_headers["X-SZL-Verification-State"])
checks = {item["name"]: item["status"] for item in verified["checks"]}
self.assertEqual("PASS", checks["artifact_set"])
self.assertEqual("UNAVAILABLE", checks["signature"])
def test_tampered_receipt_fails(self):
_, _, receipt = self.request("/api/anatomy/v1/receipt")
tampered = copy.deepcopy(receipt)
tampered["receipt"]["claim"]["purpose"] = "changed"
status, _, result = self.request(
"/api/anatomy/v1/verify/receipt", method="POST", body=tampered
)
self.assertEqual(400, status)
self.assertEqual("FAIL", result["verdict"])
checks = {item["name"]: item["status"] for item in result["checks"]}
self.assertEqual("FAIL", checks["receipt_digest"])
def test_source_attestation_matches_estate_schema(self):
previous = os.environ.get("SPACE_REPOSITORY_COMMIT")
os.environ["SPACE_REPOSITORY_COMMIT"] = "a" * 40
try:
status, _, body = self.request("/.well-known/szl-source.json")
finally:
if previous is None:
os.environ.pop("SPACE_REPOSITORY_COMMIT", None)
else:
os.environ["SPACE_REPOSITORY_COMMIT"] = previous
self.assertEqual(200, status)
self.assertEqual("szl.deployment-source/v1", body["schema"])
self.assertEqual("szl-holdings/anatomy", body["source"]["repository"])
self.assertEqual("SZLHOLDINGS/anatomy", body["deployment"]["hf_space"])
self.assertEqual("a" * 40, body["deployment"]["hf_revision"])
self.assertEqual("PENDING_GITHUB_SYNC", body["alignment_state"])
def test_frontend_and_public_verifier_use_current_contract(self):
index = (ROOT / "index.html").read_text(encoding="utf-8")
bay = (ROOT / "frontier_anatomy.js").read_text(encoding="utf-8")
widget = (ROOT / "lib" / "szl_verify_widget.js").read_text(encoding="utf-8")
self.assertIn('src="./frontier_anatomy.js"', index)
self.assertIn("Evidence Bay", bay)
self.assertIn("/api/anatomy/v1", bay)
self.assertIn("/api/a11oy/v1/verify/receipt", widget)
self.assertNotIn("base+'/api/a11oy/v1/verify'", widget)
if __name__ == "__main__":
unittest.main(verbosity=2)
|