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

from __future__ import annotations

import json
import shutil
from copy import deepcopy
from pathlib import Path

import pytest

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.policy_json import JsonPolicyBundleStore, _select_dispatch

RULE_PACK_JSON = Path("product_documents/规则包/douyin_rule_packs.v1.json")


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="dispatch not found for Content/video"):
        JsonPolicyBundleStore(root).load_policy_bundle("V1")


def test_dispatch_conflict_raises_config_error_with_rule_pack_ids(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(ContentAgentError) as exc_info:
        JsonPolicyBundleStore(root).load_policy_bundle("V1")
    error = exc_info.value
    assert error.error_code == ErrorCode.CONFIG_RULE_PACK_DISPATCH_CONFLICT
    assert "douyin_content_discovery_rule_pack_v1" in error.message
    assert error.detail["conflict_rule_pack_ids"] == [
        "douyin_content_discovery_rule_pack_v1",
        "douyin_content_discovery_rule_pack_v1",
    ]


def test_select_dispatch_still_returns_content_for_default_bundle():
    rule_package = json.loads(RULE_PACK_JSON.read_text(encoding="utf-8"))

    dispatch = _select_dispatch(rule_package, "V1")

    assert dispatch["dispatch_id"] == "dispatch_content"
    assert dispatch["target_entity"] == "Content"


def test_select_dispatch_can_select_non_content_when_enabled():
    rule_package = json.loads(RULE_PACK_JSON.read_text(encoding="utf-8"))
    rule_package = deepcopy(rule_package)
    author = next(d for d in rule_package["rule_pack_dispatch"] if d["target_entity"] == "Author")
    author["dispatch_enabled"] = True
    author["runtime_stage"] = "V1.0"
    author["strategy_version"] = "V1"

    dispatch = _select_dispatch(
        rule_package, "V1", target_entity="Author", content_format=author["content_format"]
    )

    assert dispatch["rule_pack_id"] == "douyin_author_expand_rule_pack_v1"


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

    assert bundle["target_entity"] == "Content"
    assert bundle["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
    assert bundle["rule_pack"] is bundle["rule_pack_by_entity"]["Content"]["rule_pack"]


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

    by_entity = bundle["rule_pack_by_entity"]
    assert set(by_entity) == {"Content"}
    assert by_entity["Content"]["dispatch"]["dispatch_id"] == "dispatch_content"
    assert by_entity["Content"]["rule_pack"]["rule_pack_id"] == bundle["rule_pack_id"]


def test_enabled_author_dispatch_can_be_found_by_entity_without_replacing_content_shim(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"))
    author = next(d for d in data["rule_pack_dispatch"] if d["target_entity"] == "Author")
    author["dispatch_enabled"] = True
    author["runtime_stage"] = "V1.0"
    author["strategy_version"] = "V1"
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

    bundle = JsonPolicyBundleStore(root).load_policy_bundle("V1")

    assert bundle["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
    assert set(bundle["rule_pack_by_entity"]) == {"Content", "Author"}
    author_pack = bundle["rule_pack_by_entity"]["Author"]["rule_pack"]
    assert author_pack["rule_pack_id"] == "douyin_author_expand_rule_pack_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",
    )
    (root / "product_documents/抖音游走策略").mkdir(parents=True)
    shutil.copy(
        "product_documents/抖音游走策略/douyin_walk_strategy.v1.json",
        root / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json",
    )
    return root