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

from __future__ import annotations

from uuid import uuid4

from acquisition.repositories.postgres import PostgresAcquisitionRepository


class RecordingPostgresRepo(PostgresAcquisitionRepository):
    def __init__(self):
        super().__init__(conn=None)
        self.one_calls = []
        self.all_calls = []
        self.one_results = []
        self.all_results = []

    def _one(self, sql, params):
        self.one_calls.append((sql, params))
        return self.one_results.pop(0)

    def _all(self, sql, params):
        self.all_calls.append((sql, params))
        return self.all_results.pop(0)


def test_postgres_summary_uses_formal_run_job_item_tables():
    run_id = uuid4()
    batch_id = uuid4()
    query_id = uuid4()
    repo = RecordingPostgresRepo()
    repo.one_results.append(
        {
            "id": run_id,
            "run_key": "r1",
            "batch_id": batch_id,
            "status": "done",
            "query_count": 1,
            "job_count": 3,
            "candidate_count": 7,
            "creation_hit_count": 2,
            "started_at": None,
            "finished_at": None,
        }
    )
    repo.all_results.append(
        [
            {
                "query_id": query_id,
                "query_text": "脚本 开头",
                "job_count": 3,
                "candidate_count": 7,
                "creation_hit_count": 2,
                "platforms": {"xiaohongshu": {"status": "done"}},
            }
        ]
    )

    summary = repo.get_run_summary(run_id)

    first_sql, first_params = repo.one_calls[0]
    second_sql, second_params = repo.all_calls[0]
    assert "FROM acquisition_runs ar" in first_sql
    assert "LEFT JOIN acquisition_jobs aj" in first_sql
    assert "LEFT JOIN candidate_items ci" in first_sql
    assert "LEFT JOIN item_classifications ic" in first_sql
    assert first_params == (run_id,)
    assert "jsonb_object_agg" in second_sql
    assert "WHERE aj.run_id = %s" in second_sql
    assert second_params == (run_id,)
    assert summary["queries"][0]["query_id"] == query_id


def test_postgres_repository_lists_only_creation_candidates_for_decode():
    run_id = uuid4()
    item_id = uuid4()
    repo = RecordingPostgresRepo()
    repo.all_results.append(
        [
            {
                "id": item_id,
                "platform": "xiaohongshu",
                "platform_item_id": "x1",
                "canonical_url": "https://xhs.test/1",
                "status": "candidate",
            }
        ]
    )

    items = repo.list_creation_candidate_items(run_id=run_id, limit=20)

    sql, params = repo.all_calls[0]
    assert "JOIN acquisition_jobs aj ON aj.id = ci.job_id" in sql
    assert "JOIN item_classifications ic ON ic.item_id = ci.id" in sql
    assert "ic.is_creation_knowledge IS TRUE" in sql
    assert "LIMIT %s" in sql
    assert params == (run_id, 20)
    assert items[0].id == item_id
    assert items[0].platform == "xiaohongshu"


def test_postgres_query_detail_loads_media_and_classifications_only_when_items_exist():
    run_id = uuid4()
    query_id = uuid4()
    item_id = uuid4()
    repo = RecordingPostgresRepo()
    repo.one_results.append({"id": query_id, "query_text": "q", "status": "ready"})
    repo.all_results.extend(
        [
            [{"id": uuid4(), "run_id": run_id, "query_id": query_id, "platform": "weixin"}],
            [{"id": item_id, "platform": "weixin", "query_id": query_id, "status": "candidate"}],
            [{"id": uuid4(), "item_id": item_id, "media_type": "image", "status": "done"}],
            [{"id": uuid4(), "item_id": item_id, "status": "classified"}],
        ]
    )

    detail = repo.get_query_detail(run_id=run_id, query_id=query_id)

    assert len(repo.all_calls) == 4
    assert "SELECT * FROM media_assets" in repo.all_calls[2][0]
    assert "SELECT * FROM item_classifications" in repo.all_calls[3][0]
    assert detail["items"][0]["id"] == item_id
    assert detail["media_assets"][0]["item_id"] == item_id