from __future__ import annotations from contextlib import nullcontext from pathlib import Path from uuid import uuid4 from fastapi.testclient import TestClient from acquisition.domain import AcquisitionRun, Query, QueryBatch from app.api import app from app.dependencies import ( get_acquisition_repository, get_creation_db_config, get_decode_repository, get_pipeline_repository, ) from app.routes import decode as decode_routes from app.routes import manual_queries 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_latest_query_result_list(self, query_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, }, "run": { "id": self.run_id, "batch_id": self.batch_id, "run_key": "run-1", "status": "running", }, "jobs": [{"id": uuid4(), "platform": "xiaohongshu", "status": "done"}], "items": [ { "id": self.item_id, "platform": "xiaohongshu", "title": "脚本开头", "raw_summary": "先定受众", "status": "candidate", "content_mode": "image_post", "metadata": { "page_index": 1, "matched_candidate": {"raw": "large"}, "source_payload": {"raw": "large"}, }, } ], "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, "status": "done", "error_message": None, } ], "decode_summaries": [ { "item_id": self.item_id, "decode_status": "decoded", "particle_count": 1, "payload_count": 1, } ], } 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() self.other_payload_id = uuid4() self.marked_payload_ids = [] self.saved_ingest_records = [] 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={ "source": {"id": "xhs_1", "source_type": "post", "title": "脚本开头", "author": "作者"}, "title": "可审核 payload", "content": "先定受众，再写开头钩子。", "dim_attributes": ["how"], "dim_creations": ["创作"], "scopes": [{"scope_type": "intent", "value": "吸引注意"}], "custom_ext": [{"key": "业务阶段", "type": "str", "value": "脚本"}], }, review_status="pending", ingest_ready=False, status="draft", ) ] def mark_payload_draft_ingested(self, payload_draft_id): self.marked_payload_ids.append(payload_draft_id) return PayloadDraft( id=payload_draft_id, item_id=uuid4(), payload=self.list_payload_drafts()[0].payload, review_status="approved", ingest_ready=True, status="ingested", ) def save_ingest_record(self, **kwargs): record = IngestRecord(id=uuid4(), **kwargs) self.saved_ingest_records.append(record) return record 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): if self.saved_ingest_records: return self.saved_ingest_records 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 get_pipeline_run_by_acquisition_run(self, acquisition_run_id): return PipelineRun( id=uuid4(), run_key=f"pipeline-for-{acquisition_run_id}", status="running", current_stage="decode", ) def list_timeline(self, run_id): return [ { "id": str(uuid4()), "pipeline_run_id": str(run_id), "stage": "decode", "event_type": "decode_item_finished", "status": "done", } ] def get_timeline_bundle(self, run_id): return { "events": self.list_timeline(run_id), "jobs": [{"id": str(uuid4()), "stage": "decode", "status": "done"}], "candidate_hits": [{"id": str(uuid4()), "platform": "xiaohongshu"}], "llm_call_traces": [{"id": str(uuid4()), "stage": "decode", "status": "success"}], } class FakeManualRepo: def __init__(self): self.batch_id = uuid4() self.run_id = uuid4() self.queries = [] self.runs = [] self.updates = [] def create_query_batch(self, **kwargs): self.batch_kwargs = kwargs return QueryBatch(id=self.batch_id, **kwargs) def add_query(self, **kwargs): query = Query(id=uuid4(), **kwargs) self.queries.append(query) return query def create_acquisition_run(self, **kwargs): run = AcquisitionRun(id=self.run_id, **kwargs) self.runs.append(run) return run def update_acquisition_run(self, run_id, **kwargs): assert run_id == self.run_id self.updates.append(kwargs) base = self.runs[-1].model_dump() base.update(kwargs) return AcquisitionRun(**base) def _client(repo: FakeAcquisitionRepo): app.dependency_overrides[get_acquisition_repository] = lambda: repo return TestClient(app) def test_manual_query_batch_api_creates_ready_batch_and_starts_pipeline(monkeypatch, tmp_path): repo = FakeManualRepo() popen_calls = [] class FakeProcess: pid = 12345 def fake_popen(cmd, **kwargs): popen_calls.append((cmd, kwargs)) return FakeProcess() monkeypatch.setattr(manual_queries, "transaction", lambda _config: nullcontext(object())) monkeypatch.setattr(manual_queries, "PostgresAcquisitionRepository", lambda _conn: repo) monkeypatch.setattr(manual_queries, "RUNTIME_MANUAL_DIR", tmp_path) monkeypatch.setattr(manual_queries.subprocess, "Popen", fake_popen) app.dependency_overrides[get_creation_db_config] = lambda: object() client = TestClient(app) try: response = client.post( "/api/query-batches/manual", json={ "queries": [ "符号 视频 灵感 怎么做", "符号 视频 灵感 怎么做", {"query_text": "叙事体裁 视频 灵感 有哪些", "axes": {"实质": "叙事体裁"}}, ] }, ) assert response.status_code == 200 data = response.json() assert data["status"] == "queued" assert data["batch_id"] == str(repo.batch_id) assert "pipeline_run_id" in data assert data["run_id"] == str(repo.run_id) assert data["pid"] == 12345 assert data["query_count"] == 2 assert data["summary_url"] == f"/api/acquisition/runs/{repo.run_id}/summary" assert repo.batch_kwargs["generation_method"] == "manual_query_api_v1" assert repo.batch_kwargs["target_platforms"] == ["xiaohongshu", "weixin", "douyin"] assert [query.query_text for query in repo.queries] == [ "符号 视频 灵感 怎么做", "叙事体裁 视频 灵感 有哪些", ] assert all(query.keep is True and query.status == "ready" for query in repo.queries) assert all(query.metadata["family_key"] == "manual" for query in repo.queries) assert repo.runs[0].status == "pending" assert repo.runs[0].metadata["log_path"].endswith(".log") assert repo.runs[0].metadata["command"] assert popen_calls cmd, kwargs = popen_calls[0] assert "scripts/run_creation_pipeline.py" in cmd[1] assert ["--platform", "xiaohongshu"] == cmd[-6:-4] assert "--no-dry-ingest-record" not in cmd assert kwargs["cwd"] == str(manual_queries.ROOT) assert str(manual_queries.ROOT) in kwargs["env"]["PYTHONPATH"] assert repo.updates == [] finally: app.dependency_overrides.clear() def test_manual_query_batch_api_extracts_family_json(monkeypatch, tmp_path): repo = FakeManualRepo() monkeypatch.setattr(manual_queries, "transaction", lambda _config: nullcontext(object())) monkeypatch.setattr(manual_queries, "PostgresAcquisitionRepository", lambda _conn: repo) monkeypatch.setattr(manual_queries, "RUNTIME_MANUAL_DIR", tmp_path) monkeypatch.setattr(manual_queries.subprocess, "Popen", lambda *args, **kwargs: type("P", (), {"pid": 1})()) app.dependency_overrides[get_creation_db_config] = lambda: object() client = TestClient(app) try: response = client.post( "/api/query-batches/manual", json={ "families": [ { "key": "f1", "name": "实质正交", "items": [{"query": "公共安全 视频 灵感 怎么做"}], } ] }, ) assert response.status_code == 200 assert repo.queries[0].query_text == "公共安全 视频 灵感 怎么做" assert repo.queries[0].metadata["family_key"] == "manual" assert repo.queries[0].metadata["source_family_key"] == "f1" finally: app.dependency_overrides.clear() def test_manual_query_batch_api_rejects_empty_and_invalid_platform(): client = TestClient(app) assert client.post("/api/query-batches/manual", json={"queries": [" "]}).status_code == 422 assert client.post( "/api/query-batches/manual", json={"queries": ["符号 视频 灵感 怎么做"], "target_platforms": ["bilibili"]}, ).status_code == 422 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 latest = client.get(f"/api/query-generation/latest/queries/{repo.query_id}").json() lightweight_item = latest["items"][0] assert "body_text" not in lightweight_item assert "source_payload" not in lightweight_item assert "matched_candidate" not in lightweight_item["metadata"] assert "source_payload" not in lightweight_item["metadata"] assert lightweight_item["raw_summary"] == "先定受众" assert len(lightweight_item["media_assets"]) == 1 assert lightweight_item["decode_summary"]["particle_count"] == 1 assert latest["platforms"]["xiaohongshu"]["item_ids"] == [str(repo.item_id)] assert "items" not in latest["platforms"]["xiaohongshu"] finally: app.dependency_overrides.clear() def test_app_compresses_large_json_responses(): repo = FakeAcquisitionRepo() client = _client(repo) try: response = client.get( f"/api/query-generation/latest/queries/{repo.query_id}", headers={"Accept-Encoding": "gzip"}, ) assert response.status_code == 200 assert response.headers.get("content-encoding") == "gzip" 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_routes_return_trace_run_and_timeline(): pipeline_repo = FakePipelineRepo() app.dependency_overrides[get_pipeline_repository] = lambda: pipeline_repo client = TestClient(app) run_id = uuid4() acquisition_run_id = uuid4() try: run = client.get(f"/api/pipeline/runs/{run_id}").json()["run"] assert run["id"] == str(run_id) assert run["current_stage"] == "decode" timeline = client.get(f"/api/pipeline/runs/{run_id}/timeline").json() assert timeline["events"][0]["event_type"] == "decode_item_finished" assert timeline["jobs"][0]["stage"] == "decode" assert timeline["candidate_hits"][0]["platform"] == "xiaohongshu" assert timeline["llm_call_traces"][0]["status"] == "success" by_acq = client.get(f"/api/pipeline/runs/by-acquisition/{acquisition_run_id}").json()["run"] assert by_acq["run_key"] == f"pipeline-for-{acquisition_run_id}" finally: app.dependency_overrides.clear() 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" ingest = client.post(f"/api/decode/items/{item_id}/ingest").json() assert ingest["dry_run"] is True assert ingest["total"] == 1 assert ingest["ingested_count"] == 1 assert ingest["failed_count"] == 0 assert decode_repo.marked_payload_ids == [decode_repo.payload_id] assert decode_repo.saved_ingest_records[0].target_system == "dry-run" 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_item_ingest_real_mode_requires_configured_url(monkeypatch): acquisition_repo = FakeAcquisitionRepo() decode_repo = FakeDecodeRepo() item_id = acquisition_repo.item_id app.dependency_overrides[get_decode_repository] = lambda: decode_repo monkeypatch.setattr(decode_routes.IngestApiConfig, "from_env", lambda _env_file: decode_routes.IngestApiConfig()) client = TestClient(app) try: response = client.post(f"/api/decode/items/{item_id}/ingest", json={"dry_run": False}) assert response.status_code == 400 assert "CK_INGEST_API_URL" in response.json()["detail"] assert decode_repo.saved_ingest_records == [] 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