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

from __future__ import annotations

from uuid import uuid4

from core.config import IngestApiConfig
from decode_content.ingest import KnowledgeIngestClient, ingest_payload_draft
from decode_content.models import IngestRecord, PayloadDraft


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

    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


class FakeResponse:
    def __init__(self, status_code, payload):
        self.status_code = status_code
        self._payload = payload
        self.text = str(payload)

    def json(self):
        return self._payload


def _payload():
    return {
        "source": {"id": "xhs_1", "source_type": "post", "title": "帖子", "author": "作者"},
        "title": "拍照姿势",
        "content": "拍照时用手部动作制造自然感。",
        "dim_attributes": ["how"],
        "dim_creations": ["创作"],
        "scopes": [{"scope_type": "intent", "value": "自然感"}],
        "custom_ext": [{"key": "业务阶段", "type": "str", "value": "脚本"}],
    }


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

    record = ingest_payload_draft(repo, draft, dry_run=True)

    assert repo.marked == [draft_id]
    assert record.target_system == "dry-run"
    assert record.status == "ingested"
    assert record.response_payload["dry_run"] is True


def test_real_ingest_posts_and_records_knowledge_id():
    calls = []

    def fake_post(url, **kwargs):
        calls.append((url, kwargs))
        return FakeResponse(201, {"knowledge_id": "k-1", "source_id": "xhs_1"})

    repo = FakeRepo()
    draft_id = uuid4()
    draft = PayloadDraft(id=draft_id, payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
    client = KnowledgeIngestClient(
        IngestApiConfig(url="https://ingest.test/api/v1/knowledge/ingest", retry_delays_seconds=()),
        post_fn=fake_post,
        sleep_fn=lambda _delay: None,
    )

    record = ingest_payload_draft(repo, draft, dry_run=False, client=client)

    assert calls and calls[0][0] == "https://ingest.test/api/v1/knowledge/ingest"
    assert record.target_system == "knowledge-ingest-api"
    assert record.target_id == "k-1"
    assert record.response_payload["source_id"] == "xhs_1"
    assert repo.marked == [draft_id]


def test_real_ingest_failure_is_recorded_without_marking_draft():
    def fake_post(url, **kwargs):
        return FakeResponse(502, {"detail": "embedding down"})

    repo = FakeRepo()
    draft = PayloadDraft(id=uuid4(), payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
    client = KnowledgeIngestClient(
        IngestApiConfig(url="https://ingest.test/api/v1/knowledge/ingest", retry_delays_seconds=()),
        post_fn=fake_post,
        sleep_fn=lambda _delay: None,
    )

    record = ingest_payload_draft(repo, draft, dry_run=False, client=client)

    assert record.status == "failed"
    assert record.target_system == "knowledge-ingest-api"
    assert record.response_payload["http_status"] == 502
    assert repo.marked == []


def test_real_ingest_retries_before_success():
    calls = []
    sleeps = []

    def fake_post(url, **kwargs):
        calls.append(url)
        if len(calls) == 1:
            return FakeResponse(502, {"detail": "embedding down"})
        return FakeResponse(201, {"knowledge_id": "k-2", "source_id": "xhs_1"})

    repo = FakeRepo()
    draft = PayloadDraft(id=uuid4(), payload=_payload(), review_status="pending", ingest_ready=False, status="draft")
    client = KnowledgeIngestClient(
        IngestApiConfig(url="https://ingest.test/api/v1/knowledge/ingest", retry_delays_seconds=(0.01,)),
        post_fn=fake_post,
        sleep_fn=lambda delay: sleeps.append(delay),
    )

    record = ingest_payload_draft(repo, draft, dry_run=False, client=client)

    assert len(calls) == 2
    assert sleeps == [0.01]
    assert record.status == "ingested"
    assert record.attempt_count == 2
    assert record.target_id == "k-2"