Jyotisha/tests/test_consultation_workflow_domains.py

#!/usr/bin/env python3
"""Focused consultation-workflow coverage for canonical domain adapters."""

from __future__ import annotations

import pytest

from scripts.consultation_domain_registry import CANONICAL_DOMAINS
from scripts.consultation_plan_contract import plan_route_contract
from scripts.jyotish_api_server import BadRequest, JyotishAPIHandler
from scripts.report_orchestrator import (
    INDEPENDENT_REPORT_THEMES,
    ThemeName,
    consultation_thematic_capabilities,
)


_NATIVE_THEMATIC_DOMAINS = {"career", "marriage", "wealth", "health"}


def _handler() -> JyotishAPIHandler:
    return JyotishAPIHandler.__new__(JyotishAPIHandler)


def _stub_consultation_runtime(monkeypatch, handler: JyotishAPIHandler) -> None:
    chart = {
        "success": True,
        "birth_info": {"date": "1997-08-08", "time": "05:00", "tz": 8},
        "planets": {},
        "ascendant": {},
        "modules": {},
        "ai_prompt_pack": {
            "evidence_snapshot": {
                "strict_workflow_contracts": {
                    domain: {"status": "available", "domain": domain}
                    for domain in CANONICAL_DOMAINS
                }
            }
        },
    }
    monkeypatch.setattr(handler, "_compute_chart", lambda body: chart)
    monkeypatch.setattr(
        handler,
        "_compute_rectification_gate",
        lambda body: {
            "success": True,
            "endpoint": "rectification_gate",
            "summary": {"recommended_events": [], "warned": [], "disabled": []},
        },
    )
    monkeypatch.setattr(
        handler,
        "_compute_muhurta_panchanga",
        lambda body: {"status": "ok", "scope": "muhurta_panchanga"},
    )


@pytest.mark.parametrize("domain", CANONICAL_DOMAINS)
def test_complete_consultation_workflow_handles_every_canonical_domain(
    monkeypatch,
    domain: str,
) -> None:
    handler = _handler()
    _stub_consultation_runtime(monkeypatch, handler)

    result = handler._compute_consultation_workflow({
        "entry_mode": "direct_chart",
        "question": domain,
        "theme": [domain],
        "year": 1997,
        "month": 8,
        "day": 8,
        "hour": 5,
        "minute": 0,
        "lat": 36.420487,
        "lon": 114.209936,
        "tz": 8,
        "western_mode": False,
        "defer_optional_external_evidence": True,
    })

    thematic = result["thematic_report"]
    domain_report = thematic["themes"][domain]
    adapter = domain_report["thematic_adapter"]

    assert result["success"] is True
    assert result["routes"] == list(CANONICAL_DOMAINS)
    assert thematic["available_themes"] == list(CANONICAL_DOMAINS)
    assert domain_report["theme"] == domain
    assert adapter["fallback_theme"] is None
    if domain in _NATIVE_THEMATIC_DOMAINS:
        assert domain_report["status"] == "supported"
        assert adapter["report_theme"] == domain
    else:
        assert domain_report["status"] == "degraded"
        assert adapter["capability_status"] == "blocked"
        assert adapter["report_theme"] is None
        assert adapter["upstream_contract_available"] is True
        assert "No general-theme fallback" in domain_report["boundary"]


@pytest.mark.parametrize("domain", CANONICAL_DOMAINS)
def test_versioned_plan_domains_execute_their_declared_route_not_the_question_text_route(
    monkeypatch,
    domain: str,
) -> None:
    """The product runtime sends one question per domain, so text routing agrees with at most one."""
    handler = _handler()
    _stub_consultation_runtime(monkeypatch, handler)
    contract = plan_route_contract(domain)
    assert contract is not None

    result = handler._compute_consultation_workflow({
        "entry_mode": "direct_chart",
        "question": "我的事业和财运接下来会怎么走，两者之间该怎么取舍",
        "theme": [domain],
        "plan_version": "consultation-plan-v2",
        "strict_workflow_route": domain,
        "required_layers": list(contract.required_layers),
        "claim_boundary": contract.claim_boundary,
        "plan_depth": "standard",
        "requested_domains": list(contract.requested_domains),
        "timing_horizon": "next_12_months" if domain in {"timing", "annual"} else None,
        "precision_boundary": "server_evidence_required",
        "required_evidence_categories": list(contract.required_evidence_categories),
        "year": 1997,
        "month": 8,
        "day": 8,
        "hour": 5,
        "minute": 0,
        "lat": 36.420487,
        "lon": 114.209936,
        "tz": 8,
        "western_mode": False,
        "defer_optional_external_evidence": True,
    })

    primary_theme = result["routing"]["primary_theme"]
    assert result["success"] is True
    assert result["routing"]["question_type"] == domain
    assert primary_theme == domain
    assert result["consumer_context"]["route"] == domain
    # The consumer reads thematic_report.themes[primary_theme] as this domain's evidence.
    assert result["thematic_report"]["themes"][primary_theme]["theme"] == domain


def test_consultation_thematic_capabilities_are_registry_derived_and_exclude_spirituality() -> None:
    capabilities = consultation_thematic_capabilities()

    assert tuple(capabilities) == CANONICAL_DOMAINS
    assert "spirituality" not in capabilities
    assert INDEPENDENT_REPORT_THEMES == (ThemeName.SPIRITUALITY,)
    assert capabilities["general"]["fallback_theme"] is None


@pytest.mark.parametrize("domain", ["spirituality", "muhurta", "unknown-domain"])
def test_consultation_workflow_rejects_independent_or_unknown_domains(domain: str) -> None:
    handler = _handler()

    with pytest.raises(BadRequest, match="Unknown high-rigor theme"):
        handler._compute_consultation_workflow({
            "entry_mode": "direct_chart",
            "question": domain,
            "theme": [domain],
            "year": 1997,
            "month": 8,
            "day": 8,
            "hour": 5,
            "minute": 0,
            "lat": 36.420487,
            "lon": 114.209936,
            "tz": 8,
        })


def test_independent_spirituality_report_remains_available_outside_consultation_registry() -> None:
    result = _handler()._compute_thematic_report({"theme": "spirituality"})

    spirituality = result["themes"]["spirituality"]
    assert result["success"] is True
    assert "spirituality" not in result["available_themes"]
    assert result["independent_report_themes"] == ["spirituality"]
    assert spirituality["status"] == "supported"
    assert spirituality["thematic_adapter"]["scope"] == "independent_report"