test/test_holographic_zero_cdn.py

"""Zero-CDN and honest same-origin evidence contract regressions."""
from __future__ import annotations

import hashlib
import unittest
from pathlib import Path
from unittest.mock import patch

from fastapi import FastAPI
from fastapi.testclient import TestClient

import szl3d_holographic as holographic


class HolographicZeroCdnTests(unittest.TestCase):
    def _json(self, url: str):
        if "api/models" in url:
            return [{"downloads": 3}, {"downloads": 7}]
        if "api/datasets" in url:
            return [{"downloads": 11}]
        if "api/spaces" in url:
            return [{"likes": 5}]
        if "api/collections" in url:
            return [{"slug": "governed-estate"}]
        if "api.github.com" in url:
            return [{"stargazers_count": 2}]
        raise AssertionError(f"unexpected test URL: {url}")

    @staticmethod
    def _health(_url: str):
        return {"read_state": "LIVE", "health_state": "ok", "up": True}

    def test_authored_3d_tree_has_no_external_runtime_fetch(self):
        violations = list(holographic.no_cdn_violations(holographic._base_dir()))
        self.assertEqual(violations, [])
        brain = (Path(__file__).parents[1] / "static" / "3d" / "brain.html").read_text(encoding="utf-8")
        self.assertIn('fetch(BRAIN_EVIDENCE', brain)
        self.assertIn('const BRAIN_EVIDENCE = "/api/a11oy/v1/holographic/brain/evidence"', brain)
        self.assertNotIn('fetch("https://', brain)
        self.assertNotIn("fetch('https://", brain)

    def test_complete_observation_is_live_and_measured(self):
        formula_source = "theorem one : True := by trivial\n  theorem two : True := by trivial\n"
        payload = holographic.brain_evidence(
            fetch_json=self._json,
            fetch_text=lambda _url: formula_source,
            fetch_health=self._health,
            expected_formula_sha256=hashlib.sha256(formula_source.encode("utf-8")).hexdigest(),
        )
        self.assertEqual(payload["state"], "LIVE")
        self.assertEqual(payload["live_lobes"], payload["lobe_count"])
        self.assertEqual(payload["lobes"]["formulas"]["theorem_count"], 2)
        self.assertEqual(payload["lobes"]["formulas"]["integrity_state"], "VERIFIED")
        self.assertFalse(payload["lobes"]["formulas"]["mutable_reference"])
        self.assertEqual(payload["lobes"]["models"]["downloads_sum"], 10)
        self.assertEqual(payload["lobes"]["fleet"]["up_count"], 5)
        self.assertFalse(payload["limits"]["cache_or_last_good_fallback"])
        self.assertTrue(payload["limits"]["fixed_allowlist"])

    def test_failed_composite_source_is_not_hidden_by_partial_count(self):
        def partial_json(url: str):
            if "author=zai-org" in url:
                raise TimeoutError("test timeout")
            return self._json(url)

        formula_source = "theorem one : True := by trivial\n"
        payload = holographic.brain_evidence(
            fetch_json=partial_json,
            fetch_text=lambda _url: formula_source,
            fetch_health=self._health,
            expected_formula_sha256=hashlib.sha256(formula_source.encode("utf-8")).hexdigest(),
        )
        frontier = payload["lobes"]["frontier"]
        self.assertEqual(payload["state"], "DEGRADED")
        self.assertEqual(frontier["state"], "UNAVAILABLE")
        self.assertIsNone(frontier["count"])
        failed = next(row for row in frontier["sources"] if row["org"] == "zai-org")
        self.assertEqual(failed["reason"], "upstream_unreachable_or_timeout")

    def test_formula_source_is_commit_pinned_and_hash_allowlisted(self):
        url = holographic._BRAIN_SOURCES["formulas"]
        self.assertIn(holographic._BRAIN_FORMULA_COMMIT, url)
        self.assertNotIn("/main/", url)
        self.assertEqual(len(holographic._BRAIN_FORMULA_COMMIT), 40)
        self.assertEqual(len(holographic._BRAIN_FORMULA_SHA256), 64)

    def test_formula_integrity_is_persistently_visible_not_hover_only(self):
        brain = (Path(__file__).parents[1] / "static" / "3d" / "brain.html").read_text(encoding="utf-8")
        self.assertIn('proof.id = "formula-attestation"', brain)
        self.assertIn('proof.setAttribute("aria-live", "polite")', brain)
        self.assertIn('s.source_commit', brain)
        self.assertIn('s.content_sha256', brain)
        self.assertIn('s.integrity_state', brain)
        self.assertIn('setFormulaAttestation(s.attestation, "VERIFIED")', brain)
        self.assertIn('integrity UNAVAILABLE · no theorem count trusted', brain)

    def test_formula_hash_mismatch_is_unavailable_not_measured(self):
        payload = holographic.brain_evidence(
            fetch_json=self._json,
            fetch_text=lambda _url: "theorem tampered : True := by trivial\n",
            fetch_health=self._health,
        )
        formula = payload["lobes"]["formulas"]
        self.assertEqual(payload["state"], "DEGRADED")
        self.assertEqual(formula["state"], "UNAVAILABLE")
        self.assertEqual(formula["reason"], "source_integrity_mismatch")
        self.assertNotIn("theorem_count", formula)

    def test_route_is_get_only_no_store_and_same_contract(self):
        app = FastAPI()
        expected = {
            "schema": "a11oy.holographic.brain-evidence.v1",
            "state": "UNAVAILABLE",
            "lobes": {},
        }
        with patch.object(holographic, "brain_evidence", return_value=expected):
            holographic.register(app, ns="a11oy")
            client = TestClient(app)
            response = client.get("/api/a11oy/v1/holographic/brain/evidence")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json(), expected)
        self.assertEqual(response.headers["cache-control"], "no-store")
        methods = {
            method
            for route in app.routes
            if getattr(route, "path", None) == "/api/a11oy/v1/holographic/brain/evidence"
            for method in route.methods
        }
        self.assertEqual(methods, {"GET"})


if __name__ == "__main__":
    unittest.main()