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

import json
from pathlib import Path

from content_agent.integrations.database_runtime import JSON_COLUMNS_BY_TABLE, TABLE_COLUMNS
from scripts.count_schema_registry import parse_sql_schema


ROOT = Path(__file__).resolve().parents[1]
REGISTRY_PATH = ROOT / "tech_documents/数据库字段总览/content_agent_schema_registry.json"
LEGACY_REGISTRY_FIELDS = {
    "content_agent_discovered_content_items.content_audience_profile",
    "content_agent_rule_decisions.age_50_plus_level",
    "content_agent_pattern_recall_evidence.decode_status",
    "content_agent_pattern_recall_evidence.decode_task_id",
    "content_agent_pattern_recall_evidence.matched_terms",
    "content_agent_pattern_recall_evidence.matched_category_paths",
    "content_agent_pattern_recall_evidence.decode_elements",
    "content_agent_pattern_recall_evidence.match_paths_request",
    "content_agent_pattern_recall_evidence.match_paths_response",
}
V4_TOUCHED_TABLES = {
    "content_agent_queries",
    "content_agent_pattern_recall_evidence",
    "content_agent_rule_decisions",
    "content_agent_walk_actions",
    "content_agent_source_path_records",
    "content_agent_author_assets",
    "content_agent_discovered_content_items",
    "content_agent_search_clue_assets",
}


def test_schema_registry_counts_match_sql_v4_baseline():
    registry = _registry()
    sql_tables = parse_sql_schema()

    assert len(sql_tables) == 25
    assert sum(len(table.columns) for table in sql_tables.values()) == 418
    assert sum(1 for table in sql_tables.values() for column in table.columns if column.is_json) == 63
    assert registry["database"]["sql_column_count"] == 418
    assert registry["database"]["json_column_count"] == 63
    assert registry["coverage_targets"]["sql_column_count"] == 418
    assert registry["coverage_targets"]["json_column_count"] == 63


def test_v4_touched_tables_match_sql_runtime_and_json_column_whitelist():
    registry = _registry()
    sql_tables = parse_sql_schema()

    for table_name in V4_TOUCHED_TABLES:
        sql_columns = {column.name for column in sql_tables[table_name].columns}
        registry_columns = {
            column["field_name"]
            for column in registry["tables"][table_name]["columns"]
        }
        assert registry_columns == sql_columns
        if table_name in TABLE_COLUMNS:
            assert TABLE_COLUMNS[table_name] <= sql_columns
        json_columns = {column.name for column in sql_tables[table_name].columns if column.is_json}
        assert JSON_COLUMNS_BY_TABLE.get(table_name, set()) == json_columns


def test_registry_no_longer_registers_sql_missing_legacy_fields():
    registry = _registry()
    registry_refs = {
        f"{table_name}.{column['field_name']}"
        for table_name, table in registry["tables"].items()
        for column in table["columns"]
    }
    indexed_refs = {
        ref
        for refs in registry["field_index"].values()
        for ref in refs
    }

    assert not (LEGACY_REGISTRY_FIELDS & registry_refs)
    assert not (LEGACY_REGISTRY_FIELDS & indexed_refs)


def test_raw_payload_policy_matches_runtime_v4_tables():
    policy = _registry()["raw_payload_policy"]

    assert {
        "access_token",
        "refresh_token",
        "apikey",
        "authorization",
        "cookie",
        "session",
        "credential",
    } <= set(policy["forbidden_keys"])
    assert {
        "content_agent_pattern_recall_evidence",
        "content_agent_walk_actions",
    } <= set(policy["required_tables"])


def _registry():
    return json.loads(REGISTRY_PATH.read_text(encoding="utf-8"))