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

"""V2-M1 config tooling safety net: config_store, validators, Excel↔JSON sync."""

from __future__ import annotations

import importlib.util
import json
from copy import deepcopy
from pathlib import Path

import pytest

from content_agent.integrations import config_store
from content_agent.integrations.policy_json import JsonPolicyBundleStore

ROOT = Path(__file__).resolve().parents[1]
RULE_PACK_JSON = ROOT / "product_documents/规则包/douyin_rule_packs.v1.json"


def _load_script(name: str):
    spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def test_config_store_hash_matches_raw_file_bytes():
    parsed, raw = config_store.load_json(RULE_PACK_JSON)
    assert config_store.sha256_text(raw) == config_store.sha256_text(RULE_PACK_JSON.read_text("utf-8"))
    # policy_bundle_hash must still hash the raw rule-pack text (refactor parity).
    bundle = JsonPolicyBundleStore(".").load_policy_bundle("V4")
    assert bundle["policy_bundle_hash"] == config_store.sha256_text(raw)
    assert bundle["strategy_source_ref"]["content_sha256"] == bundle["policy_bundle_hash"]


def test_rule_pack_fk_validator_has_no_failures():
    mod = _load_script("validate_rule_pack_config")
    findings = mod.validate_rule_pack_config(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    assert [f for f in findings if f["level"] == "fail"] == []


def test_rule_pack_validator_rejects_enabled_dispatch_conflict():
    mod = _load_script("validate_rule_pack_config")
    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    duplicate = dict(pkg["rule_pack_dispatch"][0])
    duplicate["dispatch_id"] = "dispatch_content_duplicate"
    pkg["rule_pack_dispatch"].append(duplicate)

    findings = mod.validate_rule_pack_config(pkg)

    conflicts = [f for f in findings if f["check_id"] == "dispatch_conflict"]
    assert len(conflicts) == 1
    assert conflicts[0]["level"] == "fail"
    assert "CONFIG_RULE_PACK_DISPATCH_CONFLICT" in conflicts[0]["message"]
    assert "douyin_content_discovery_rule_pack_v1" in conflicts[0]["message"]


def test_rule_pack_validator_allows_extra_active_dimension():
    mod = _load_script("validate_rule_pack_config")
    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    scorecard = pkg["rule_packs"][0]["scorecard"]
    scorecard["dimensions"].append(
        {
            "key": "fifty_plus_fit",
            "label": "50+ 匹配",
            "max_score": 20,
            "weight_percent": 20,
            "runtime_status": "active",
            "evidence_paths": ["content_audience_50plus.score"],
        }
    )
    scorecard["scoring_rules"].append(
        {
            "scoring_rule_id": "score_fifty_plus_precomputed",
            "dimension_key": "fifty_plus_fit",
            "field_path": "content_audience_50plus.score",
            "operator": "gte",
            "expected_value": 0,
            "score_value": 20,
            "missing_policy": "v4_score_missing_policy",
            "priority": 3,
            "enabled": True,
        }
    )

    findings = mod.validate_rule_pack_config(pkg)

    assert [finding for finding in findings if finding["level"] == "fail"] == []


def test_rule_pack_validator_accepts_m4_score_weight_profiles():
    mod = _load_script("validate_rule_pack_config")
    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    scorecard = pkg["rule_packs"][0]["scorecard"]
    profiles = scorecard["score_weight_profiles"]["profiles"]
    aliases = {
        alias
        for profile in profiles
        for alias in profile.get("historical_alias_ids", [])
    }

    assert aliases == {
        "default_two_dimension",
        "douyin_fifty_plus_ok",
        "douyin_fifty_plus_not_attempted",
    }
    assert {profile["profile_id"] for profile in profiles} == {
        "base_video_two_dimension_v1",
        "audience_50plus_available_v1",
        "audience_50plus_skipped_by_query_gate_v1",
    }

    findings = mod.validate_rule_pack_config(pkg)

    assert [finding for finding in findings if finding["level"] == "fail"] == []


def test_rule_pack_validator_rejects_invalid_m2_scorecard_fields():
    mod = _load_script("validate_rule_pack_config")
    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    dimension = pkg["rule_packs"][0]["scorecard"]["dimensions"][0]
    dimension["active"] = False
    dimension["platform_scope"] = "douyin"
    dimension["score_source_path"] = ""

    findings = mod.validate_rule_pack_config(pkg)
    check_ids = {finding["check_id"] for finding in findings if finding["level"] == "fail"}

    assert {
        "v4_scorecard_dimension_active",
        "v4_scorecard_dimension_platform_scope",
        "v4_scorecard_dimension_score_source_path",
    } <= check_ids


def test_rule_pack_validator_rejects_invalid_score_weight_profiles():
    mod = _load_script("validate_rule_pack_config")
    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    scorecard = pkg["rule_packs"][0]["scorecard"]
    profiles = scorecard["score_weight_profiles"]["profiles"]
    profiles[0]["historical_alias_ids"] = []
    profiles[1]["weights"]["made_up_dimension"] = 10
    profiles[2]["normalization_reason"] = ""

    findings = mod.validate_rule_pack_config(pkg)
    check_ids = {finding["check_id"] for finding in findings if finding["level"] == "fail"}

    assert "v4_score_weight_profiles" in check_ids
    assert "v4_score_weight_profile_dimension" in check_ids
    assert "v4_score_weight_profile_normalization" in check_ids


def test_rule_pack_validator_requires_base_dimensions_and_known_rule_refs():
    mod = _load_script("validate_rule_pack_config")
    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    scorecard = pkg["rule_packs"][0]["scorecard"]
    scorecard["dimensions"] = [
        row for row in scorecard["dimensions"] if row["key"] != "query_relevance"
    ]

    missing_base = mod.validate_rule_pack_config(pkg)

    assert any(finding["check_id"] == "v4_scorecard_dimensions" for finding in missing_base)

    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
    pkg["rule_packs"][0]["scorecard"]["scoring_rules"][0]["dimension_key"] = "made_up_dimension"

    unknown_ref = mod.validate_rule_pack_config(pkg)

    assert any(finding["check_id"] == "scoring_rule_dimension" for finding in unknown_ref)


def test_excel_meta_strategy_id_matches_walk_strategy():
    import json

    openpyxl = pytest.importorskip("openpyxl")
    workbook = openpyxl.load_workbook(
        ROOT / "tech_documents/规则包映射/规则包映射配置表.xlsx", read_only=True
    )
    rows = list(workbook["rule_package_meta"].iter_rows(values_only=True))
    meta = dict(zip(rows[0], rows[2]))  # row 2 is the data-dictionary row; row 3 is data
    rule_pack = json.loads(RULE_PACK_JSON.read_text("utf-8"))

    assert meta["strategy_id"] == rule_pack["strategy_binding"]["strategy_id"]


def test_query_prompts_validator_passes_after_m2():
    mod = _load_script("validate_query_prompts_config")
    assert mod.main() == 0


def test_query_prompts_validator_rejects_invalid_profiles():
    mod = _load_script("validate_query_prompts_config")

    assert mod.validate_query_prompts_config({}) == [
        {"level": "fail", "check_id": "profiles", "message": "no profiles defined"}
    ]

    findings = mod.validate_query_prompts_config(
        {
            "profiles": {
                "douyin/V1": {
                    "system": "s",
                    "user": "u",
                    "temperature": 3,
                    "max_tokens": 0,
                    "variants_per_seed": 0,
                }
            }
        }
    )

    assert {"level": "fail", "check_id": "missing_field", "message": "douyin/V1 missing evidence_fields"} in findings
    assert {"level": "fail", "check_id": "temperature", "message": "douyin/V1 temperature out of [0,2]: 3"} in findings
    assert {"level": "fail", "check_id": "max_tokens", "message": "douyin/V1 max_tokens must be > 0"} in findings
    assert {"level": "fail", "check_id": "variants_per_seed", "message": "douyin/V1 variants_per_seed must be >= 1"} in findings


def test_excel_matches_json_byte_equal():
    pytest.importorskip("openpyxl")
    from scripts.build_config_from_excel import build
    from scripts.check_config_json_canonical import canonical_dumps

    for path_str, obj in build().items():
        assert canonical_dumps(obj) == Path(path_str).read_text("utf-8"), f"Excel drift in {path_str}"


def test_m4_runtime_rows_require_approved_and_enabled():
    mod = _load_script("build_config_from_excel")
    rows = [
        {
            "profile_id": "approved_profile",
            "profile_label": "approved",
            "historical_alias_ids": "default_two_dimension",
            "platform_scope": "*",
            "content_format": "video",
            "applies_when": "always",
            "dimension_key": "query_relevance",
            "weight_percent": 100,
            "normalization_policy": "sum_100",
            "missing_dimension_policy": "technical_retry",
            "critical_dimension": True,
            "score_source_path": "pattern_match_result.query_relevance_score",
            "priority": 1,
            "runtime_enabled": True,
            "approval_status": "approved",
            "owner": "strategy",
            "last_reviewed_at": "2026-06-29",
        },
        {
            "profile_id": "waiting_profile",
            "dimension_key": "platform_performance",
            "weight_percent": 100,
            "runtime_enabled": True,
            "approval_status": "waiting_for_human_reivew",
            "owner": "strategy",
            "last_reviewed_at": "2026-06-29",
        },
    ]

    built = mod._build_score_weight_profiles(rows)

    assert [profile["profile_id"] for profile in built["profiles"]] == ["approved_profile"]


def test_m4_threshold_profiles_generate_compact_runtime_json():
    mod = _load_script("build_config_from_excel")
    base = {"schema_version": "score_thresholds.v1", "_meaning": "x"}
    rows = [
        {
            "platform_scope": "*",
            "threshold_key": "pool_total",
            "threshold_value": 70,
            "runtime_enabled": True,
            "approval_status": "approved",
            "owner": "strategy",
            "last_reviewed_at": "2026-06-29",
        },
        {
            "platform_scope": "douyin",
            "threshold_key": "pool_total",
            "threshold_value": 65,
            "runtime_enabled": False,
            "approval_status": "approved",
            "owner": "strategy",
            "last_reviewed_at": "2026-06-29",
        },
    ]

    built = mod._build_score_thresholds(base, rows)

    assert built["default"]["pool_total"] == 70
    assert "douyin" not in built


def test_m4_dead_config_archive_validator_passes_and_blocks_runtime_sources(tmp_path):
    mod = _load_script("validate_v5_m4_dead_config_archive")

    assert mod.validate_archive(ROOT / "archive/v5_m4_dead_config_archive") == []

    archive_root = tmp_path / "archive"
    batch = archive_root / "batch"
    batch.mkdir(parents=True)
    (batch / "score_thresholds.json").write_text("{}", encoding="utf-8")
    (batch / "manifest.json").write_text(
        json.dumps(
            {
                "items": [
                    {
                        "archive_id": "bad",
                        "source_path": "tech_documents/数据接口与来源/score_thresholds.json",
                        "archive_path": str((batch / "score_thresholds.json").relative_to(ROOT))
                        if (batch / "score_thresholds.json").is_relative_to(ROOT)
                        else "archive/batch/score_thresholds.json",
                        "archived_at": "2026-06-29T00:00:00Z",
                        "archived_by": "test",
                        "proof_no_runtime_read": "none",
                        "restore_instruction": "restore",
                        "approval_status": "approved",
                    }
                ]
            }
        ),
        encoding="utf-8",
    )

    findings = mod.validate_archive(archive_root)

    assert any(finding["check_id"] == "archive_forbidden_source" for finding in findings)