Spaces:
Running
Running
| # SPDX-License-Identifier: Apache-2.0 | |
| # © 2026 Lutar, Stephen P. — SZL Holdings · ORCID 0009-0001-0110-4173 · Doctrine v11 | |
| # Co-Authored-By: Perplexity Computer Agent <agent@perplexity.ai> | |
| # Signed-off-by: Stephen P. Lutar Jr. <stephenlutar2@gmail.com> | |
| """ | |
| Tests for szl_parity_gaps — parity gap closure + differentiator endpoints. | |
| Coverage: | |
| GAP-A: /api/a11oy/v1/compliance/export | |
| GAP-B: /api/a11oy/v1/lineage | |
| GAP-C: /api/a11oy/v1/policy/validate | |
| DIFF-1: /api/a11oy/v1/receipts/replay | |
| DIFF-2: /api/a11oy/v1/lambda/score | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import pytest | |
| from fastapi import FastAPI | |
| from fastapi.testclient import TestClient | |
| import sys, os | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| import szl_parity_gaps as pg | |
| # --------------------------------------------------------------------------- | |
| # Minimal gate manifest stubs for tests | |
| # --------------------------------------------------------------------------- | |
| _STUB_GATES = [ | |
| {"name": "thresholdPolicySeverity", "file": "thresholdPolicySeverity_gate.ts", | |
| "description": "Threshold gate", "lean_theorem": "ThresholdAxiom", | |
| "lean_file": "Lutar/LambdaInvariant/Boundary.lean", "lean_verified": True}, | |
| {"name": "adversarialRobustness", "file": "adversarialRobustness_gate.ts", | |
| "description": "Adversarial robustness", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "hashChainIntegrity", "file": "hashChainIntegrity_gate.ts", | |
| "description": "Hash chain", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "merkleDagBatch", "file": "merkleDagBatch_gate.ts", | |
| "description": "Merkle DAG", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "doctrineCompleteness", "file": "doctrineCompleteness_gate.ts", | |
| "description": "Doctrine completeness", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "constructiveTransparency", "file": "constructiveTransparency_gate.ts", | |
| "description": "Transparency", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "soundnessAxiom", "file": "soundnessAxiom_gate.ts", | |
| "description": "Soundness", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "deterministicReplay", "file": "deterministicReplay_gate.ts", | |
| "description": "Deterministic replay", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "certifiedRobustness", "file": "certifiedRobustness_gate.ts", | |
| "description": "Certified robustness", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "bekensteinBound", "file": "bekensteinBound_gate.ts", | |
| "description": "Bekenstein", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "composability", "file": "composability_gate.ts", | |
| "description": "Composability", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "crossRegionPolicy", "file": "crossRegionPolicy_gate.ts", | |
| "description": "Cross-region", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "witnessQuorum", "file": "witnessQuorum_gate.ts", | |
| "description": "Witness quorum", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "humanEscalation", "file": "humanEscalation_gate.ts", | |
| "description": "Human escalation", "lean_theorem": "", "lean_verified": False}, | |
| {"name": "provenance", "file": "provenance_gate.ts", | |
| "description": "Provenance", "lean_theorem": "", "lean_verified": False}, | |
| ] | |
| _STUB_BY_NAME = {g["name"]: g for g in _STUB_GATES} | |
| def client(): | |
| app = FastAPI() | |
| pg.register(app, _STUB_GATES, _STUB_BY_NAME) | |
| return TestClient(app, raise_server_exceptions=True) | |
| # --------------------------------------------------------------------------- | |
| # GAP-A: Compliance Evidence Export | |
| # --------------------------------------------------------------------------- | |
| class TestComplianceExport: | |
| def test_eu_ai_act_200(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=eu-ai-act") | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["framework"] == "eu-ai-act" | |
| assert d["schema"] == "szl.compliance.export.v1" | |
| assert isinstance(d["controls"], list) | |
| assert len(d["controls"]) > 0 | |
| assert d["doctrine"] == "v11" | |
| def test_nist_ai_rmf_200(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=nist-ai-rmf") | |
| assert r.status_code == 200 | |
| assert r.json()["framework"] == "nist-ai-rmf" | |
| def test_iso_42001_200(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=iso-42001") | |
| assert r.status_code == 200 | |
| assert r.json()["framework"] == "iso-42001" | |
| def test_soc2_200(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=soc2") | |
| assert r.status_code == 200 | |
| assert r.json()["framework"] == "soc2" | |
| def test_unknown_framework_400(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=fedramp-high") | |
| assert r.status_code == 400 | |
| d = r.json() | |
| assert "available" in d | |
| def test_controls_have_gate_evidence(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=eu-ai-act") | |
| d = r.json() | |
| for ctrl in d["controls"]: | |
| assert "gate_evidence" in ctrl | |
| assert isinstance(ctrl["gate_evidence"], list) | |
| def test_lambda_conjecture_labelled(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export?framework=eu-ai-act") | |
| d = r.json() | |
| assert "Conjecture" in d["lambda"] | |
| def test_default_framework_is_eu_ai_act(self, client): | |
| r = client.get("/api/a11oy/v1/compliance/export") | |
| assert r.status_code == 200 | |
| assert r.json()["framework"] == "eu-ai-act" | |
| # --------------------------------------------------------------------------- | |
| # GAP-B: Decision Lineage Query | |
| # --------------------------------------------------------------------------- | |
| class TestDecisionLineage: | |
| def test_lineage_200(self, client): | |
| r = client.get("/api/a11oy/v1/lineage") | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["schema"] == "szl.lineage.v1" | |
| assert "nodes" in d | |
| assert d["doctrine"] == "v11" | |
| def test_lineage_empty_on_cold_dag(self, client): | |
| """On a cold start with no real DAG the endpoint returns honest empty nodes.""" | |
| r = client.get("/api/a11oy/v1/lineage?limit=10") | |
| d = r.json() | |
| # Either empty (cold start) or has nodes (if DAG populated) | |
| assert isinstance(d["nodes"], list) | |
| assert "returned" in d | |
| def test_lineage_limit_honoured(self, client): | |
| r = client.get("/api/a11oy/v1/lineage?limit=5") | |
| d = r.json() | |
| assert d["limit"] == 5 | |
| assert len(d["nodes"]) <= 5 | |
| def test_lineage_limit_capped_at_200(self, client): | |
| r = client.get("/api/a11oy/v1/lineage?limit=9999") | |
| d = r.json() | |
| assert d["limit"] == 200 | |
| def test_lineage_leader_parity_refs(self, client): | |
| r = client.get("/api/a11oy/v1/lineage") | |
| d = r.json() | |
| lp = d.get("leader_parity", {}) | |
| assert "palantir" in " ".join(lp.values()).lower() | |
| # --------------------------------------------------------------------------- | |
| # GAP-C: Policy-as-Code Validation | |
| # --------------------------------------------------------------------------- | |
| class TestPolicyValidate: | |
| def test_valid_policy(self, client): | |
| payload = { | |
| "name": "test-policy", | |
| "gates": ["thresholdPolicySeverity", "adversarialRobustness"], | |
| "lambda_floor": 0.90, | |
| "min_witnesses": 2, | |
| "severity": "medium", | |
| } | |
| r = client.post("/api/a11oy/v1/policy/validate", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["valid"] is True | |
| assert d["violations"] == [] | |
| def test_missing_name_invalid(self, client): | |
| payload = {"gates": ["thresholdPolicySeverity"]} | |
| r = client.post("/api/a11oy/v1/policy/validate", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["valid"] is False | |
| assert any(v["field"] == "name" for v in d["violations"]) | |
| def test_unknown_gate_invalid(self, client): | |
| payload = {"name": "test", "gates": ["nonexistent_gate_xyz"]} | |
| r = client.post("/api/a11oy/v1/policy/validate", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["valid"] is False | |
| assert any("Unknown gates" in v["message"] for v in d["violations"]) | |
| def test_lambda_floor_out_of_range(self, client): | |
| payload = {"name": "test", "gates": ["thresholdPolicySeverity"], | |
| "lambda_floor": 1.5} | |
| r = client.post("/api/a11oy/v1/policy/validate", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["valid"] is False | |
| assert any("lambda_floor" in v["field"] for v in d["violations"]) | |
| def test_capital_severity_quorum_check(self, client): | |
| payload = {"name": "test", "gates": ["thresholdPolicySeverity"], | |
| "severity": "capital", "min_witnesses": 2} | |
| r = client.post("/api/a11oy/v1/policy/validate", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| # capital requires 3-of-N quorum | |
| assert d["required_quorum"] == 3 | |
| assert d["valid"] is False | |
| def test_gate_coverage_reported(self, client): | |
| payload = {"name": "test", "gates": ["thresholdPolicySeverity"]} | |
| r = client.post("/api/a11oy/v1/policy/validate", json=payload) | |
| d = r.json() | |
| assert "gate_coverage" in d | |
| assert d["gate_coverage"]["known"] >= 1 | |
| def test_invalid_json_400(self, client): | |
| r = client.post("/api/a11oy/v1/policy/validate", | |
| content=b"not json", headers={"Content-Type": "application/json"}) | |
| assert r.status_code == 400 | |
| # --------------------------------------------------------------------------- | |
| # DIFF-1: Receipt Replay | |
| # --------------------------------------------------------------------------- | |
| class TestReceiptReplay: | |
| def test_replay_allow(self, client): | |
| payload = { | |
| "action": { | |
| "severity": "medium", | |
| "confidence": 0.85, | |
| "actionId": "replay-test-001", | |
| "witnesses": [ | |
| {"id": "w1", "role": "op", "attested": True}, | |
| {"id": "w2", "role": "auditor", "attested": True}, | |
| ], | |
| }, | |
| "gate": "thresholdPolicySeverity", | |
| } | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["schema"] == "szl.receipt.replay.v1" | |
| assert d["replay_decision"] in ("allow", "deny") | |
| assert "replay_receipt_hash" in d | |
| assert "determinism_note" in d | |
| def test_replay_deny_low_confidence(self, client): | |
| payload = { | |
| "action": { | |
| "severity": "high", | |
| "confidence": 0.5, | |
| "actionId": "replay-deny-001", | |
| "witnesses": [{"id": "w1", "role": "op", "attested": True}], | |
| }, | |
| "gate": "thresholdPolicySeverity", | |
| } | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert d["replay_decision"] == "deny" | |
| def test_replay_hash_comparison(self, client): | |
| payload = { | |
| "action": {"severity": "medium", "confidence": 0.85, "actionId": "a", | |
| "witnesses": [{"id": "w1", "attested": True}, | |
| {"id": "w2", "attested": True}]}, | |
| "original_receipt_hash": "deadbeef1234567890", | |
| "gate": "thresholdPolicySeverity", | |
| } | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| d = r.json() | |
| assert "hashes_match" in d | |
| # Hashes won't match (different hash), but the field must be present | |
| assert d["hashes_match"] is False # deadbeef won't match real sha256 | |
| def test_replay_unknown_gate_400(self, client): | |
| payload = {"action": {"severity": "medium"}, "gate": "nonexistent_gate_xyz"} | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| assert r.status_code == 400 | |
| def test_replay_no_action_400(self, client): | |
| payload = {"gate": "thresholdPolicySeverity"} | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| assert r.status_code == 400 | |
| def test_replay_dsse_envelope_present(self, client): | |
| payload = { | |
| "action": {"severity": "medium", "confidence": 0.85, "actionId": "a", | |
| "witnesses": [{"id": "w1", "attested": True}, | |
| {"id": "w2", "attested": True}]}, | |
| "gate": "thresholdPolicySeverity", | |
| } | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| d = r.json() | |
| assert "replay_dsse" in d | |
| # replay_dsse may wrap a {receipt, dsse} or a {payloadType, ...} envelope | |
| env = d["replay_dsse"] | |
| has_payload_type = "payloadType" in env | |
| has_nested_dsse = "dsse" in env or "receipt" in env | |
| assert has_payload_type or has_nested_dsse, f"unexpected dsse shape: {list(env.keys())}" | |
| def test_replay_differentiator_note_present(self, client): | |
| payload = { | |
| "action": {"severity": "medium", "confidence": 0.85, "actionId": "a", | |
| "witnesses": [{"id": "w1", "attested": True}, | |
| {"id": "w2", "attested": True}]}, | |
| } | |
| r = client.post("/api/a11oy/v1/receipts/replay", json=payload) | |
| d = r.json() | |
| assert "differentiator" in d | |
| # --------------------------------------------------------------------------- | |
| # DIFF-2: Λ-Gated Decision Scoring | |
| # --------------------------------------------------------------------------- | |
| class TestLambdaScore: | |
| _PERFECT = {a: 1.0 for a in pg.CANONICAL_AXES} | |
| _TYPICAL = {a: 0.91 for a in pg.CANONICAL_AXES} | |
| _ZERO_ONE = {**{a: 0.91 for a in pg.CANONICAL_AXES}, "soundness": 0.0} | |
| def test_perfect_axes_lambda_one(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._PERFECT}) | |
| assert r.status_code == 200 | |
| d = r.json() | |
| assert abs(d["lambda"] - 1.0) < 1e-6 | |
| assert d["gate_pass"] is True | |
| def test_typical_axes_lambda_in_range(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._TYPICAL}) | |
| d = r.json() | |
| lv = d["lambda"] | |
| assert 0.8 < lv < 1.0 | |
| def test_zero_axis_zero_pinning(self, client): | |
| """A2: zero-pinning — Λ = 0 when any positive-weight axis is 0.""" | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._ZERO_ONE}) | |
| d = r.json() | |
| assert d["lambda"] == 0.0 | |
| assert d["zero_pinned"] is True | |
| assert d["gate_pass"] is False | |
| def test_gate_pass_with_high_floor(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", | |
| json={"axes": self._TYPICAL, "lambda_floor": 0.99}) | |
| d = r.json() | |
| assert d["gate_pass"] is False | |
| def test_gate_decision_deny(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", | |
| json={"axes": self._ZERO_ONE, "lambda_floor": 0.90}) | |
| d = r.json() | |
| assert d["gate_decision"] == "deny" | |
| def test_schema_correct(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._TYPICAL}) | |
| d = r.json() | |
| assert d["schema"] == "szl.lambda.score.v1" | |
| assert d["doctrine"] == "v11" | |
| def test_lean_citation_conjecture(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._TYPICAL}) | |
| d = r.json() | |
| lc = d["lean_citation"] | |
| assert "Conjecture 1" in lc["status"] | |
| assert "NOT a closed theorem" in lc["status"] | |
| def test_axes_breakdown_present(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._TYPICAL}) | |
| d = r.json() | |
| assert len(d["axes_computed"]) == 9 | |
| def test_missing_axes_flagged(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", | |
| json={"axes": {"soundness": 0.9}}) | |
| d = r.json() | |
| assert len(d["missing_axes"]) == 8 | |
| def test_monotonicity_a1(self, client): | |
| """A1: raising one axis while others fixed should not decrease Λ.""" | |
| base = {a: 0.80 for a in pg.CANONICAL_AXES} | |
| high = {**base, "soundness": 0.95} | |
| r_base = client.post("/api/a11oy/v1/lambda/score", json={"axes": base}) | |
| r_high = client.post("/api/a11oy/v1/lambda/score", json={"axes": high}) | |
| assert r_high.json()["lambda"] >= r_base.json()["lambda"] | |
| def test_no_action_400(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"lambda_floor": 0.9}) | |
| assert r.status_code == 400 | |
| def test_differentiator_note_present(self, client): | |
| r = client.post("/api/a11oy/v1/lambda/score", json={"axes": self._TYPICAL}) | |
| d = r.json() | |
| assert "differentiator" in d | |
| # --------------------------------------------------------------------------- | |
| # Lambda computation unit tests (no HTTP) | |
| # --------------------------------------------------------------------------- | |
| class TestLambdaComputation: | |
| def test_perfect_is_one(self): | |
| axes = {a: 1.0 for a in pg.CANONICAL_AXES} | |
| result = pg._compute_lambda(axes) | |
| assert abs(result["lambda"] - 1.0) < 1e-9 | |
| def test_zero_pinning(self): | |
| axes = {a: 0.9 for a in pg.CANONICAL_AXES} | |
| axes["soundness"] = 0.0 | |
| result = pg._compute_lambda(axes) | |
| assert result["lambda"] == 0.0 | |
| assert result["zero_pinned"] is True | |
| def test_monotonicity(self): | |
| base = {a: 0.80 for a in pg.CANONICAL_AXES} | |
| high = {**base, "calibration": 0.99} | |
| r_base = pg._compute_lambda(base) | |
| r_high = pg._compute_lambda(high) | |
| assert r_high["lambda"] >= r_base["lambda"] | |
| def test_symmetry(self): | |
| """A4 corollary: all-equal axes → Λ = that value (geometric = arithmetic for equal values).""" | |
| axes = {a: 0.90 for a in pg.CANONICAL_AXES} | |
| result = pg._compute_lambda(axes) | |
| assert abs(result["lambda"] - 0.90) < 1e-6 | |
| def test_weights_normalised(self): | |
| """Custom weights that don't sum to 1 should still give valid Λ ∈ [0,1].""" | |
| axes = {a: 0.90 for a in pg.CANONICAL_AXES} | |
| weights = {a: 2.0 for a in pg.CANONICAL_AXES} # sum = 18, not 1 | |
| result = pg._compute_lambda(axes, weights) | |
| assert 0.0 <= result["lambda"] <= 1.0 + 1e-9 | |