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

from __future__ import annotations

from uuid import uuid4

from decode_content.models import IngestRecord, PayloadDraft
from scripts.ingest_payloads import ingest_existing_payloads, list_payload_drafts_for_run


def _payload():
    return {
        "source": {"id": "xhs_1", "source_type": "post", "title": "帖子"},
        "title": "街拍姿势",
        "content": "侧身站立可以拉长身体线条。",
        "dim_attributes": ["how"],
        "dim_creations": ["创作"],
        "scopes": [{"scope_type": "effect", "value": "显高"}],
        "custom_ext": [{"key": "业务阶段", "type": "str", "value": "灵感"}],
    }


class FakeRepo:
    def __init__(self):
        self.queries = []
        self.marked = []
        self.records = []

    def _all(self, sql, params):
        self.queries.append((sql, params))
        return [
            {
                "id": uuid4(),
                "particle_id": None,
                "item_id": uuid4(),
                "payload": _payload(),
                "review_status": "pending",
                "ingest_ready": False,
                "status": "draft",
                "error_message": None,
                "created_at": None,
                "updated_at": None,
            }
        ]

    def mark_payload_draft_ingested(self, payload_draft_id):
        self.marked.append(payload_draft_id)
        return PayloadDraft(
            id=payload_draft_id,
            payload=_payload(),
            review_status="approved",
            ingest_ready=True,
            status="ingested",
        )

    def save_ingest_record(self, **kwargs):
        record = IngestRecord(id=uuid4(), **kwargs)
        self.records.append(record)
        return record


def test_list_payload_drafts_for_run_skips_real_ingested_by_default():
    repo = FakeRepo()
    run_id = uuid4()

    drafts = list_payload_drafts_for_run(repo, run_id=run_id)

    sql, params = repo.queries[0]
    assert drafts and isinstance(drafts[0], PayloadDraft)
    assert params == (run_id,)
    assert "target_system = 'knowledge-ingest-api'" in sql
    assert "ir.status = 'ingested'" in sql


def test_list_payload_drafts_for_run_can_include_real_ingested():
    repo = FakeRepo()

    list_payload_drafts_for_run(repo, run_id=uuid4(), include_real_ingested=True)

    sql, _params = repo.queries[0]
    assert "target_system = 'knowledge-ingest-api'" not in sql


def test_ingest_existing_payloads_dry_run_records_without_http():
    repo = FakeRepo()
    draft = PayloadDraft(id=uuid4(), payload=_payload(), review_status="pending", ingest_ready=False, status="draft")

    records = ingest_existing_payloads(repo, [draft], dry_run=True, env_file=".env")

    assert repo.marked == [draft.id]
    assert records[0]["target_system"] == "dry-run"
    assert records[0]["status"] == "ingested"