"""Native technique layers in the consultation output (TASK-consult-evidence-card-v2-20260927 T1). Real engine runs on a public AA chart (the evidence-card research charts). Each layer is compared with its native producer called directly, the annual pack with `build_annual_tajika_pack`, and the rest of the response with a run where the new layer is switched off: only the new key may differ. """ from __future__ import annotations import json import re import sys from datetime import datetime from pathlib import Path from types import SimpleNamespace import pytest ROOT = Path(__file__).resolve().parents[1] for path in (ROOT, ROOT / "scripts", ROOT / "scripts" / "research"): if str(path) not in sys.path: sys.path.insert(0, str(path)) import consult_evidence_card_run as runner # noqa: E402 from consult_evidence_card_lib import load_public_charts # noqa: E402 import scripts.consultation_native_layers as native # noqa: E402 QUESTIONS = { "marriage": "我的关系模式是什么？", "annual": "未来一年重点是什么？", "timing": "接下来一年哪些时间点重要？", } @pytest.fixture(scope="module", autouse=True) def _no_external_vedastro(): runner._block_external_vedastro() def _chart(index: int = 0) -> dict: return load_public_charts(ROOT)[index] def _run(domain: str, index: int = 0) -> dict: return runner._run_workflow(runner._workflow_body(_chart(index), { "id": domain, "domain": domain, "question": QUESTIONS[domain], })) def _layers(workflow: dict) -> dict: layers = workflow["chart"]["consultation_native_layers"] assert layers["schema"] == native.SCHEMA assert layers["status"] == "executed" return layers def _birth_args(workflow: dict, house: int) -> SimpleNamespace: birth = workflow["chart"]["birth"] body = _chart()["body"] return SimpleNamespace( year=body["year"], month=body["month"], day=body["day"], hour=body["hour"], minute=body["minute"], second=0, lat=body["lat"], lon=body["lon"], tz=float(body["tz"]), node_mode=birth["node_mode"], ayanamsa=birth["ayanamsa_name"], date=body["today"], house=house, ) def test_quick_gate_runs_this_file() -> None: import run_quality_gate assert "tests/test_consultation_native_layers.py" in run_quality_gate.CORE_PYTEST_TARGETS def test_every_layer_is_the_native_producers_own_output() -> None: workflow = _run("marriage") layers = _layers(workflow) chart = workflow["chart"] modules = chart["modules"] engine = native._load_engine() d9 = modules["varga_full"]["D9_Navamsa"] summary = layers["d9_summary"] assert summary["lagna"] == d9["ascendant"]["sign"] expected_vargottama = engine._calc_vargottama(chart["planets"], {"D9_Navamsa": d9["planets"]}) context = engine._build_dignity_context(d9["planets"]) for name, row in summary["planets"].items(): assert row["d9_sign"] == d9["planets"][name]["sign"] assert row["vargottama"] is expected_vargottama[name]["is_vargottama"] level = engine._get_dignity_level(name, d9["planets"][name]["sign"], d9["planets"][name]["degree"], context) assert row["d9_dignity"] == engine.DIGNITY_LABELS[level] assert summary["vargottama"] == [name for name, row in summary["planets"].items() if row["vargottama"]] assert len(summary["planets"]) == 9 from punarphoo import detect_natal_punarphoo natal = detect_natal_punarphoo(chart["planets"]) assert layers["punarphoo"]["status"] == "observation_only" assert layers["punarphoo"]["natal_present"] == natal["natal_present"] assert layers["punarphoo"]["orb_deg"] == natal["orb_deg"] assert "not marriage" in layers["punarphoo"]["claim_boundary"] vivah = engine._calc_vivah_saham( {name: {"degree": row["lon"]} for name, row in chart["planets"].items()}, chart["ascendant"]["lon"], ) assert layers["vivah_saham"]["longitude"] == vivah["saham_lon"] assert layers["vivah_saham"]["house"] == vivah["saham_house"] assert layers["vivah_saham"]["sign"] == vivah["saham_sign"] assert layers["day_night"]["is_day"] is modules["gulika"]["daynight_evidence"]["is_daytime"] seventh = engine.cmd_double_transit_pac(_birth_args(workflow, 7)) double = layers["double_transit"] assert double["house_7_summary"] == seventh["summary"] assert [row["target"] for row in double["conclusions"] if row["house"] == 7] == [ row["target"] for row in seventh["double_transit"] ] assert set(double["karaka_targets"]) == {"DK", "UL"} assert double["karaka_targets"]["DK"]["target"] == f"DK({modules['jaimini']['chara_karakas']['DK']['planet']})" assert double["karaka_targets"]["UL"]["target"] == f"UL({modules['arudha_padas']['padas']['UL']['sign']})" # Conclusions, not raw degrees. assert not re.search(r"\d+\.\d+°", json.dumps(double["conclusions"], ensure_ascii=False)) slow = layers["slow_transits"] assert set(slow["current"]) == {"Jupiter", "Saturn", "Rahu", "Ketu"} sav = modules["ashtakavarga"]["sav"]["scores"] for body in ("Jupiter", "Saturn"): row = slow["current"][body] assert row["sav"] == sav[row["sign"]] assert row["bav"] == modules["ashtakavarga"]["bav"][body]["bindus"][native.SIGNS.index(row["sign"])] from ephemeris_events import build_ephemeris_events scan = build_ephemeris_events({ "start_date": slow["window"]["start"], "end_date": slow["window"]["end"], "ayanamsa": chart["birth"]["ayanamsa_name"], "bodies": ["jupiter", "saturn"], }) expected = [(row["kind"], row["date"], row["body"]) for row in scan["events"]] got = [(row["kind"], row["date"], row["body"]) for row in slow["key_dates"] if row["body"] in ("jupiter", "saturn")] assert got == expected assert slow["houses_in_window"]["Jupiter"][0] == slow["current"]["Jupiter"]["natal_house"] ak = modules["jaimini"]["chara_karakas"]["AK"]["planet"] assert layers["karakamsha"]["atmakaraka"] == ak assert layers["karakamsha"]["karakamsha_sign"] == d9["planets"][ak]["sign"] assert layers["argala"]["summary"]["supported_count"] == len(layers["argala"]["summary"]["supported_houses"]) assert set(layers["dispositor_chains"]["chains"]) >= {"Sun", "Moon"} assert layers["inter_chart_linkage"]["planets"]["Sun"]["D9"]["sign"] == d9["planets"]["Sun"]["sign"] assert layers["moon_transit"]["sign"] in native.SIGNS assert layers["annual_tajika"]["status"] == "not_applicable" def test_annual_route_attaches_the_annual_pack_verbatim() -> None: workflow = _run("annual") annual = _layers(workflow)["annual_tajika"] assert annual["status"] == "parameter_sensitive" assert "parameter_sensitive" in annual["boundary"] from annual_tajika_pack import build_annual_tajika_pack body = _chart()["body"] payload = native._annual_payload({**body, "tz": float(body["tz"]), "ayanamsa": "raman", "node_mode": "mean"}, annual["target_year"]) pack = build_annual_tajika_pack(payload) assert annual["annual_lagna"] == pack["annual_chart"]["data"]["asc_sign"] assert annual["varshesha"] == pack["year_lord"]["data"]["year_lord"] assert annual["muntha"]["sign"] == pack["muntha"]["data"]["muntha_sign"] assert annual["muntha"]["lord"] == pack["muntha"]["data"]["muntha_lord"] periods = pack["mudda_dasha"]["periods"] assert [(row["lord"], row["months"]) for row in annual["mudda_dasha"]] == [(row["lord"], row["months"]) for row in periods] segments = annual["mudda_dasha"] assert segments[0]["start"] == annual["varsha_year"]["start"] assert segments[-1]["end"] == annual["varsha_year"]["end"] for before, after in zip(segments, segments[1:]): assert before["end"] == after["start"] assert annual["varsha_year"]["start"] <= body["today"] < annual["varsha_year"]["end"] focus = {"Sun", "Moon", annual["varshesha"]} assert annual["tajika_aspects"], "the annual chart has Tajika aspects for the Sun, Moon or year lord" for row in annual["tajika_aspects"]: assert focus.intersection(row["planets"]) assert row["name"] in {"Ithasala_candidate", "Easarapha_candidate"} def test_an_unreliable_annual_pack_is_blocked_and_never_filled_with_natal_data(monkeypatch) -> None: import annual_tajika_pack real = annual_tajika_pack.build_annual_tajika_pack def conflicted(payload, **kwargs): pack = real(payload, **kwargs) pack["muntha"] = {"status": "conflict", "field": "muntha", "reason": "same_profile_annual_producers_disagree"} return pack monkeypatch.setattr(annual_tajika_pack, "build_annual_tajika_pack", conflicted) annual = _layers(_run("annual"))["annual_tajika"] assert annual["status"] == "blocked" assert "muntha" in annual["reason"] for key in ("annual_lagna", "varshesha", "muntha", "mudda_dasha", "tajika_aspects"): assert key not in annual def test_a_failing_producer_blocks_its_own_layer_only(monkeypatch) -> None: engine = native._load_engine() def broken(*args, **kwargs): raise RuntimeError("fixture") monkeypatch.setattr(engine, "cmd_double_transit_pac", broken) workflow = _run("timing") layers = _layers(workflow) assert workflow["success"] is True assert layers["double_transit"] == {"status": "blocked", "reason": "RuntimeError"} assert layers["d9_summary"]["status"] == "executed" assert layers["slow_transits"]["status"] == "executed" def test_the_new_layer_leaves_every_existing_output_unchanged(monkeypatch) -> None: # The first run of a chart fills the chart cache; a cache hit re-reads the # stored chart, whose VedAstro request manifest and key order differ from a # fresh compute. Compare two cache-hit runs. _run("marriage", index=1) with_layers = _run("marriage", index=1) monkeypatch.setattr(native, "attach_native_consultation_layers", lambda chart, **_: chart) without = _run("marriage", index=1) assert "consultation_native_layers" not in without["chart"] volatile = {"runtime_cache", "vedastro_gateway"} def strip(value): if isinstance(value, dict): return {key: strip(item) for key, item in value.items() if key not in volatile and key != "consultation_native_layers"} if isinstance(value, list): return [strip(item) for item in value] return value assert json.dumps(strip(with_layers), sort_keys=True, default=str) == json.dumps(strip(without), sort_keys=True, default=str) def test_mudda_segment_dates_split_the_solar_year_by_the_native_months() -> None: start, end = datetime(2026, 2, 25), datetime(2027, 2, 25) rows = native._mudda_segments([{"lord": "Sun", "months": 6.0}, {"lord": "Moon", "months": 6.0}], start, end) assert rows == [ {"lord": "Sun", "months": 6.0, "start": "2026-02-25", "end": "2026-08-26"}, {"lord": "Moon", "months": 6.0, "start": "2026-08-26", "end": "2027-02-25"}, ]