Jyotisha/tests/test_consult_evidence_card_research.py

"""Pure-function checks for the consultation evidence-card research scripts."""

from __future__ import annotations

import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts" / "research"))

from consult_evidence_card_lib import (  # noqa: E402
    CARD_SPECS,
    CATEGORIES,
    TELEMETRY_SPEC,
    account,
    category_report,
    cut_evidence_card,
    dumps,
    parse_methodology_routes,
    parse_must_use_layers,
    parse_registry_layers,
    verbatim_check,
)


def test_category_partition_sums_exactly():
    payload = {
        "ascendant": {"sign": "Leo", "degree": 12.5},
        "methodology": {"sections": ["checklist"]},
        "western_spectrum": {"zodiac": "tropical", "natal": {"sun": "Aries"}},
        "varga_spectrum": {
            "formal": {"D1": {"lagna": "Leo"}},
            "research_dn": {"D3": {"lagna": "Virgo"}},
            "extended": {"D81": {"lagna": "Cancer"}},
        },
        "technique_audit_table": [
            {"technique": "Functional Benefic/Malefic", "status": "executed"},
            {"technique": "MEVG / Global Web Evidence", "status": "blocked"},
            {"technique": "Prashna chart", "status": "not_applicable"},
            {"technique": "Research D-N through D60", "status": "executed"},
            {"technique": "Western natal (tropical)", "status": "executed", "system": "western"},
        ],
    }
    report = category_report(payload)
    assert report["sum_error_ratio"] == 0.0
    assert report["category_sum"] == report["total_chars"] == len(dumps(payload))
    assert set(report["categories"]) == set(CATEGORIES)
    assert report["categories"]["western"] > 0
    assert report["categories"]["research"] > 0
    assert report["categories"]["status"] > 0
    assert report["categories"]["not_applicable"] > 0
    assert report["categories"]["core"] > 0
    assert sum(account(payload).values()) == len(dumps(payload))


def test_card_copies_projection_values_without_rewriting():
    model = {
        "question": "我和父母关系如何",
        "route": "family",
        "claim_cards": [
            {
                "category": "natal_foundation",
                "evidence": {
                    "ascendant": {"sign": "Capricorn", "degree": 3.25},
                    "houses": [{"number": 4, "sign": "Aries", "lord": "Mars"}],
                    "planets": {"sun": {"sign": "Aquarius", "house": 2}, "moon": {"sign": "Cancer", "house": 7}},
                    "functional_benefic_malefic": {"functional_benefics": ["Venus"], "functional_malefics": ["Mars"]},
                    "varga_spectrum": {"formal": {"D12": {"lagna": "Pisces", "planets": {"moon": "Leo"}}}},
                },
            },
            {
                "category": "timing",
                "evidence": {
                    "dasha_sub_periods": {
                        "current": {
                            "mahadasha": {"lord": "Saturn", "start": "2020-01-01", "end": "2026-12-31"},
                            "antardasha": {"lord": "Mercury", "start": "2024-01-01", "end": "2025-01-01"},
                        },
                    },
                    "narayana_dasha": {"current_dasha": {"md": {"sign": "Scorpio"}}},
                },
            },
        ],
    }
    cut = cut_evidence_card(model, "parents")
    assert cut["card"]["domain"]["vargas"]["D12"]["lagna"] == "Pisces"
    assert cut["card"]["base"]["ascendant"]["degree"] == 3.25
    assert "D7" not in cut["card"]["domain"]["vargas"]
    check = verbatim_check(cut["card"], {
        "ascendant_sign": "Capricorn",
        "d12_lagna": "Pisces",
        "d12_moon": "Leo",
        "vimshottari_start": "2020-01-01",
        "vimshottari_end": "2026-12-31",
        "moon_sign": "Cancer",
        "moon_house": 7,
    })
    assert check["ok"] is True
    assert check["checks"]["vimshottari_start"]["status"] == "match"


def test_source_parsers_keep_family_split_gap():
    registry = (ROOT / "frontend/src/lib/consultation-domain-registry.ts").read_text(encoding="utf-8")
    methodology = (ROOT / "frontend/src/lib/consultation-methodology.ts").read_text(encoding="utf-8")
    workflow = (ROOT / "frontend/src/mastra/consultation-workflow.ts").read_text(encoding="utf-8")
    layers = parse_registry_layers(registry)
    routes = parse_methodology_routes(methodology)
    must_use = parse_must_use_layers(workflow)
    assert "D12" in layers["family"] and "D7" in layers["family"]
    assert "父母" not in "".join(layers["family"])
    assert routes["family"] is None
    assert "family" not in must_use
    assert CARD_SPECS["parents"]["vargas"] == ["D12"]
    assert CARD_SPECS["children"]["vargas"] == ["D7"]


def test_telemetry_spec_stores_no_user_text():
    banned = {"question text", "answer text", "birth date", "name", "email", "user id"}
    assert banned <= set(TELEMETRY_SPEC["does_not_store"])
    stored = {field["name"] for field in TELEMETRY_SPEC["fields"]}
    assert stored == {"domain", "card_version", "card_chars", "card_token_estimate", "cited_field_ids", "feedback"}
    assert "matched substring" in TELEMETRY_SPEC["does_not_store"]


def test_single_domain_wrap_source_still_matches_projector():
    source = (ROOT / "frontend/src/mastra/consultation-tools.ts").read_text(encoding="utf-8")
    assert "return { ...consultations[0], ...plan, consultations };" in source