4d2506c8c1
The report used the same nine-year rule as Shoola. Start from the stronger second or eighth sign and list mahadashas only. Co-Authored-By: Grok (agent loop) Co-authored-by: Cursor <cursoragent@cursor.com>
247 lines
11 KiB
Python
247 lines
11 KiB
Python
"""full-reading keeps the upstream module keys and does not go online."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from scripts.jaimini import SIGNS
|
||
from scripts.jyotish_engine import cmd_full_reading
|
||
from scripts.professional_report_reference import _export_args
|
||
from scripts.punarphoo import detect_punarphoo
|
||
from scripts.rashi_dashas_native import SIGN_LORDS, navamsha_sign
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
# 原值：docs/research/upstream_full_synastry_bde5d615/.../validate_full_synastry_packets.py.txt
|
||
# 新值：docs/research/upstream_full_synastry_7a773d13/.../validate_full_synastry_packets.py.txt
|
||
# 原因：修改单 M3。新校验器把 Niryaana Shoola 固定判为 not_applicable，合盘不用这套大运。
|
||
VALIDATOR = (
|
||
ROOT
|
||
/ "docs/research/upstream_full_synastry_7a773d13/skill/scripts/validate_full_synastry_packets.py.txt"
|
||
)
|
||
|
||
# Baseline full-reading keys on origin/staging b3ef2d815, plus the five keys this
|
||
# change adds. 81 + 5 = the upstream 86-key set.
|
||
UPSTREAM_MODULE_KEYS = frozenset({
|
||
"actionable_context", "argala", "ashtakavarga", "ashtottari_dasha", "aspects",
|
||
"audit", "avasthas", "bhava_bala", "bhava_chalit", "bhrigu_pada_dasha",
|
||
"career_strict_evidence", "chart", "chaturshitisama_dasha", "congregation",
|
||
"d9_navamsa_expanded", "dasa_convergence", "dasha", "dasha_sandhi",
|
||
"dispositor_chains", "double_transit_pac", "dwadashottari_dasha",
|
||
"dwisaptatisama_dasha", "final_dispositors", "finance_strict_evidence",
|
||
"functional_benefic_malefic", "guided_topics", "house_map", "inter_chart_linkage",
|
||
"jaimini", "kalachakra_dasha", "kp", "lagna_vargottamamsa", "marriage_counting",
|
||
"moon_chart", "muntha", "nakshatra_advanced", "nakshatra_dasha", "narayana_dasha",
|
||
"navamsha_dasha", "niryaana_shoola_dasha", "pancha_pakshi", "panchanga",
|
||
"panchottari_dasha", "planetary_congregation", "planetary_friendship", "prashna",
|
||
"punarphoo", "pushkara", "pushkara_planets", "rashi_tulya_navamsa",
|
||
"relationship_strict_evidence", "remedies", "sahams", "satabdika_dasha",
|
||
"sensitive_points", "shadbala", "shastihayani_dasha", "shattrimshatsama_dasha",
|
||
"shodashottari_dasha", "solar_return", "source_priority", "special_lagnas",
|
||
"sudarshana", "tajika", "tajika_yogas", "tithi_lord", "transit_ll7l",
|
||
"transit_multi_reference", "transit_positions", "tribhagi_40_dasha",
|
||
"tribhagi_dasha", "trimshamsa_d30", "unified_western_reading", "upagrahas",
|
||
"validation", "varga_extended", "varga_full", "varga_research_high", "vargottama",
|
||
"vedastro_official_full_snapshot", "vedastro_range_scan_result", "vimsopaka",
|
||
"vivah_saham", "yoga", "yogas_doshas", "yogini_dasha",
|
||
})
|
||
|
||
|
||
def _case(path: Path, case_id: str) -> dict:
|
||
rows = json.loads(path.read_text(encoding="utf-8"))
|
||
birth = next(row for row in rows if row["id"] == case_id)
|
||
return {
|
||
"year": birth["year"],
|
||
"month": birth["month"],
|
||
"day": birth["day"],
|
||
"hour": birth["hour"],
|
||
"minute": birth["minute"],
|
||
"second": birth.get("second") or 0,
|
||
"lat": birth["lat"],
|
||
"lon": birth["lon"],
|
||
"tz": birth["tz"],
|
||
"ayanamsa": "lahiri",
|
||
"node_mode": "mean",
|
||
"today": "2026-10-18",
|
||
}
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def reports():
|
||
cases = {
|
||
"pitt": _case(ROOT / "tests/indastro_cases.json", "brad_pitt"),
|
||
"jolie": _case(ROOT / "tests/celebrity_cases.json", "jolie"),
|
||
}
|
||
return {name: cmd_full_reading(_export_args(birth)) for name, birth in cases.items()}
|
||
|
||
|
||
def _validator():
|
||
from importlib.machinery import SourceFileLoader
|
||
|
||
loader = SourceFileLoader("validate_full_synastry_packets", str(VALIDATOR))
|
||
spec = importlib.util.spec_from_loader(loader.name, loader)
|
||
module = importlib.util.module_from_spec(spec)
|
||
loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
def _goel_method_1_first(d9_lagna: int, lord_sign_by_sign: dict[int, int]) -> tuple[int, int]:
|
||
start = d9_lagna if d9_lagna % 2 == 0 else (d9_lagna + 6) % 12
|
||
lord_sign = int(lord_sign_by_sign[start]) % 12
|
||
if start % 2 == 0:
|
||
distance = (lord_sign - start) % 12
|
||
else:
|
||
distance = (start - lord_sign) % 12
|
||
years = 12 if distance == 0 else distance + 1
|
||
return start, years
|
||
|
||
|
||
def test_unified_western_reading_module_imports() -> None:
|
||
assert not (ROOT / "scripts" / "_script_root.py").exists()
|
||
import classical_astrology
|
||
import unified_western_reading
|
||
|
||
assert callable(unified_western_reading.build_unified_western_reading)
|
||
assert callable(classical_astrology.build_classical_astrology_packet)
|
||
|
||
|
||
def test_engine_does_not_import_the_retired_niryaana_file() -> None:
|
||
source = (ROOT / "scripts/jyotish_engine.py").read_text(encoding="utf-8")
|
||
assert "from niryaana_shoola_dasha import" not in source
|
||
assert "import niryaana_shoola_dasha" not in source
|
||
|
||
|
||
def test_full_reading_module_keys_cover_the_upstream_set(reports) -> None:
|
||
assert len(UPSTREAM_MODULE_KEYS) == 86
|
||
for name, report in reports.items():
|
||
missing = UPSTREAM_MODULE_KEYS - set(report["modules"])
|
||
assert not missing, name
|
||
|
||
|
||
def test_simple_modules_match_the_upstream_shapes(reports) -> None:
|
||
for report in reports.values():
|
||
modules = report["modules"]
|
||
lagna = modules["lagna_vargottamamsa"]
|
||
assert set(lagna) == {"d1_sign", "d9_sign", "is_lagna_vargottamamsa", "source"}
|
||
assert lagna["source"] == "raman_horoscope_v1_bounded_pp64"
|
||
assert lagna["is_lagna_vargottamamsa"] is bool(
|
||
lagna["d1_sign"] and lagna["d9_sign"] and lagna["d1_sign"] == lagna["d9_sign"]
|
||
)
|
||
flags = modules["pushkara"]
|
||
expected = [
|
||
f"{planet} ({data['sign']} {data['degree_in_sign']:.1f}°)"
|
||
for planet, data in flags.items()
|
||
if data.get("pushkara_navamsa") or data.get("pushkara_bhaga")
|
||
]
|
||
assert modules["pushkara_planets"] == expected
|
||
assert modules["punarphoo"] == detect_punarphoo(report["chart"]["planets"])
|
||
|
||
|
||
def test_secondary_dashas_use_native_profiles_and_match_goel_method_1(reports) -> None:
|
||
for report in reports.values():
|
||
modules = report["modules"]
|
||
navamsha = modules["navamsha_dasha"]
|
||
niryaana = modules["niryaana_shoola_dasha"]
|
||
assert navamsha["status"] == "parameter_sensitive"
|
||
assert navamsha["profile_id"] == "goel_method_1_v1"
|
||
# 原值：niryaana == {execution_status: blocked, reason: 口径更正中，见 TASK-niryaana-shoola-pvr-20261018}
|
||
# 新值：status parameter_sensitive，profile pvr_niryaana_shoola_v1，periods 是 PVR 主运
|
||
# 原因：#84 用真结果换掉 #81 的占位。合盘校验器仍把这一行固定判为不适用。
|
||
assert niryaana["status"] == "parameter_sensitive"
|
||
assert niryaana["profile_id"] == "pvr_niryaana_shoola_v1"
|
||
assert niryaana["periods"] == niryaana["native_result"]["periods"]
|
||
assert niryaana["native_result"]["status"] == "listed"
|
||
assert "antardasha" not in niryaana["periods"][0]
|
||
assert sum(row["years"] for row in niryaana["periods"]) == 96
|
||
assert navamsha["native_result"]["status"] == "listed"
|
||
first = navamsha["periods"][0]
|
||
d9 = modules["varga_full"]["D9_Navamsa"]
|
||
d9_lagna = int(d9["Ascendant"]["sign_idx"])
|
||
lord_signs = {
|
||
index: int(d9[SIGN_LORDS[SIGNS[index]]]["sign_idx"])
|
||
for index in range(12)
|
||
}
|
||
# Upstream goel_method_1_v1 reads the lord's D9 sign. Ours derives the
|
||
# same sign from the natal longitude. Both must name the same first major.
|
||
assert (first["sign_index"], first["years"]) == _goel_method_1_first(d9_lagna, lord_signs)
|
||
longitudes = {
|
||
name: float(row["degree_raw"])
|
||
for name, row in report["chart"]["planets"].items()
|
||
if isinstance(row, dict) and isinstance(row.get("degree_raw"), (int, float))
|
||
}
|
||
asc_lon = float(report["chart"]["ascendant"]["degree_raw"])
|
||
longitude_lords = {
|
||
index: navamsha_sign(longitudes[SIGN_LORDS[SIGNS[index]]])
|
||
for index in range(12)
|
||
}
|
||
assert (first["sign_index"], first["years"]) == _goel_method_1_first(
|
||
navamsha_sign(asc_lon), longitude_lords
|
||
)
|
||
|
||
|
||
def test_synastry_validator_accepts_the_secondary_dashas(reports) -> None:
|
||
validator = _validator()
|
||
receipt = validator.build_execution_receipt(reports["pitt"], reports["jolie"])
|
||
rows = {row["technique"]: row for row in receipt["individual_technique_audit"]}
|
||
assert rows["Navamsha Dasha"]["status"] in {"executed", "parameter_sensitive"}
|
||
# 原值：Niryaana Shoola 必须是 executed 或 parameter_sensitive
|
||
# 新值：not_applicable
|
||
# 原因：修改单 M3。上游 7a773d13 校验器对这一行固定判不适用，合盘任务不因它 blocked。
|
||
assert rows["Niryaana Shoola Dasha"]["status"] == "not_applicable"
|
||
tasks = {row["task"]: row for row in receipt["pair_execution_tasks"]}
|
||
assert tasks["secondary-dasha convergence and disagreement review"]["status"] != "blocked"
|
||
|
||
|
||
def test_full_reading_still_succeeds_when_network_and_subprocess_raise() -> None:
|
||
import subprocess
|
||
import urllib.request
|
||
|
||
from unified_western_reading import build_unified_western_reading
|
||
|
||
calls = {"count": 0, "stacks": []}
|
||
original_urlopen = urllib.request.urlopen
|
||
original_run = subprocess.run
|
||
|
||
def boom(*_args, **_kwargs):
|
||
import traceback
|
||
|
||
calls["count"] += 1
|
||
calls["stacks"].append("".join(traceback.format_stack()))
|
||
raise RuntimeError("network and external programs are not allowed")
|
||
|
||
urllib.request.urlopen = boom
|
||
subprocess.run = boom
|
||
try:
|
||
birth = _case(ROOT / "tests/indastro_cases.json", "brad_pitt")
|
||
report = cmd_full_reading(_export_args(birth))
|
||
western = report["modules"]["unified_western_reading"]
|
||
assert western["status"] != "blocked"
|
||
# VedAstro's existing local runner may attempt subprocess and fail closed.
|
||
# The western packet must not be on that stack.
|
||
for stack in calls["stacks"]:
|
||
assert "unified_western_reading.py" not in stack
|
||
assert "classical_astrology.py" not in stack
|
||
calls["count"] = 0
|
||
packet = build_unified_western_reading(
|
||
year=birth["year"],
|
||
month=birth["month"],
|
||
day=birth["day"],
|
||
hour=birth["hour"],
|
||
minute=birth["minute"],
|
||
second=birth["second"],
|
||
latitude=birth["lat"],
|
||
longitude=birth["lon"],
|
||
timezone=birth["tz"],
|
||
reference_date=birth["today"],
|
||
duration_scan_end_date=birth["today"],
|
||
lunar_return_months=3,
|
||
)
|
||
assert packet["system"] == "western_astrology"
|
||
assert calls["count"] == 0
|
||
finally:
|
||
urllib.request.urlopen = original_urlopen
|
||
subprocess.run = original_run
|