"""v0.19.0 / roadmap item #30 — `/api/docs/sessions/*` endpoints. The framework mediates reads against the public `wiggleverse/ohm-session-history` gitea repo so the rendered `/docs/sessions/*` surface inherits the same chrome as `/docs/user-guide`. This test suite covers the four endpoints + the in-process TTL cache, mocking the upstream HTTP via `httpx.MockTransport` (the same shape the rest of the test suite uses for Gitea). The tests do NOT spin up the full FakeGitea — they only need to mock the gitea raw URL surface (and the contents API for the session-index endpoint). Each test owns its mock transport so we can dial in 200 / 404 / 5xx / timeout responses per case. Path-validation tests intentionally bypass the network — a malformed `nnnn` or `filename` MUST be rejected at the route layer before any upstream call is made. """ from __future__ import annotations import json import httpx import pytest from fastapi.testclient import TestClient from app import docs_sessions # Reuse the proven app-construction fixtures from the proposal vertical # (same shape every test file in this repo uses). from test_propose_vertical import ( # noqa: F401 app_with_fake_gitea, tmp_env, ) # --------------------------------------------------------------------------- # Test scaffolding # --------------------------------------------------------------------------- class _UpstreamHandler: """Records every URL the docs_sessions module fetched and returns canned responses keyed by URL substring. Allows the test to assert on call count (for cache verification) without needing a full Gitea simulator. `calls` tracks only URLs that hit the session-history host (the `OHM_SESSION_HISTORY_*` bases) so reconciler/Gitea-side calls — which also pass through this handler because `httpx.AsyncClient` is a shared attribute the gitea-side fixture also monkeypatches — don't inflate the count we use for cache-hit assertions. """ _SESSION_HOST_MARKERS = ("ohm-session-history", "wiggleverse/ohm-session-history") def __init__(self, responses: dict[str, tuple[int, str]]): self.responses = responses self.calls: list[str] = [] def __call__(self, request: httpx.Request) -> httpx.Response: url = str(request.url) if any(m in url for m in self._SESSION_HOST_MARKERS): self.calls.append(url) for key, (status, body) in self.responses.items(): if key in url: return httpx.Response(status, text=body) # Default: 404. Lets tests skip declaring "the rest is 404". return httpx.Response(404, text="not found") @pytest.fixture def patched_httpx(monkeypatch): """Provide a hook the test can call to install a MockTransport. Returns a closure: `install(handler)` patches `app.docs_sessions.httpx.AsyncClient` so every constructed client uses the handler's transport. NB: the upstream `app_with_fake_gitea` fixture also patches `httpx.AsyncClient` (to route gitea calls to a FakeGitea handler), and because `httpx` is a single shared module, that patch mutates the *same* `AsyncClient` attribute we're about to overwrite. We therefore import the unpatched class directly from the `httpx._client` module so our install path can construct a fresh real client around our MockTransport without going through the FakeGitea wrapper. """ from httpx._client import AsyncClient as RealAsyncClient def install(handler): def patched(*args, **kwargs): kwargs["transport"] = httpx.MockTransport(handler) return RealAsyncClient(*args, **kwargs) monkeypatch.setattr("app.docs_sessions.httpx.AsyncClient", patched) return handler yield install @pytest.fixture def app(app_with_fake_gitea): """Wrap the shared app fixture, resetting the docs-sessions cache so cross-test state can't leak. Returns just the FastAPI app — the fake-Gitea handle is irrelevant for the docs-sessions surface. """ docs_sessions.reset_cache() fastapi_app, _fake = app_with_fake_gitea return fastapi_app # --------------------------------------------------------------------------- # Manifest endpoint # --------------------------------------------------------------------------- def test_manifest_happy_path(app, patched_httpx): manifest_body = json.dumps( { "0001": {"title": "Bootstrap"}, "0014": {"title": "Wave 7 driver"}, } ) patched_httpx( _UpstreamHandler({"sessions.json": (200, manifest_body)}) ) with TestClient(app) as client: r = client.get("/api/docs/sessions/manifest") assert r.status_code == 200, r.text payload = r.json() assert payload == { "0001": {"title": "Bootstrap"}, "0014": {"title": "Wave 7 driver"}, } def test_manifest_empty_state(app, patched_httpx): """A 404 from gitea means the manifest hasn't been published yet. The endpoint returns HTTP 200 + `{}` so the frontend can render the no-sessions-yet state without an error banner. """ patched_httpx(_UpstreamHandler({"sessions.json": (404, "not found")})) with TestClient(app) as client: r = client.get("/api/docs/sessions/manifest") assert r.status_code == 200, r.text assert r.json() == {} def test_manifest_upstream_5xx_returns_502(app, patched_httpx): patched_httpx(_UpstreamHandler({"sessions.json": (500, "internal")})) with TestClient(app) as client: r = client.get("/api/docs/sessions/manifest") assert r.status_code == 502, r.text body = r.json() assert body["detail"]["error"] == "session-history fetch failed" # --------------------------------------------------------------------------- # About endpoint # --------------------------------------------------------------------------- def test_about_happy_path(app, patched_httpx): readme = "# OHM session history\n\nWelcome.\n" patched_httpx(_UpstreamHandler({"README.md": (200, readme)})) with TestClient(app) as client: r = client.get("/api/docs/sessions/about") assert r.status_code == 200, r.text assert "text/markdown" in r.headers["content-type"] assert r.text == readme def test_about_404(app, patched_httpx): patched_httpx(_UpstreamHandler({"README.md": (404, "")})) with TestClient(app) as client: r = client.get("/api/docs/sessions/about") assert r.status_code == 404, r.text def test_about_upstream_5xx_returns_502(app, patched_httpx): patched_httpx(_UpstreamHandler({"README.md": (503, "down")})) with TestClient(app) as client: r = client.get("/api/docs/sessions/about") assert r.status_code == 502, r.text # --------------------------------------------------------------------------- # Transcript endpoint # --------------------------------------------------------------------------- def test_transcript_happy_path(app, patched_httpx): body = "# Session 0017.1 — Transcript\n\nbody.\n" fname = "SESSION-0017.1-TRANSCRIPT-2026-05-28T08-50--2026-05-28T11-20.md" patched_httpx(_UpstreamHandler({fname: (200, body)})) with TestClient(app) as client: r = client.get(f"/api/docs/sessions/0017/{fname}") assert r.status_code == 200, r.text assert "text/markdown" in r.headers["content-type"] assert r.text == body def test_transcript_404(app, patched_httpx): fname = "SESSION-9999.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md" patched_httpx(_UpstreamHandler({})) # everything 404s with TestClient(app) as client: r = client.get(f"/api/docs/sessions/9999/{fname}") assert r.status_code == 404, r.text def test_transcript_rejects_invalid_session_dir(app, patched_httpx): """`nnnn` must be exactly 4 digits. `abcd` fails before any network call. """ handler = _UpstreamHandler({}) patched_httpx(handler) with TestClient(app) as client: r = client.get( "/api/docs/sessions/abcd/" "SESSION-0001.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md" ) assert r.status_code == 400, r.text assert handler.calls == [], "rejected path must not hit the network" def test_transcript_rejects_path_traversal(app, patched_httpx): """A filename that doesn't match the SESSION-NNNN.M-TRANSCRIPT regex is rejected. `../etc/passwd` doesn't match; neither does the legacy flat-root `SESSION-A-TRANSCRIPT.md`. """ handler = _UpstreamHandler({}) patched_httpx(handler) with TestClient(app) as client: # Path traversal — but FastAPI normalizes `..` in the path before # routing, so this resolves to /api/docs/sessions/0001/etc/passwd # which routes to the same handler with filename=etc/passwd, and # gets rejected as an invalid transcript filename. Even if the # normalization didn't apply (some intermediary), the regex # check rejects anything not matching the SESSION- prefix. r = client.get( "/api/docs/sessions/0001/etc%2Fpasswd" ) # 400 (filename validation) or 404 (path didn't match the # route); both reject before any network call. Either is # acceptable — what matters is that we never fetched it. assert r.status_code in (400, 404), r.text assert handler.calls == [], "rejected path must not hit the network" def test_transcript_rejects_legacy_flat_filename(app, patched_httpx): """Legacy `SESSION-A-TRANSCRIPT.md` (letter form) doesn't match the numeric regex — by design, since post-#23 transcripts live in `NNNN/` folders with numeric names. Reject 400. """ handler = _UpstreamHandler({}) patched_httpx(handler) with TestClient(app) as client: r = client.get("/api/docs/sessions/0001/SESSION-A-TRANSCRIPT.md") assert r.status_code == 400, r.text assert handler.calls == [], "rejected path must not hit the network" def test_transcript_upstream_5xx_returns_502(app, patched_httpx): fname = "SESSION-0001.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md" patched_httpx(_UpstreamHandler({fname: (502, "bad gateway")})) with TestClient(app) as client: r = client.get(f"/api/docs/sessions/0001/{fname}") assert r.status_code == 502, r.text # --------------------------------------------------------------------------- # Session-index endpoint # --------------------------------------------------------------------------- def test_session_index_happy_path(app, patched_httpx): """The contents API returns a JSON list of file entries. The endpoint filters to entries that match the transcript regex and sorts them. """ # Two transcripts (driver + subagent) + a non-transcript sibling # that must be filtered out. listing = json.dumps( [ { "name": "SESSION-0017.0-TRANSCRIPT-" "2026-05-28T08-30--2026-05-28T12-00.md", "type": "file", }, { "name": "SESSION-0017.1-TRANSCRIPT-" "2026-05-28T08-50--2026-05-28T11-20.md", "type": "file", }, {"name": "notes.md", "type": "file"}, # not a transcript {"name": "attached-dir", "type": "dir"}, # not a file ] ) patched_httpx(_UpstreamHandler({"/contents/0017": (200, listing)})) with TestClient(app) as client: r = client.get("/api/docs/sessions/0017/index") assert r.status_code == 200, r.text files = r.json()["files"] assert files == [ "SESSION-0017.0-TRANSCRIPT-2026-05-28T08-30--2026-05-28T12-00.md", "SESSION-0017.1-TRANSCRIPT-2026-05-28T08-50--2026-05-28T11-20.md", ] def test_session_index_404(app, patched_httpx): patched_httpx(_UpstreamHandler({})) # everything 404s with TestClient(app) as client: r = client.get("/api/docs/sessions/9999/index") assert r.status_code == 404, r.text def test_session_index_rejects_invalid_session_dir(app, patched_httpx): handler = _UpstreamHandler({}) patched_httpx(handler) with TestClient(app) as client: r = client.get("/api/docs/sessions/abc/index") assert r.status_code == 400, r.text assert handler.calls == [], "rejected path must not hit the network" # --------------------------------------------------------------------------- # Cache behavior # --------------------------------------------------------------------------- def test_manifest_cache_hits_within_ttl(app, patched_httpx, monkeypatch): """Two consecutive manifest calls within the TTL window should issue exactly one HTTP request to gitea. """ # Generous TTL so the test never races. monkeypatch.setenv("OHM_DOCS_SESSIONS_MANIFEST_TTL_SEC", "60") handler = _UpstreamHandler( {"sessions.json": (200, json.dumps({"0001": {"title": "x"}}))} ) patched_httpx(handler) with TestClient(app) as client: r1 = client.get("/api/docs/sessions/manifest") r2 = client.get("/api/docs/sessions/manifest") assert r1.status_code == 200 assert r2.status_code == 200 assert len(handler.calls) == 1, ( f"expected one upstream call, got {handler.calls}" ) def test_transcript_cache_hits_within_ttl(app, patched_httpx, monkeypatch): monkeypatch.setenv("OHM_DOCS_SESSIONS_CONTENT_TTL_SEC", "300") fname = "SESSION-0001.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md" handler = _UpstreamHandler({fname: (200, "# body\n")}) patched_httpx(handler) with TestClient(app) as client: r1 = client.get(f"/api/docs/sessions/0001/{fname}") r2 = client.get(f"/api/docs/sessions/0001/{fname}") assert r1.status_code == 200 assert r2.status_code == 200 assert len(handler.calls) == 1 def test_transcript_404_is_cached(app, patched_httpx, monkeypatch): """Negative caching: a 404 result is cached at the content TTL so a deployment with no published transcripts doesn't hammer gitea on every navigation. Documented in `docs_sessions.fetch_transcript`. """ monkeypatch.setenv("OHM_DOCS_SESSIONS_CONTENT_TTL_SEC", "300") fname = "SESSION-9999.0-TRANSCRIPT-2026-01-01T00-00--2026-01-01T00-01.md" handler = _UpstreamHandler({}) # everything 404s patched_httpx(handler) with TestClient(app) as client: r1 = client.get(f"/api/docs/sessions/9999/{fname}") r2 = client.get(f"/api/docs/sessions/9999/{fname}") assert r1.status_code == 404 assert r2.status_code == 404 assert len(handler.calls) == 1, "negative caching should suppress the 2nd call"