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, IngestRecord, KnowledgeParticle, PayloadDraft, ScopeResult, ) 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", "unique_key": "xhs:xhs1", "canonical_url": "https://xhs.test/1", "content_type": "图文", "content_mode": "image_post", "title": "脚本开头", "author_name": "作者", "body_text": "先定受众，再写开头钩子", "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", "error_message": None, } ], } def get_candidate_item(self, item_id): assert item_id == self.item_id return self.get_query_detail(run_id=self.run_id, query_id=self.query_id)["items"][0] def list_media_assets_for_item(self, item_id): assert item_id == self.item_id return self.get_query_detail(run_id=self.run_id, query_id=self.query_id)["media_assets"] 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", ) ] def list_knowledge_particles(self, item_id=None): return [ KnowledgeParticle( id=uuid4(), item_id=item_id or uuid4(), particle_type="what", title="What I know", content={"content": "一个可复用知识点"}, status="draft", ) ] def list_scope_results(self, item_id=None): return [ ScopeResult( id=uuid4(), item_id=item_id or uuid4(), scope_type="intent", scope_value="吸引注意", status="draft", ) ] def list_ingest_records(self, item_id=None): return [ IngestRecord( id=uuid4(), payload_draft_id=self.payload_id, target_system="dry-run", status="ingested", ) ] 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["content_mode"] == "image_post" assert item["body_text"] == "先定受众，再写开头钩子" 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_query_generation_preview_defaults_to_first_two_families(): client = TestClient(app) data = client.get("/api/query-generation/preview?per=2").json() assert data["summary"]["family_keys"] == ["f1", "f2"] assert [family["key"] for family in data["families"]] == ["f1", "f2"] assert data["summary"]["query_count"] == 4 assert data["metadata"]["active_family_keys"] == ["f1", "f2"] def test_query_generation_preview_can_return_full_cartesian_combinations(): client = TestClient(app) data = client.get("/api/query-generation/preview?per=0&batch_n=1").json() assert data["summary"]["family_keys"] == ["f1", "f2"] assert data["summary"]["query_count"] == 36 assert [len(family["items"]) for family in data["families"]] == [18, 18] 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_pipeline_route_returns_501_instead_of_legacy_data_when_not_wired(): client = TestClient(app) run_id = uuid4() assert client.get(f"/api/pipeline/runs/{run_id}").status_code == 501 def test_formal_decode_payload_pipeline_routes_have_override_success_paths(): acquisition_repo = FakeAcquisitionRepo() decode_repo = FakeDecodeRepo() pipeline_repo = FakePipelineRepo() item_id = acquisition_repo.item_id run_id = uuid4() app.dependency_overrides[get_acquisition_repository] = lambda: acquisition_repo 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" detail = client.get(f"/api/decode/items/{item_id}/detail").json() assert detail["item"]["id"] == str(item_id) assert detail["knowledge_particles"][0]["title"] == "What I know" assert detail["ingest_records"][0]["status"] == "ingested" 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