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

from copy import deepcopy
from pathlib import Path

from content_agent.business_modules.rule_judgment.evaluator import decide
from content_agent.errors import ErrorCode
from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest
from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE


def test_rule_pack_thresholds_drive_decision(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    run_id = state["run_id"]

    policy_bundle = deepcopy(state["policy_bundle"])
    thresholds = policy_bundle["rule_pack"]["thresholds"]
    thresholds[0]["min_score"] = 80
    thresholds[1]["min_score"] = 70
    thresholds[1]["max_score"] = 79

    # 2-dim score(2026-06-12 贴题档位 45/25→40/20):relevance 0.6 -> 40, platform_heat 0.6 -> 30
    # => 70, lands in the reconfigured 70<=score<=79 review band.
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["relevance_score"] = 0.6
    bundle["content_engagement_metrics"]["platform_heat"] = 0.6

    decision = decide(run_id, state["policy_run_id"], 1, bundle, policy_bundle)
    assert decision["score"] == 70
    assert decision["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
    assert decision["decision_reason_code"] == "content_score_review"
    assert decision["decision_replay_data"]["matched_threshold"] == "70<=score<=79"
    assert decision["policy_run_id"] == state["policy_run_id"]
    assert decision["strategy_version"] == "V1"


def test_rule_pack_hard_gate_reason_code_drives_decision(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    run_id = state["run_id"]

    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["source_evidence"] = {}

    decision = decide(run_id, state["policy_run_id"], 1, bundle, state["policy_bundle"])
    assert decision["decision_action"] == "REJECT_CONTENT"
    assert decision["decision_reason_code"] == "missing_source_evidence"
    assert decision["search_query_effect_status"] == "rule_blocked"
    assert decision["triggered_blocking_rules"] == ["missing_source_evidence"]
    assert decision["decision_replay_data"]["primary_reason_code"] == "missing_source_evidence"


def test_missing_score_uses_rule_pack_missing_policy(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    run_id = state["run_id"]

    bundle = deepcopy(state["evidence_bundles"][0])
    # No evidence for either active dim -> rule pack's score_missing_policy applies.
    bundle["pattern_match_result"].pop("relevance_score", None)
    bundle["content_engagement_metrics"].pop("platform_heat", None)

    decision = decide(run_id, state["policy_run_id"], 1, bundle, state["policy_bundle"])
    assert decision["decision_action"] == "REJECT_CONTENT"
    assert decision["decision_reason_code"] == "missing_score"
    assert decision["search_query_effect_status"] == "failed"
    assert decision["decision_replay_data"]["score_missing_policy"]["decision_reason_code"] == "missing_score"


def test_unknown_strategy_version_fails_before_rule_decision(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )

    state = service.start_run(
        RunStartRequest(
            platform_mode="mock",
            strategy_version="missing_strategy",
            source=str(REAL_SOURCE_FIXTURE),
        )
    )

    assert state["status"] == "failed"
    assert state["error_code"] == ErrorCode.POLICY_BUNDLE_NOT_FOUND.value
    assert state["errors"] == ["policy bundle not found"]


def test_rule_judgment_does_not_call_pattern_recall_integrations():
    root = Path("content_agent/business_modules/rule_judgment")
    text = "\n".join(path.read_text(encoding="utf-8") for path in root.rglob("*.py"))

    assert "decode_api" not in text
    assert "category_match" not in text
    assert "match-paths" not in text


def test_disabled_future_packs_not_marked_as_active_entity_packs():
    from content_agent.integrations.policy_json import JsonPolicyBundleStore

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

    assert set(bundle["rule_pack_by_entity"]) == {"Content"}
    for entity in ("Author", "Hashtag", "Path", "Budget"):
        assert entity not in bundle["rule_pack_by_entity"]