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

from __future__ import annotations

from uuid import uuid4

import pytest
from pydantic import ValidationError

from app.schemas import (
    AcquisitionRunSummarySchema,
    CandidateItemSchema,
    ErrorResponse,
    Page,
)
from core.config import CreationDbConfig
from decode_content.models import (
    IngestRecord,
    KnowledgeParticle,
    PayloadDraft,
    ReadResult,
    ScopeResult,
)
from pipeline.models import PipelineJob, PipelineRun, ResumeCursor


def test_creation_db_config_reads_ck_db_without_open_aigc(tmp_path):
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "OPEN_AIGC_PG_HOST=wrong-open-aigc",
                "OPEN_AIGC_PG_USER=wrong-user",
                "CK_DB_HOST=127.0.0.1",
                "CK_DB_PORT=6543",
                "CK_DB_NAME=creation_knowledge_prod",
                "CK_DB_USER=ck_app",
                "CK_DB_PASSWORD=secret",
                "CK_DB_SCHEMA=creation_knowledge",
            ]
        ),
        encoding="utf-8",
    )

    cfg = CreationDbConfig.from_env(env_file)

    assert cfg.host == "127.0.0.1"
    assert cfg.port == 6543
    assert cfg.user == "ck_app"
    assert cfg.database == "creation_knowledge_prod"
    assert cfg.schema == "creation_knowledge"


def test_decode_content_models_are_contract_shaped():
    item_id = uuid4()
    read = ReadResult(
        text="一条创作判断",
        cards=[{"index": 1, "kind": "image", "content": "构图说明"}],
        media={"platform": "xiaohongshu"},
    )
    knowledge = KnowledgeParticle(
        item_id=item_id,
        particle_type="how",
        title="先定受众再写脚本",
        content={"steps": [{"input": "目标受众", "directive": "收窄", "output": "脚本方向"}]},
    )
    scope = ScopeResult(
        item_id=item_id,
        particle_id=uuid4(),
        scope_type="意图",
        scope_value="吸引注意",
        is_reused=True,
    )
    payload = PayloadDraft(item_id=item_id, particle_id=knowledge.id, payload={"kind": "how"})
    ingest = IngestRecord(
        payload_draft_id=payload.id,
        target_system="creation-knowledge-db",
        target_id="remote-1",
    )

    assert read.cards[0].content == "构图说明"
    assert knowledge.particle_type == "how"
    assert scope.scope_value == "吸引注意"
    assert payload.review_status == "pending"
    assert ingest.status == "pending"


def test_decode_content_rejects_unknown_particle_type():
    with pytest.raises(ValidationError):
        KnowledgeParticle(particle_type="idea", title="错误类型")


def test_pipeline_models_capture_resume_boundary():
    run_id = uuid4()
    run = PipelineRun(id=run_id, run_key="formal-run", current_stage="decode")
    job = PipelineJob(run_id=run_id, stage="decode", target_id=uuid4())
    cursor = ResumeCursor(stage="decode", last_successful_job_id=job.id)

    assert run.status == "pending"
    assert job.status == "pending"
    assert cursor.stage == "decode"


def test_app_schemas_cover_formal_workbench_shapes():
    run_id = uuid4()
    item_id = uuid4()
    summary = AcquisitionRunSummarySchema(
        id=run_id,
        status="running",
        query_count=3,
        job_count=9,
        candidate_count=21,
        creation_hit_count=5,
    )
    item = CandidateItemSchema(
        id=item_id,
        platform="douyin",
        title="短视频脚本创作",
        status="candidate",
    )
    page = Page(total=1, items=[{"id": str(item_id)}])
    error = ErrorResponse(code="not_found", message="missing")

    assert summary.creation_hit_count == 5
    assert item.media_assets == []
    assert page.total == 1
    assert error.detail == {}