content-find-agent-new/tests/test_database_runtime.py

import json
import re
from pathlib import Path

from content_agent.integrations.database_runtime import (
    ContentSupplyDbConfig,
    DatabaseRuntimeStore,
)


class FakeCursor:
    def __init__(self, connection):
        self.connection = connection
        self._one = None
        self._all = []

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return None

    def execute(self, sql, params=None):
        self.connection.statements.append((sql, list(params or [])))
        if sql.startswith("SELECT COUNT(*)"):
            self._one = {"cnt": 0}
            self._all = []
        elif sql.startswith("SELECT"):
            self._one = self.connection.select_one_result
            self._all = self.connection.select_all_result

    def fetchone(self):
        return self._one

    def fetchall(self):
        return self._all


class FakeConnection:
    def __init__(self):
        self.statements = []
        self.commit_count = 0
        self.select_one_result = None
        self.select_all_result = []

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return None

    def cursor(self):
        return FakeCursor(self)

    def commit(self):
        self.commit_count += 1


def test_content_supply_db_config_reads_env_file(tmp_path):
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "CONTENT_SUPPLY_DB_HOST=127.0.0.1",
                "CONTENT_SUPPLY_DB_PORT=3307",
                "CONTENT_SUPPLY_DB_NAME=content-deconstruction-supply",
                "CONTENT_SUPPLY_DB_USER=content_rw",
                "CONTENT_SUPPLY_DB_" + "PASS" + "WORD=dummy_password",
            ]
        ),
        encoding="utf-8",
    )

    config = ContentSupplyDbConfig.from_env(env_file=env_file)

    assert config.host == "127.0.0.1"
    assert config.port == 3307
    assert config.database == "content-deconstruction-supply"
    assert config.user == "content_rw"


def test_content_supply_db_config_requires_all_project_db_keys(tmp_path):
    env_file = tmp_path / ".env"
    env_file.write_text(
        "\n".join(
            [
                "CONTENT_SUPPLY_DB_HOST=127.0.0.1",
                "CONTENT_SUPPLY_DB_PORT=3307",
                "CONTENT_SUPPLY_DB_NAME=content-deconstruction-supply",
                "CONTENT_SUPPLY_DB_USER=content_rw",
            ]
        ),
        encoding="utf-8",
    )

    try:
        ContentSupplyDbConfig.from_env(env_file=env_file)
    except ValueError as exc:
        assert "CONTENT_SUPPLY_DB_PASSWORD" in str(exc)
    else:
        raise AssertionError("expected missing db env key to fail")


def test_database_runtime_writes_source_context_with_db_schema_version():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.write_json(
        "run_001",
        "source_context.json",
        {
            "schema_version": "runtime_record.v1",
            "run_id": "run_001",
            "demand_content_id": "123",
            "ext_data": {
                "evidence_pack": {
                    "pattern_source_system": "pg_pattern_v2",
                    "source_kind": "pattern_itemset",
                    "source_post_id": "post_001",
                    "pattern_execution_id": 581,
                    "mining_config_id": 2081,
                }
            },
        },
    )

    sql, params = connection.statements[-1]
    values = _insert_values(sql, params)
    assert "INSERT INTO `content_agent_source_contexts`" in sql
    assert values["schema_version"] == "content_agent.v1"
    assert values["run_id"] == "run_001"
    assert values["demand_content_id"] == 123
    assert json.loads(values["evidence_pack"])["source_post_id"] == "post_001"
    assert json.loads(values["source_context"])["schema_version"] == "runtime_record.v1"


def test_database_runtime_derives_itemset_ids_from_seed_pack_itemsets():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.write_json(
        "run_001",
        "pattern_seed_pack.json",
        {
            "schema_version": "runtime_record.v1",
            "run_id": "run_001",
            "policy_run_id": "policy_run_001",
            "source_post_id": "post_001",
            "pattern_execution_id": 581,
            "itemsets": [{"itemset_id": 1608352}, {"itemset_id": 1608352}],
            "seed_terms": ["爱国情感", "人物故事"],
        },
    )

    sql, params = connection.statements[-1]
    values = _insert_values(sql, params)
    assert "INSERT INTO `content_agent_pattern_seed_packs`" in sql
    assert values["schema_version"] == "content_agent.v1"
    assert json.loads(values["itemset_ids"]) == [1608352]
    assert json.loads(values["pattern_seed_pack"])["schema_version"] == "runtime_record.v1"


def test_database_runtime_appends_jsonl_with_raw_payload():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.append_jsonl(
        "run_001",
        "search_queries.jsonl",
        [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": "run_001",
                "policy_run_id": "policy_run_001",
                "search_query_id": "q_001",
                "search_query": "对比分析",
                "search_query_generation_method": "item_single",
                "pattern_seed_ref": {
                    "source_field": "seed_terms",
                    "source_index": 0,
                    "seed_term": "对比分析",
                },
                "raw_payload": {"run_id": "run_001", "search_query_id": "q_001"},
            }
        ],
    )

    sql, params = connection.statements[-1]
    values = _insert_values(sql, params)
    assert "INSERT INTO `content_agent_queries`" in sql
    assert values["schema_version"] == "content_agent.v1"
    assert json.loads(values["pattern_seed_ref"])["seed_term"] == "对比分析"
    assert json.loads(values["raw_payload"])["search_query_id"] == "q_001"


