Use explicit fresh hosted capability snapshots for plan checks

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

tests/test_hosted_plan_check.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import urllib.error
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from reuse_surface.cli import main
+from reuse_surface.plan_check import load_query_from_intent, run_plan_check
+from reuse_surface.plan_snapshot import MAX_SNAPSHOT_BYTES, SnapshotUnavailable
+
+
+def snapshot():
+    return {
+        "updated": datetime.now(timezone.utc).date().isoformat(),
+        "composed_at": datetime.now(timezone.utc).isoformat(),
+        "stale": False,
+        "capabilities": [{"id": "capability.hosted.only", "name": "Tender evidence",
+                          "summary": "Tender evidence", "tags": ["tender"]}],
+    }
+
+
+def serve(monkeypatch, payload, *, warnings=None, raw=None):
+    body = raw if raw is not None else json.dumps(payload).encode()
+
+    def urlopen(request, *, timeout):
+        assert request.full_url == "https://registry.example/v1/federated"
+        assert request.get_method() == "GET"
+        assert request.data is None
+        assert not request.has_header("Authorization")
+        assert timeout == 30
+        response = io.BytesIO(body)
+        response.status = 200
+        response.headers = {"X-Federation-Warnings": warnings} if warnings else {}
+        return response
+
+    monkeypatch.setattr("urllib.request.urlopen", urlopen)
+    return body
+
+
+def check(**kwargs):
+    return run_plan_check(load_query_from_intent("Tender evidence"),
+                          hub_url="https://registry.example", use_llm=False, **kwargs)
+
+
+def test_hosted_only_capability_and_provenance_without_local_fallback(monkeypatch):
+    raw = serve(monkeypatch, snapshot())
+    monkeypatch.setenv("REUSE_SURFACE_TOKEN", "must-never-be-sent")
+    monkeypatch.setattr("reuse_surface.plan_check.load_federated_capabilities",
+                        lambda: pytest.fail("local source must not be consulted"))
+    result = check()
+    assert result["verdict"] == "reuse"
+    assert result["matches"][0]["id"] == "capability.hosted.only"
+    assert result["index_source"]["sha256"] == hashlib.sha256(raw).hexdigest()
+    assert result["index_source"]["capability_count"] == 1
+
+
+@pytest.mark.parametrize("change", [
+    {"stale": True}, {"stale": None}, {"composed_at": None},
+    {"composed_at": "2020-01-01T00:00:00Z"},
+    {"composed_at": "2026-01-01T00:00:00"}, {"updated": "bad-date"},
+    {"capabilities": []}, {"capabilities": [None]}, {"capabilities": {}},
+    {"capabilities": [{"id": "bad", "name": "bad", "summary": "bad", "tags": None}]},
+])
+def test_invalid_source_never_produces_a_verdict(monkeypatch, change):
+    serve(monkeypatch, snapshot() | change)
+    with pytest.raises(SnapshotUnavailable):
+        check()
+
+
+def test_partial_federation_refuses_before_verdict_and_side_effects(monkeypatch, capsys):
+    serve(monkeypatch, snapshot(), warnings="one source unavailable")
+    monkeypatch.setattr("reuse_surface.cli.record_outcome",
+                        lambda *a, **k: pytest.fail("must not record a failed check"))
+    monkeypatch.setattr("reuse_surface.cli.maybe_file_capability_request",
+                        lambda *a, **k: pytest.fail("must not create demand on source failure"))
+    assert main(["plan-check", "--intent", "unrelated", "--hub-url",
+                 "https://registry.example", "--no-llm", "--format", "json",
+                 "--record-outcome", "new", "--file-request"]) == 2
+    output = capsys.readouterr()
+    assert output.out == ""
+    assert "federation warnings" in output.err
+
+
+@pytest.mark.parametrize("error", [
+    urllib.error.URLError("unreachable"), TimeoutError(),
+    urllib.error.HTTPError("https://registry.example", 503, "down", {}, None),
+])
+def test_transport_failure_is_not_an_empty_index(monkeypatch, error):
+    def fail(*args, **kwargs):
+        raise error
+    monkeypatch.setattr("urllib.request.urlopen", fail)
+    with pytest.raises(SnapshotUnavailable):
+        check()
+
+
+@pytest.mark.parametrize("raw", [b"<html>login</html>", b"[]", b"x" * (MAX_SNAPSHOT_BYTES + 1)])
+def test_invalid_or_oversized_response(monkeypatch, raw):
+    serve(monkeypatch, None, raw=raw)
+    with pytest.raises(SnapshotUnavailable):
+        check()
+
+
+@pytest.mark.parametrize("age", [-2, 0, float("nan"), float("inf")])
+def test_invalid_age_limit(monkeypatch, age):
+    monkeypatch.setattr("urllib.request.urlopen", lambda *a, **k: pytest.fail("invalid input"))
+    with pytest.raises(SnapshotUnavailable):
+        check(max_index_age_hours=age)
+
+
+def test_future_and_over_age_composition(monkeypatch):
+    for delta in (timedelta(hours=1), timedelta(hours=-25)):
+        serve(monkeypatch, snapshot() | {"composed_at": (datetime.now(timezone.utc) + delta).isoformat()})
+        with pytest.raises(SnapshotUnavailable):
+            check()
+
+
+def test_duplicate_ids_refused(monkeypatch):
+    payload = snapshot()
+    payload["capabilities"] *= 2
+    serve(monkeypatch, payload)
+    with pytest.raises(SnapshotUnavailable):
+        check()