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

from __future__ import annotations

from copy import deepcopy

from content_agent.business_modules.rule_judgment.evaluator import decide
from tests.test_rule_judgment_scorecard import _bundle, _policy


def test_v4_effect_status_mapping_for_pool_review_reject_technical_and_blocked():
    policy = _policy()

    success = decide("run_1", "policy_1", 1, _bundle(query=80, platform=70), policy)
    assert success["decision_action"] == "ADD_TO_CONTENT_POOL"
    assert success["search_query_effect_status"] == "success"
    assert success["decision_replay_data"]["effect_mapping_id"] == "map_add_to_pool_success"

    pending = decide("run_1", "policy_1", 2, _bundle(query=60, platform=60), policy)
    assert pending["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
    assert pending["search_query_effect_status"] == "pending"
    assert pending["decision_replay_data"]["effect_mapping_id"] == "map_keep_for_review_pending"

    failed = decide("run_1", "policy_1", 3, _bundle(query=54, platform=100), policy)
    assert failed["decision_action"] == "REJECT_CONTENT"
    assert failed["search_query_effect_status"] == "failed"
    assert failed["decision_replay_data"]["effect_mapping_id"] == "map_reject_failed"

    technical = decide("run_1", "policy_1", 4, _bundle(query=None, platform=70), policy)
    assert technical["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
    assert technical["decision_reason_code"] == "v4_technical_retry_needed"
    assert technical["search_query_effect_status"] == "failed"
    assert technical["decision_replay_data"]["effect_mapping_id"] == "map_technical_retry_failed"

    blocked_policy = deepcopy(policy)
    blocked_policy["rule_pack"]["hard_gates"] = [
        {
            "gate_id": "missing_source_evidence",
            "when": {"field": "source_evidence", "op": "is_empty"},
            "decision_action": "REJECT_CONTENT",
            "decision_reason_code": "missing_source_evidence",
            "severity": "fatal",
            "stop_scoring": True,
            "priority": 1,
        }
    ]
    blocked_policy["effect_status_mapping"].append(
        {
            "mapping_id": "map_reject_rule_blocked",
            "target_level": "content",
            "decision_action": "REJECT_CONTENT",
            "reason_category": "hard_gate",
            "is_hard_gate": True,
            "content_effect_status": "rule_blocked",
            "priority": 30,
            "enabled": True,
        }
    )
    blocked_bundle = _bundle(query=80, platform=70)
    blocked_bundle["source_evidence"] = {}

    blocked = decide("run_1", "policy_1", 5, blocked_bundle, blocked_policy)
    assert blocked["decision_reason_code"] == "missing_source_evidence"
    assert blocked["search_query_effect_status"] == "rule_blocked"
    assert blocked["decision_replay_data"]["effect_mapping_id"] == "map_reject_rule_blocked"