def test_database_runtime_preserves_llm_variant_payload_fields():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.append_jsonl(
        "run_001",
        "search_queries.jsonl",
        [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": "run_001",
                "policy_run_id": "policy_run_001",
                "search_query_id": "q_002",
                "search_query": "人物叙事素材",
                "search_query_generation_method": "llm_variant",
                "pattern_seed_ref": {
                    "source_field": "seed_terms",
                    "source_index": 0,
                    "seed_term": "人物故事",
                },
                "llm_variant_of": "q_001",
                "raw_payload": {
                    "run_id": "run_001",
                    "policy_run_id": "policy_run_001",
                    "search_query_id": "q_002",
                    "search_query_generation_method": "llm_variant",
                    "llm_variant_of": "q_001",
                    "llm_input_evidence": {"seed_term": "人物故事"},
                    "llm_prompt_version": "fake-query-prompt-v1",
                    "llm_generation_model": "fake-query-model",
                },
            }
        ],
    )

    sql, params = connection.statements[-1]
    values = _insert_values(sql, params)
    assert "INSERT INTO `content_agent_queries`" in sql
    assert "llm_variant_of" not in values
    payload = json.loads(values["raw_payload"])
    assert payload["llm_variant_of"] == "q_001"
    assert payload["llm_input_evidence"]["seed_term"] == "人物故事"
    assert payload["llm_prompt_version"] == "fake-query-prompt-v1"
    assert payload["llm_generation_model"] == "fake-query-model"


def test_database_runtime_upserts_pattern_recall_evidence():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.append_jsonl(
        "run_001",
        "pattern_recall_evidence.jsonl",
        [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": "run_001",
                "policy_run_id": "policy_run_001",
                "recall_evidence_id": "recall_001",
                "content_discovery_id": "content_001",
                "platform": "douyin",
                "platform_content_id": "7390000000000000000",
                "decode_status": "success",
                "decode_task_id": "decode_task_001",
                "recall_status": "matched",
                "matched_terms": ["爱国情感"],
                "matched_category_paths": [
                    "/理念/观念/个人观念/情感认同/国家民族认同/爱国情感"
                ],
                "match_paths_request": {"source_type": "实质"},
                "match_paths_response": {"data": []},
                "evidence_summary": {
                    "primary_matched_category_path": (
                        "/理念/观念/个人观念/情感认同/国家民族认同/爱国情感"
                    )
                },
                "raw_payload": {
                    "platform": "douyin",
                    "primary_matched_category_path": (
                        "/理念/观念/个人观念/情感认同/国家民族认同/爱国情感"
                    ),
                },
            }
        ],
    )

    sql, params = connection.statements[-1]
    values = _insert_values(sql, params)
    assert "INSERT INTO `content_agent_pattern_recall_evidence`" in sql
    assert "ON DUPLICATE KEY UPDATE" in sql
    assert values["schema_version"] == "content_agent.v1"
    assert values["recall_evidence_id"] == "recall_001"
    assert "platform" not in values
    assert "primary_matched_category_path" not in values
    assert json.loads(values["matched_terms"]) == ["爱国情感"]
    assert json.loads(values["raw_payload"])["platform"] == "douyin"


def test_database_runtime_read_jsonl_reconstructs_runtime_payload():
    connection = FakeConnection()
    connection.select_all_result = [
        {
            "raw_payload": json.dumps(
                {
                    "record_schema_version": "runtime_record.v1",
                    "run_id": "run_001",
                    "policy_run_id": "policy_run_001",
                    "search_query_id": "q_001",
                }
            )
        }
    ]
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    rows = store.read_jsonl("run_001", "search_queries.jsonl")

    assert rows[0]["search_query_id"] == "q_001"
    assert rows[0]["raw_payload"]["search_query_id"] == "q_001"


def test_database_runtime_rejects_forbidden_raw_payload_keys_in_lists():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    try:
        store.append_jsonl(
            "run_001",
            "search_queries.jsonl",
            [
                {
                    "record_schema_version": "runtime_record.v1",
                    "run_id": "run_001",
                    "policy_run_id": "policy_run_001",
                    "search_query_id": "q_001",
                    "search_query": "对比分析",
                    "raw_payload": {"items": [{"dsn": "should_not_be_stored"}]},
                }
            ],
        )
    except ValueError as exc:
        assert "forbidden key" in str(exc)
    else:
        raise AssertionError("expected forbidden raw_payload key to be rejected")


def test_database_runtime_update_run_record_ignores_empty_sanitized_updates():
    connection = FakeConnection()
    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)

    store.update_run_record("run_001", {"unknown_field": "ignored", "status": None})

    assert connection.statements == []


def test_business_modules_do_not_import_or_name_database_tables():
    root = Path("content_agent/business_modules")
    text = "\n".join(path.read_text(encoding="utf-8") for path in root.rglob("*.py"))

    assert not re.search(
        r"pymysql|sqlalchemy|psycopg|sqlite3|SELECT |INSERT |UPDATE |DELETE |SHOW |CREATE |ALTER |content_agent_",
        text,
    )


def _config():
    return ContentSupplyDbConfig(
        host="127.0.0.1",
        port=3306,
        user="content_rw",
        password="dummy_password",
        database="content-deconstruction-supply",
    )


def _insert_values(sql, params):
    match = re.search(r"\((.*?)\) VALUES", sql)
    assert match, sql
    columns = [part.strip().strip("`") for part in match.group(1).split(",")]
    return dict(zip(columns, params))