Jyotisha/tests/test_consultation_plan_contract.py

from __future__ import annotations

import pytest

from scripts.consultation_plan_contract import (
    ConsultationPlanContractError,
    apply_plan_precision_boundary,
    validate_consultation_plan_contract,
)


def timing_body(**overrides):
    body = {
        "plan_version": "consultation-plan-v2",
        "strict_workflow_route": "timing",
        "required_layers": ["Vimshottari", "Narayana", "Transit", "Varga", "negative holdout gate"],
        "claim_boundary": "candidate_day_month_window_only_until_holdout_passes",
        "plan_depth": "standard",
        "requested_domains": ["timing"],
        "timing_horizon": "next_12_months",
        "precision_boundary": "precise_timing_blocked",
        "required_evidence_categories": ["natal_foundation", "timing", "validation"],
    }
    body.update(overrides)
    return body


def test_versioned_plan_is_validated_against_server_route_allowlist():
    contract = validate_consultation_plan_contract(
        timing_body(),
        themes=["career"],
        route_packet={"question_type": "timing"},
    )

    assert contract == {
        "plan_version": "consultation-plan-v2",
        "strict_workflow_route": "timing",
        "plan_depth": "standard",
        "requested_domains": ["timing"],
        "timing_horizon": "next_12_months",
        "required_evidence_categories": ["natal_foundation", "timing", "validation"],
        "required_layers": ["Vimshottari", "Narayana", "Transit", "Varga", "negative holdout gate"],
        "claim_boundary": "candidate_day_month_window_only_until_holdout_passes",
        "precision_boundary": "precise_timing_blocked",
        "enforcement": "server_allowlist_validated",
    }


@pytest.mark.parametrize(
    ("override", "message"),
    [
        ({"plan_version": "consultation-plan-v1"}, "version"),
        ({"strict_workflow_route": "free_script"}, "route"),
        ({"required_layers": ["D1"]}, "required layers"),
        ({"claim_boundary": "exact_date_allowed"}, "claim boundary"),
        ({"requested_domains": ["career"]}, "requested domains"),
        ({"required_evidence_categories": ["natal_foundation"]}, "evidence categories"),
        ({"plan_depth": "unbounded"}, "depth"),
        ({"timing_horizon": None}, "requires a horizon"),
        ({"precision_boundary": "precise_timing_allowed"}, "precision boundary"),
    ],
)
def test_versioned_plan_rejects_free_form_or_tampered_workflow_control(override, message):
    with pytest.raises(ConsultationPlanContractError, match=message):
        validate_consultation_plan_contract(
            timing_body(**override),
            themes=["career"],
            route_packet={"question_type": "timing"},
        )


def test_versioned_plan_rejects_theme_and_resolved_route_mismatch():
    with pytest.raises(ConsultationPlanContractError, match="theme mismatch"):
        validate_consultation_plan_contract(
            timing_body(),
            themes=["marriage"],
            route_packet={"question_type": "timing"},
        )
    with pytest.raises(ConsultationPlanContractError, match="route mismatch"):
        validate_consultation_plan_contract(
            timing_body(),
            themes=["career"],
            route_packet={"question_type": "career"},
        )


def test_legacy_research_callers_without_plan_metadata_remain_compatible():
    assert validate_consultation_plan_contract(
        {"question": "事业如何"},
        themes=["career"],
        route_packet={"question_type": "career"},
    ) is None


@pytest.mark.parametrize(
    "partial_body",
    [
        {"strict_workflow_route": "timing"},
        {"required_layers": ["Vimshottari"]},
        {"claim_boundary": "candidate_day_month_window_only_until_holdout_passes"},
        {"plan_version": "consultation-plan-v2"},
    ],
)
def test_partial_plan_metadata_cannot_downgrade_to_legacy(partial_body):
    with pytest.raises(ConsultationPlanContractError, match="incomplete consultation plan metadata"):
        validate_consultation_plan_contract(
            partial_body,
            themes=["career"],
            route_packet={"question_type": "timing"},
        )


@pytest.mark.parametrize("missing_key", list(timing_body()))
def test_versioned_plan_requires_the_complete_metadata_set(missing_key):
    body = timing_body()
    body.pop(missing_key)
    with pytest.raises(ConsultationPlanContractError, match="incomplete consultation plan metadata"):
        validate_consultation_plan_contract(
            body,
            themes=["career"],
            route_packet={"question_type": "timing"},
        )


def test_blocked_plan_can_only_restrict_evidence_owned_precision_policy():
    context = {
        "answer_policy": {
            "can_answer_precise_timing": True,
            "should_lead_with_limitations": False,
        }
    }
    blocked = apply_plan_precision_boundary(context, {"precision_boundary": "precise_timing_blocked"})
    evidence_owned = apply_plan_precision_boundary(context, {"precision_boundary": "server_evidence_required"})

    assert blocked["answer_policy"]["can_answer_precise_timing"] is False
    assert blocked["answer_policy"]["should_lead_with_limitations"] is True
    assert evidence_owned == context


def test_api_dry_run_exposes_validated_plan_contract_and_rejects_tampering():
    from scripts.jyotish_api_server import BadRequest, JyotishAPIHandler

    handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
    body = {
        **timing_body(),
        "dry_run": True,
        "entry_mode": "direct_chart",
        "question": "应期与阶段问题：未来哪些阶段值得把握？",
        "theme": ["career"],
        "year": 1990,
        "month": 1,
        "day": 1,
        "hour": 12,
        "minute": 0,
        "lat": 25,
        "lon": 121,
        "tz": 8,
    }

    result = handler._compute_consultation_workflow(body)
    assert result["routing"]["question_type"] == "timing"
    assert result["consultation_plan_contract"]["enforcement"] == "server_allowlist_validated"

    with pytest.raises(BadRequest, match="claim boundary mismatch"):
        handler._compute_consultation_workflow({**body, "claim_boundary": "exact_date_allowed"})