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

from __future__ import annotations

import json
import shutil
from pathlib import Path

import pytest

from content_agent.integrations.policy_json import JsonPolicyBundleStore


def test_policy_bundle_uses_content_dispatch_and_exports_runtime_contracts():
    bundle = JsonPolicyBundleStore().load_policy_bundle("V1")

    assert bundle["dispatch_id"] == "dispatch_content"
    assert bundle["runtime_stage"] == "V1.0"
    assert bundle["target_entity"] == "Content"
    assert bundle["content_format"] == "video"
    assert bundle["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
    assert bundle["effect_status_mapping"]
    assert bundle["query_effect_aggregation"]
    assert bundle["runtime_status_contract"]["query_effect_status"] == [
        "success",
        "pending",
        "failed",
        "rule_blocked",
    ]
    assert bundle["policy_bundle_hash"]


def test_policy_bundle_fails_when_content_dispatch_is_missing(tmp_path):
    root = _copy_policy_files(tmp_path)
    path = root / "product_documents/规则包/douyin_rule_packs.v1.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    for dispatch in data["rule_pack_dispatch"]:
        if dispatch["dispatch_id"] == "dispatch_content":
            dispatch["dispatch_enabled"] = False
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

    with pytest.raises(ValueError, match="expected exactly one Content dispatch"):
        JsonPolicyBundleStore(root).load_policy_bundle("V1")


def test_policy_bundle_fails_when_content_dispatch_is_ambiguous(tmp_path):
    root = _copy_policy_files(tmp_path)
    path = root / "product_documents/规则包/douyin_rule_packs.v1.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    duplicate = dict(data["rule_pack_dispatch"][0])
    duplicate["dispatch_id"] = "dispatch_content_duplicate"
    duplicate["priority"] = 2
    data["rule_pack_dispatch"].append(duplicate)
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

    with pytest.raises(ValueError, match="expected exactly one Content dispatch"):
        JsonPolicyBundleStore(root).load_policy_bundle("V1")


def test_policy_bundle_fails_when_dispatch_points_to_missing_rule_pack(tmp_path):
    root = _copy_policy_files(tmp_path)
    path = root / "product_documents/规则包/douyin_rule_packs.v1.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    for dispatch in data["rule_pack_dispatch"]:
        if dispatch["dispatch_id"] == "dispatch_content":
            dispatch["rule_pack_id"] = "missing_rule_pack"
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

    with pytest.raises(ValueError, match="dispatch dispatch_content matched 0 enabled rule packs"):
        JsonPolicyBundleStore(root).load_policy_bundle("V1")


def test_policy_bundle_fails_when_dispatch_points_to_disabled_rule_pack(tmp_path):
    root = _copy_policy_files(tmp_path)
    path = root / "product_documents/规则包/douyin_rule_packs.v1.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    for rule_pack in data["rule_packs"]:
        if rule_pack["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1":
            rule_pack["enabled"] = False
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

    with pytest.raises(ValueError, match="dispatch dispatch_content matched 0 enabled rule packs"):
        JsonPolicyBundleStore(root).load_policy_bundle("V1")


def _copy_policy_files(tmp_path: Path) -> Path:
    root = tmp_path / "repo"
    (root / "product_documents/规则包").mkdir(parents=True)
    shutil.copy(
        "product_documents/规则包/douyin_rule_packs.v1.json",
        root / "product_documents/规则包/douyin_rule_packs.v1.json",
    )
    return root