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"login", 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()