Create-knowledge-find-decode/tests/test_app_api.py

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",
                    "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",
                }
            ],
        }

    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["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_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