The two upstream files stay verbatim in scripts/. The path helper moved into a subdirectory so the root file budget rises by only those two. Niryaana Shoola is a placeholder until its rule is corrected. Targeted tests passed. Co-Authored-By: Grok (agent loop) Co-authored-by: Cursor <cursoragent@cursor.com>
242 lines
10 KiB
Python
242 lines
10 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"
|
|
assert niryaana == {
|
|
"execution_status": "blocked",
|
|
"reason": "口径更正中,见 TASK-niryaana-shoola-pvr-20261018",
|
|
}
|
|
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
|