from __future__ import annotations from dataclasses import dataclass import httpx import psycopg import pytest from fastapi import FastAPI from app.api.v1.demo import ( DemoCandidate, DemoCounts, DemoRepository, get_demo_repository, make_citation_id, make_page_label, router, ) @dataclass class StubRepository: count_result: DemoCounts candidates: list[DemoCandidate] failure: psycopg.Error | None = None def counts(self) -> DemoCounts: if self.failure is not None: raise self.failure return self.count_result def search(self, query_vector: tuple[float, ...], *, limit: int) -> list[DemoCandidate]: if self.failure is not None: raise self.failure assert len(query_vector) == 1024 return self.candidates[:limit] def create_app(repository: DemoRepository) -> FastAPI: app = FastAPI() app.include_router(router) app.dependency_overrides[get_demo_repository] = lambda: repository return app @pytest.mark.asyncio async def test_status_reports_only_public_synthetic_counts() -> None: app = create_app( StubRepository( count_result=DemoCounts(chunks=20, vectors=20, searchable=20), candidates=[], ) ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.get("/api/v1/demo/status") assert response.status_code == 200 assert response.json() == { "status": "ready", "dataset": "synthetic-demo", "counts": {"chunks": 20, "vectors": 20, "searchable": 20}, } @pytest.mark.asyncio async def test_status_explicitly_reports_empty_dataset() -> None: app = create_app( StubRepository( count_result=DemoCounts(chunks=0, vectors=0, searchable=0), candidates=[], ) ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.get("/api/v1/demo/status") assert response.status_code == 200 assert response.json()["status"] == "empty_dataset" @pytest.mark.asyncio async def test_status_does_not_mark_a_partial_demo_index_ready() -> None: app = create_app( StubRepository( count_result=DemoCounts(chunks=20, vectors=19, searchable=18), candidates=[], ) ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.get("/api/v1/demo/status") assert response.status_code == 200 assert response.json()["status"] == "incomplete_dataset" @pytest.mark.asyncio async def test_search_embeds_reranks_and_returns_only_safe_fields() -> None: repository = StubRepository( count_result=DemoCounts(chunks=2, vectors=2, searchable=2), candidates=[ DemoCandidate( source_key="internal-chunk-one", title="西岭铜矿化合成记录", text="该合成记录描述花岗斑岩接触带，未包含真实矿权或人员信息。", page_start=3, page_end=3, ), DemoCandidate( source_key="internal-chunk-two", title="北谷金异常合成记录", text="北谷合成样点出现低强度金异常，仅用于离线验证。", page_start=7, page_end=8, ), ], ) app = create_app(repository) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.post( "/api/v1/demo/search", json={"query": "花岗斑岩 铜矿化", "top_k": 1}, ) assert response.status_code == 200 payload = response.json() assert payload["status"] == "ok" assert payload["dataset"] == "synthetic-demo" assert len(payload["results"]) == 1 result = payload["results"][0] assert set(result) == {"title", "snippet", "page_label", "score", "citation_id"} assert result["title"].startswith("合成资料｜") assert result["page_label"] == "第 3 页" assert result["citation_id"] == make_citation_id("internal-chunk-one") assert "internal-chunk-one" not in str(payload) assert "manifest" not in str(payload).lower() assert "sha256" not in str(payload).lower() @pytest.mark.asyncio async def test_search_returns_explicit_empty_dataset_state() -> None: app = create_app( StubRepository( count_result=DemoCounts(chunks=0, vectors=0, searchable=0), candidates=[], ) ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.post( "/api/v1/demo/search", json={"query": "不存在的合成资料", "top_k": 3}, ) assert response.status_code == 200 assert response.json() == { "status": "empty_dataset", "dataset": "synthetic-demo", "results": [], } @pytest.mark.asyncio @pytest.mark.parametrize( "body", [ {"query": " ", "top_k": 3}, {"query": "x" * 501, "top_k": 3}, {"query": "铜矿", "top_k": 0}, {"query": "铜矿", "top_k": 11}, ], ) async def test_search_rejects_invalid_bounded_input(body: dict[str, object]) -> None: app = create_app( StubRepository( count_result=DemoCounts(chunks=0, vectors=0, searchable=0), candidates=[], ) ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.post("/api/v1/demo/search", json=body) assert response.status_code == 422 @pytest.mark.asyncio @pytest.mark.parametrize("path", ["/api/v1/demo/status", "/api/v1/demo/search"]) async def test_database_failure_is_sanitized_as_503(path: str) -> None: app = create_app( StubRepository( count_result=DemoCounts(chunks=0, vectors=0, searchable=0), candidates=[], failure=psycopg.OperationalError("host=db password=must-not-leak"), ) ) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: if path.endswith("search"): response = await client.post(path, json={"query": "铜矿", "top_k": 3}) else: response = await client.get(path) assert response.status_code == 503 assert response.json() == {"detail": "database unavailable"} assert "must-not-leak" not in response.text def test_citation_and_page_labels_are_stable_and_opaque() -> None: first = make_citation_id("private-database-uuid") assert first == make_citation_id("private-database-uuid") assert first != make_citation_id("another-private-database-uuid") assert "private" not in first assert make_page_label(None, None) == "页码未知" assert make_page_label(2, 4) == "第 2-4 页"