stage-topo-b topology-export: /federation/topology endpoint + CORS cache

tests/test_topology_export.py

@@ -1,24 +1,25 @@
-"""Topology export — whitelist sanitization unit tests.
+"""Topology export — whitelist sanitization + endpoint contract.
 
 The big invariant: nothing from docker_view.topology() escapes that isn't
 in the Pydantic schema. We assert via model_fields introspection AND via a
 JSON-dump scan over a fixture that contains every dangerous field.
-
-The /federation/topology endpoint contract lives in the sibling tests added
-alongside it; this module covers the builder + the sanitizer in isolation.
 """
 
 from __future__ import annotations
 
 import json
-from typing import Any, Dict
+from typing import Any, Dict, List
 from unittest.mock import patch
 
 import pytest
+from fastapi import FastAPI
+from fastapi.templating import Jinja2Templates
+from fastapi.testclient import TestClient
 from sqlalchemy import create_engine
+from starlette.middleware.sessions import SessionMiddleware
 
 from psyc import db
-from psyc.cockpit import docker_view
+from psyc.cockpit import docker_view, federation_routes
 from psyc.lines import federation, topology_export
 from psyc.lines.topology_export import (
     TopologyContainer,
@@ -50,6 +51,14 @@ def fed_dir(tmp_path, monkeypatch):
     yield d
 
 
+@pytest.fixture(autouse=True)
+def reset_topology_cache():
+    if hasattr(federation_routes, "_TOPOLOGY_CACHE"):
+        federation_routes._TOPOLOGY_CACHE["payload"] = None
+        federation_routes._TOPOLOGY_CACHE["ts"] = 0.0
+    yield
+
+
 # ---------- fixture data: hostile docker_view output --------------------
 
 # This payload has every leaky field docker_view *could* surface, plus
@@ -243,3 +252,62 @@ def test_build_export_caps_at_max_containers(fresh_db, fed_dir, monkeypatch):
     monkeypatch.setattr(docker_view, "topology", lambda: fake)
     out = build_export()
     assert out.container_count == topology_export.MAX_CONTAINERS
+
+
+# ---------- HTTP endpoint -----------------------------------------------
+
+def _mk_app() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(SessionMiddleware, secret_key="test-secret")
+    from pathlib import Path as _Path
+    here = _Path(__file__).resolve().parent.parent / "src" / "psyc" / "cockpit" / "templates"
+    templates = Jinja2Templates(directory=str(here))
+    federation_routes.register(app, templates)
+    return app
+
+
+def test_federation_topology_endpoint_returns_json_with_cors(fresh_db, fed_dir, monkeypatch):
+    monkeypatch.setattr(docker_view, "topology", lambda: _LEAKY_TOPOLOGY)
+    client = TestClient(_mk_app())
+    r = client.get("/federation/topology")
+    assert r.status_code == 200
+    assert r.headers.get("access-control-allow-origin") == "*"
+    data = r.json()
+    # Schema check.
+    for key in ("node_fingerprint", "generated_at", "host_name",
+                "container_count", "network_count", "containers", "networks"):
+        assert key in data
+    assert data["container_count"] == 2
+    assert len(data["containers"]) == 2
+    # No leaks in the wire response either.
+    blob = r.text
+    for forbidden in _FORBIDDEN_STRINGS:
+        assert forbidden not in blob, f"leak via endpoint: {forbidden!r}"
+
+
+def test_federation_topology_endpoint_resilient_when_docker_unavailable(fresh_db, fed_dir, monkeypatch):
+    def boom():
+        raise docker_view.DockerProxyError("proxy down")
+    monkeypatch.setattr(docker_view, "topology", boom)
+    client = TestClient(_mk_app())
+    r = client.get("/federation/topology")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["container_count"] == 0
+    assert data["containers"] == []
+
+
+def test_federation_topology_cache_short_circuits_repeated_calls(fresh_db, fed_dir, monkeypatch):
+    """Within TTL, a second hit must not re-call docker_view."""
+    calls = {"n": 0}
+
+    def counted():
+        calls["n"] += 1
+        return _LEAKY_TOPOLOGY
+
+    monkeypatch.setattr(docker_view, "topology", counted)
+    client = TestClient(_mk_app())
+    r1 = client.get("/federation/topology")
+    r2 = client.get("/federation/topology")
+    assert r1.status_code == 200 and r2.status_code == 200
+    assert calls["n"] == 1, "cache should suppress the second docker_view call"