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

from pathlib import Path
from copy import deepcopy
import json

from scripts.validate_v4_config_contract import (
    _check_v4_rule_pack_contract,
    assert_no_v4_legacy_fields,
    validate_v4_config_contract,
)


ROOT = Path(__file__).resolve().parents[1]


def test_v4_config_contract_passes_with_m3_rule_pack_switch():
    assert validate_v4_config_contract(ROOT) == []


def test_v4_contract_fixture_contains_no_legacy_rule_fields():
    fixture = {
        "schema_version": "v4_scorecard.v1",
        "scorecard": {
            "query_relevance_score": 80,
            "platform_performance_score": 70,
            "missing_observable_fields": [],
        },
        "decision_replay_data": {"allow_walk": True},
    }

    assert assert_no_v4_legacy_fields(fixture) == []


def test_v4_legacy_blocklist_reports_precise_paths():
    fixture = {
        "schema_version": "v4_scorecard.v1",
        "scorecard": {
            "query_relevance_score": 80,
            "platform_heat": 70,
        },
    }

    assert assert_no_v4_legacy_fields(fixture) == ["v4_contract.scorecard.platform_heat"]


def test_production_v4_rule_pack_contains_no_legacy_scoring_fields():
    pkg = json.loads(
        (ROOT / "product_documents/规则包/douyin_rule_packs.v1.json").read_text(encoding="utf-8")
    )
    pack = pkg["rule_packs"][0]

    assert pkg["strategy_binding"]["strategy_version"] == "V4"
    assert pack["scorecard"]["schema_version"] == "v4_scorecard.v1"
    assert assert_no_v4_legacy_fields(pack, "rule_pack") == []
    assert [row["key"] for row in pack["scorecard"]["dimensions"]] == [
        "query_relevance",
        "platform_performance",
        "fifty_plus",
    ]
    assert [row["max_score"] for row in pack["scorecard"]["dimensions"]] == [50, 50, 100]
    assert [row["weight_percent"] for row in pack["scorecard"]["dimensions"]] == [50, 50, 30]
    profiles = pack["scorecard"]["score_weight_profiles"]["profiles"]
    by_id = {profile["profile_id"]: profile for profile in profiles}
    assert by_id["base_video_two_dimension_v1"]["weights"] == {
        "query_relevance": 50,
        "platform_performance": 50,
    }
    assert by_id["audience_50plus_available_v1"]["weights"] == {
        "query_relevance": 35,
        "platform_performance": 35,
        "fifty_plus": 30,
    }
    assert by_id["audience_50plus_skipped_by_query_gate_v1"]["weights"] == {
        "query_relevance": 35,
        "platform_performance": 35,
    }
    assert {
        alias
        for profile in profiles
        for alias in profile.get("historical_alias_ids", [])
    } == {
        "default_two_dimension",
        "douyin_fifty_plus_ok",
        "douyin_fifty_plus_not_attempted",
    }


def test_v4_contract_allows_extra_active_dimension_but_requires_base_dimensions():
    pkg = json.loads(
        (ROOT / "product_documents/规则包/douyin_rule_packs.v1.json").read_text(encoding="utf-8")
    )
    with_extra = deepcopy(pkg)
    scorecard = with_extra["rule_packs"][0]["scorecard"]
    scorecard["dimensions"].append(
        {
            "key": "fifty_plus_fit",
            "label": "50+ 匹配",
            "max_score": 20,
            "weight_percent": 20,
            "runtime_status": "active",
        }
    )
    scorecard["scoring_rules"].append(
        {
            "scoring_rule_id": "score_fifty_plus_precomputed",
            "dimension_key": "fifty_plus_fit",
        }
    )

    findings: list[dict[str, str]] = []
    _check_v4_rule_pack_contract(findings, with_extra)

    assert findings == []

    missing_base = deepcopy(pkg)
    missing_base["rule_packs"][0]["scorecard"]["dimensions"] = [
        row
        for row in missing_base["rule_packs"][0]["scorecard"]["dimensions"]
        if row["key"] != "platform_performance"
    ]
    findings = []
    _check_v4_rule_pack_contract(findings, missing_base)

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


def test_v4_contract_rejects_invalid_m2_score_weight_profiles():
    pkg = json.loads(
        (ROOT / "product_documents/规则包/douyin_rule_packs.v1.json").read_text(encoding="utf-8")
    )
    missing_profile = deepcopy(pkg)
    missing_profile["rule_packs"][0]["scorecard"]["score_weight_profiles"]["profiles"][2]["historical_alias_ids"] = []

    findings: list[dict[str, str]] = []
    _check_v4_rule_pack_contract(findings, missing_profile)

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

    unknown_dimension = deepcopy(pkg)
    unknown_dimension["rule_packs"][0]["scorecard"]["score_weight_profiles"]["profiles"][0]["weights"]["made_up_dimension"] = 10
    findings = []
    _check_v4_rule_pack_contract(findings, unknown_dimension)

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