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
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("V1")
    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():
    import json

    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():
    import json
    from copy import deepcopy

    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_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
    walk_strategy = json.loads(
        (ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json").read_text("utf-8")
    )

    assert meta["strategy_id"] == walk_strategy["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}"