from __future__ import annotations from pathlib import Path from uuid import uuid4 from fastapi.testclient import TestClient from acquisition.domain import Query, QueryBatch from app.api import app from app.dependencies import ( get_acquisition_repository, get_decode_repository, get_pipeline_repository, ) from decode_content.models import DecodeJob, DecodeResult, PayloadDraft from pipeline.models import PipelineRun class FakeAcquisitionRepo: def __init__(self): self.batch_id = uuid4() self.run_id = uuid4() self.query_id = uuid4() self.item_id = uuid4() self.media_id = uuid4() self.classification_id = uuid4() def get_query_batch(self, batch_id): assert batch_id == self.batch_id return QueryBatch( id=batch_id, name="正式 query 批次", source_type="manual", target_platforms=["xiaohongshu", "weixin", "douyin"], status="ready", ) def list_queries_for_batch(self, batch_id, *, keep=None): assert batch_id == self.batch_id return [ Query( id=self.query_id, batch_id=batch_id, query_text="短视频脚本 开头 怎么写", keep=True, status="ready", ) ] def get_run_summary(self, run_id): assert run_id == self.run_id return { "id": run_id, "run_key": "run-1", "batch_id": self.batch_id, "status": "running", "query_count": 1, "job_count": 3, "candidate_count": 5, "creation_hit_count": 2, "queries": [ { "query_id": self.query_id, "query_text": "短视频脚本 开头 怎么写", "candidate_count": 5, "creation_hit_count": 2, "platforms": {"xiaohongshu": {"status": "done"}}, } ], } def get_query_detail(self, *, run_id, query_id): assert run_id == self.run_id assert query_id == self.query_id return { "query": { "id": query_id, "batch_id": self.batch_id, "query_text": "短视频脚本 开头 怎么写", "axes": {}, "keep": True, "filter_reason": None, "status": "ready", "sort_order": 0, }, "jobs": [{"id": uuid4(), "platform": "xiaohongshu", "status": "done"}], "items": [ { "id": self.item_id, "platform": "xiaohongshu", "platform_item_id": "xhs1", "canonical_url": "https://xhs.test/1", "content_type": "图文", "title": "脚本开头", "author_name": "作者", "raw_summary": "先定受众", "status": "candidate", } ], "media_assets": [ { "id": self.media_id, "item_id": self.item_id, "media_type": "image", "source_url": "https://origin.test/1.jpg", "oss_url": "https://oss.test/1.jpg", "cdn_url": "https://cdn.test/1.jpg", "position": 1, "status": "done", } ], "classifications": [ { "id": self.classification_id, "item_id": self.item_id, "is_creation_knowledge": True, "label": "creation", "confidence": 0.98, "reason": "可复用", "status": "done", } ], } class FakeDecodeRepo: def __init__(self): self.decode_result_id = uuid4() self.decode_job_id = uuid4() self.payload_id = uuid4() def get_decode_result_for_item(self, item_id): return DecodeResult( id=self.decode_result_id, item_id=item_id, status="decoded", read_result={"text": "读懂后的帖子内容"}, gate_result={"passed": True, "reason": "具备可复用创作方法"}, framing_result={"particles": []}, ) def create_decode_job(self, *, item_id, status="pending", metadata=None): return DecodeJob( id=self.decode_job_id, item_id=item_id, status=status, metadata=metadata or {}, ) def list_payload_drafts(self, item_id=None): return [ PayloadDraft( id=self.payload_id, item_id=item_id or uuid4(), payload={"title": "可审核 payload"}, review_status="pending", ingest_ready=False, status="draft", ) ] class FakePipelineRepo: def get_pipeline_run(self, run_id): return PipelineRun( id=run_id, run_key="pipeline-run-1", status="running", current_stage="decode", ) def _client(repo: FakeAcquisitionRepo): app.dependency_overrides[get_acquisition_repository] = lambda: repo return TestClient(app) def test_app_health_and_formal_acquisition_routes(): repo = FakeAcquisitionRepo() client = _client(repo) try: assert client.get("/health").json()["entrypoint"] == "app.api:app" batch = client.get(f"/api/query-batches/{repo.batch_id}").json() assert batch["batch"]["name"] == "正式 query 批次" assert batch["queries"][0]["id"] == str(repo.query_id) summary = client.get(f"/api/acquisition/runs/{repo.run_id}/summary").json() assert summary["creation_hit_count"] == 2 assert summary["queries"][0]["query_id"] == str(repo.query_id) detail = client.get(f"/api/acquisition/runs/{repo.run_id}/queries/{repo.query_id}").json() item = detail["platforms"]["xiaohongshu"]["items"][0] assert item["media_assets"][0]["cdn_url"] == "https://cdn.test/1.jpg" assert item["classification"]["is_creation_knowledge"] is True finally: app.dependency_overrides.clear() def test_formal_app_does_not_expose_legacy_static_mounts(): paths = {route.path for route in app.routes if hasattr(route, "path")} assert "/data" not in paths assert "/frames" not in paths assert "/api/creation-search/summary" not in paths assert "/api/creation-search/query" not in paths def test_unimplemented_formal_routes_return_501_instead_of_legacy_data(): client = TestClient(app) item_id = uuid4() run_id = uuid4() assert client.get(f"/api/decode/items/{item_id}/result").status_code == 501 assert client.post(f"/api/decode/items/{item_id}/jobs").status_code == 501 assert client.get("/api/payloads/drafts").status_code == 501 assert client.get(f"/api/pipeline/runs/{run_id}").status_code == 501 def test_formal_decode_payload_pipeline_routes_have_override_success_paths(): decode_repo = FakeDecodeRepo() pipeline_repo = FakePipelineRepo() item_id = uuid4() run_id = uuid4() app.dependency_overrides[get_decode_repository] = lambda: decode_repo app.dependency_overrides[get_pipeline_repository] = lambda: pipeline_repo client = TestClient(app) try: result = client.get(f"/api/decode/items/{item_id}/result").json() assert result["item_id"] == str(item_id) assert result["read_result"]["text"] == "读懂后的帖子内容" job = client.post(f"/api/decode/items/{item_id}/jobs").json()["job"] assert job["item_id"] == str(item_id) assert job["status"] == "pending" drafts = client.get(f"/api/payloads/drafts?item_id={item_id}").json()["items"] assert drafts[0]["item_id"] == str(item_id) assert drafts[0]["payload"]["title"] == "可审核 payload" run = client.get(f"/api/pipeline/runs/{run_id}").json()["run"] assert run["id"] == str(run_id) assert run["current_stage"] == "decode" finally: app.dependency_overrides.clear() def test_no_legacy_api_dependencies_in_formal_app_sources(): roots = [Path("app"), Path("pipeline")] text = "\n".join( path.read_text(encoding="utf-8") for root in roots for path in root.rglob("*") if path.is_file() and path.suffix in {".py", ".jsx", ".js"} ) assert "acquisition.store" not in text assert "CK_SQLITE_PATH" not in text assert "/data/queries" not in text assert "/api/creation-search" not in text assert "creation_knowledge.api" not in text def test_legacy_top_level_package_is_archived(): assert not Path("creation_knowledge").exists() assert Path("archive/2026-06-30-step6/legacy_creation_knowledge/creation_knowledge").exists() def test_active_sources_do_not_import_legacy_store_or_package(): roots = [Path("app"), Path("pipeline"), Path("acquisition"), Path("decode_content"), Path("scripts")] text = "\n".join( path.read_text(encoding="utf-8") for root in roots for path in root.rglob("*") if path.is_file() and path.suffix in {".py", ".jsx", ".js"} ) assert "from acquisition import store" not in text assert "import acquisition.store" not in text assert "from creation_knowledge" not in text assert "import creation_knowledge" not in text