From 1bb0ad9a599c580b62d5a4979ac540e52d1915d4 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Thu, 16 Jul 2026 20:36:44 +0800 Subject: [PATCH] Add external oracle parity tooling --- scripts/external_oracle_raw_import.py | 65 ++++ scripts/pyjhora_parity_summary.py | 61 +++ scripts/skill_release_package.py | 4 + .../three_engine_parity_replay_validator.py | 36 +- scripts/three_engine_parity_runner.py | 360 ++++++++++++++++++ tests/test_external_oracle_raw_import.py | 35 ++ ...t_three_engine_parity_artifact_contract.py | 21 + tests/test_three_engine_parity_runner.py | 147 +++++++ 8 files changed, 727 insertions(+), 2 deletions(-) create mode 100644 scripts/external_oracle_raw_import.py create mode 100644 scripts/pyjhora_parity_summary.py create mode 100644 scripts/three_engine_parity_runner.py create mode 100644 tests/test_external_oracle_raw_import.py create mode 100644 tests/test_three_engine_parity_artifact_contract.py create mode 100644 tests/test_three_engine_parity_runner.py diff --git a/scripts/external_oracle_raw_import.py b/scripts/external_oracle_raw_import.py new file mode 100644 index 00000000..e20889a6 --- /dev/null +++ b/scripts/external_oracle_raw_import.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""Validate a reviewable external raw-oracle artifact before parity replay.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +from typing import Any + + +SUPPORTED_ENGINES = {"VedAstro", "PyJHora_JHora", "jyotishganit"} + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def build_raw_oracle_import(engine: str, artifact_path: str | Path, metadata: dict[str, Any]) -> dict[str, Any]: + if engine not in SUPPORTED_ENGINES: + raise ValueError(f"unsupported oracle engine: {engine}") + path = Path(artifact_path).expanduser().resolve() + if not path.is_file(): + raise ValueError("source artifact does not exist") + required = ("case_id", "license_boundary", "collection_method", "birth_data_policy") + missing = [key for key in required if not metadata.get(key)] + if missing: + raise ValueError(f"missing raw-oracle metadata: {', '.join(missing)}") + if metadata["birth_data_policy"] != "public_case_only": + raise ValueError("raw-oracle imports require public_case_only birth data") + return { + "scope": "external_raw_oracle_import", + "schema_version": 1, + "engine": engine, + "status": "raw_imported_uncompared", + "source_artifact": str(path), + "source_artifact_sha256": sha256_file(path), + "metadata": { + key: metadata[key] + for key in (*required, "engine_version", "ayanamsa", "node_mode", "captured_at") + if metadata.get(key) is not None + }, + "comparison_ready": False, + "boundary": "Import integrity only. Parity is external_verified only after normalized field comparison passes.", + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--engine", required=True, choices=sorted(SUPPORTED_ENGINES)) + parser.add_argument("--artifact", required=True) + parser.add_argument("--metadata-json", required=True, help="JSON file containing import metadata") + args = parser.parse_args() + metadata = json.loads(Path(args.metadata_json).read_text(encoding="utf-8")) + print(json.dumps(build_raw_oracle_import(args.engine, args.artifact, metadata), ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/pyjhora_parity_summary.py b/scripts/pyjhora_parity_summary.py new file mode 100644 index 00000000..bb40ea76 --- /dev/null +++ b/scripts/pyjhora_parity_summary.py @@ -0,0 +1,61 @@ +#!/usr/bin/env python3 +"""Summarize reviewable PyJHora comparison matrices without overstating coverage.""" + +from __future__ import annotations + +import argparse +import csv +import json +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any + + +REQUIRED_FULL_PARITY = ("D1", "D9", "D10", "D2", "D4", "Vimshottari", "Shadbala", "Ashtakavarga") +MATRIX_SECTION_MAP = {"ascendant": "D1", "planet": "D1", "dasha": "Vimshottari", "D9": "D9", "D10": "D10"} + + +def summarize_matrix(path: str | Path, *, settings: dict[str, Any]) -> dict[str, Any]: + path = Path(path) + rows = list(csv.DictReader(path.open(encoding="utf-8"))) + status_counts = Counter(str(row.get("status") or "unknown") for row in rows) + sections: dict[str, dict[str, int]] = defaultdict(lambda: {"total": 0, "match": 0, "mismatch": 0}) + covered = set() + for row in rows: + section = MATRIX_SECTION_MAP.get(str(row.get("section") or "")) + if not section: + continue + covered.add(section) + sections[section]["total"] += 1 + if row.get("status") == "match": + sections[section]["match"] += 1 + elif row.get("status") == "mismatch": + sections[section]["mismatch"] += 1 + missing = [field for field in REQUIRED_FULL_PARITY if field not in covered] + return { + "scope": "pyjhora_same_chart_parity_summary", + "matrix_path": str(path), + "tested": bool(rows), + "settings": settings, + "row_counts": dict(status_counts), + "coverage": dict(sorted(sections.items())), + "covered_outputs": sorted(covered), + "missing_required_outputs": missing, + "status": "partial_verified" if rows and not status_counts.get("mismatch") else "partial_mismatch", + "full_parity_verified": not missing and not status_counts.get("mismatch"), + "boundary": "Only covered outputs are compared. This summary cannot promote full parity while required outputs are absent.", + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("matrix") + parser.add_argument("--ayanamsa", default="lahiri") + parser.add_argument("--node-mode", default="mean", choices=["mean", "true"]) + args = parser.parse_args() + print(json.dumps(summarize_matrix(args.matrix, settings={"ayanamsa": args.ayanamsa, "node_mode": args.node_mode}), ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/skill_release_package.py b/scripts/skill_release_package.py index e1bc14bf..0ff81548 100644 --- a/scripts/skill_release_package.py +++ b/scripts/skill_release_package.py @@ -110,6 +110,10 @@ REQUIRED_CONTRACTS = [ "references/oracle/western_oracle_adapter_contract.md", "scripts/user_invocation_acceptance_check.py", "scripts/diagnose_external_engine_adapters.py", + "scripts/external_oracle_raw_import.py", + "scripts/three_engine_parity_runner.py", + "scripts/three_engine_parity_replay_validator.py", + "scripts/pyjhora_parity_summary.py", "scripts/western_chart_engine.py", "scripts/western_timing_engine.py", ] diff --git a/scripts/three_engine_parity_replay_validator.py b/scripts/three_engine_parity_replay_validator.py index 5d3a9ba5..39081435 100644 --- a/scripts/three_engine_parity_replay_validator.py +++ b/scripts/three_engine_parity_replay_validator.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse +import hashlib import json from pathlib import Path from typing import Any @@ -12,6 +13,33 @@ from typing import Any REQUIRED_ENGINES = {"VedAstro", "PyJHora_JHora", "jyotishganit"} REQUIRED_ROW_FIELDS = {"section", "field", "local_value", "oracle_values", "status"} VALID_ROW_STATUSES = {"match", "mismatch", "blocked", "not_comparable"} +RAW_VERIFIED_STATUSES = {"verified", "official_verified", "imported"} + + +def _artifact_errors(engine: str, payload: Any, manifest_dir: Path) -> list[dict[str, Any]]: + if not isinstance(payload, dict): + return [{"field": f"engines.{engine}", "error": "not_object"}] + if payload.get("status") not in RAW_VERIFIED_STATUSES: + return [] + raw_path = payload.get("official_raw_response_path") or payload.get("raw_output_path") + artifact_hash = payload.get("artifact_hash") + errors: list[dict[str, Any]] = [] + if not isinstance(raw_path, str) or not raw_path: + errors.append({"field": f"engines.{engine}.raw_output_path", "error": "required_for_verified_status"}) + return errors + if not isinstance(artifact_hash, str) or len(artifact_hash) != 64: + errors.append({"field": f"engines.{engine}.artifact_hash", "error": "sha256_required_for_verified_status"}) + return errors + artifact_path = (manifest_dir / raw_path).resolve() + if not artifact_path.is_file(): + errors.append({"field": f"engines.{engine}.raw_output_path", "error": "missing_artifact"}) + return errors + actual_hash = hashlib.sha256(artifact_path.read_bytes()).hexdigest() + if actual_hash != artifact_hash: + errors.append({"field": f"engines.{engine}.artifact_hash", "error": "hash_mismatch"}) + if not isinstance(payload.get("settings"), dict): + errors.append({"field": f"engines.{engine}.settings", "error": "required_for_verified_status"}) + return errors def _row_errors(row: Any, index: int) -> list[dict[str, Any]]: @@ -37,6 +65,8 @@ def validate_manifest(path: str | Path) -> dict[str, Any]: missing_engines = sorted(REQUIRED_ENGINES - set(engines)) for engine in missing_engines: errors.append({"field": f"engines.{engine}", "error": "missing"}) + for engine, payload in engines.items(): + errors.extend(_artifact_errors(engine, payload, manifest_path.parent)) if not isinstance(rows, list): rows = [] @@ -89,9 +119,11 @@ def validate_manifest(path: str | Path) -> dict[str, Any]: def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("manifest", nargs="?", default="references/oracle/three_engine_parity_replay_manifest.json") + parser.add_argument("--require-pass", action="store_true", help="Return nonzero unless all comparison rows pass.") args = parser.parse_args(argv) - print(json.dumps(validate_manifest(args.manifest), ensure_ascii=False, indent=2, sort_keys=True)) - return 0 + report = validate_manifest(args.manifest) + print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if not args.require_pass or report["status"] == "pass" else 1 if __name__ == "__main__": diff --git a/scripts/three_engine_parity_runner.py b/scripts/three_engine_parity_runner.py new file mode 100644 index 00000000..3a01e87b --- /dev/null +++ b/scripts/three_engine_parity_runner.py @@ -0,0 +1,360 @@ +#!/usr/bin/env python3 +"""Capture a public same-chart parity packet without overstating oracle closure.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import sys +from datetime import datetime +from pathlib import Path +from typing import Any + +from domain_calculation_service import compute_chart + + +ROOT = Path(__file__).resolve().parents[1] +PYJHORA_ARTIFACT = ROOT / "references/oracle/artifacts/pyjhora_steve_jobs_dasha_stdout_20260627.txt" +JYOTISHGANIT_ROOT = ROOT / "references/open_source_sources/jyotishganit" +VEDASTRO_ARTIFACT_DIR = ROOT / "scratch/local/vedastro_adapter" + +PUBLIC_CASE = { + "case_id": "steve_jobs_public_1955_lahiri", + "year": 1955, + "month": 2, + "day": 24, + "hour": 19, + "minute": 15, + "second": 0, + "lat": 37.7749, + "lon": -122.4194, + "tz": -8.0, + "ayanamsa": "lahiri", + "node_mode": "mean", +} +PLANETS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn") +SIGNS = ("Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces") +LONGITUDE_TOLERANCE_DEGREES = 0.02 + + +def _write_json(path: Path, value: dict[str, Any]) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8") + return path + + +def _capture_jyotishganit_raw(output_dir: Path) -> tuple[dict[str, Any], str]: + sys.path.insert(0, str(JYOTISHGANIT_ROOT)) + try: + from jyotishganit import calculate_birth_chart, get_birth_chart_json + + chart = calculate_birth_chart( + datetime( + PUBLIC_CASE["year"], + PUBLIC_CASE["month"], + PUBLIC_CASE["day"], + PUBLIC_CASE["hour"], + PUBLIC_CASE["minute"], + PUBLIC_CASE["second"], + ), + PUBLIC_CASE["lat"], + PUBLIC_CASE["lon"], + PUBLIC_CASE["tz"], + location_name="San Francisco, CA", + name="Steve Jobs (public benchmark)", + ) + raw = get_birth_chart_json(chart) + path = _write_json(output_dir / "jyotishganit_raw.json", raw) + return raw, str(path) + except Exception as exc: + return {"error": f"{exc.__class__.__name__}: {exc}"}, "" + finally: + try: + sys.path.remove(str(JYOTISHGANIT_ROOT)) + except ValueError: + pass + + +def _capture_pyjhora_structured_d1(output_dir: Path) -> tuple[dict[str, Any], str]: + try: + import contextlib + import importlib + import io + + with contextlib.redirect_stdout(io.StringIO()): + utils = importlib.import_module("jhora.utils") + charts = importlib.import_module("jhora.horoscope.chart.charts") + drik = importlib.import_module("jhora.panchanga.drik") + + jd = utils.julian_day_number( + (PUBLIC_CASE["year"], PUBLIC_CASE["month"], PUBLIC_CASE["day"]), + (PUBLIC_CASE["hour"], PUBLIC_CASE["minute"], PUBLIC_CASE["second"]), + ) + drik.set_ayanamsa_mode("LAHIRI", jd=jd) + place = drik.Place("San Francisco, CA", PUBLIC_CASE["lat"], PUBLIC_CASE["lon"], PUBLIC_CASE["tz"]) + with contextlib.redirect_stdout(io.StringIO()): + raw = charts.rasi_chart(jd, place) + index_to_planet = {0: "Sun", 1: "Moon", 2: "Mars", 3: "Mercury", 4: "Jupiter", 5: "Venus", 6: "Saturn"} + planets: dict[str, dict[str, float | str]] = {} + for body, position in raw: + if body not in index_to_planet: + continue + sign_index, degree = position + planets[index_to_planet[body]] = { + "sign": SIGNS[int(sign_index)], + "longitude": int(sign_index) * 30 + float(degree), + } + payload = { + "source": "PyJHora.jhora.horoscope.chart.charts.rasi_chart", + "settings": {"ayanamsa": "LAHIRI", "jd_input": "local_birth_time", "node_mode": "PyJHora default"}, + "raw": raw, + "planets": planets, + } + path = _write_json(output_dir / "pyjhora_structured_d1.json", payload) + return payload, str(path) + except Exception as exc: + return {"error": f"{exc.__class__.__name__}: {exc}"}, "" + + +def _vedastro_state(*, allow_network: bool) -> dict[str, Any]: + if not allow_network: + return { + "status": "blocked", + "official_raw_response_path": "", + "reason": "network_disabled_for_public_replay", + } + artifact = _latest_vedastro_official_raw_artifact() + if artifact: + return { + "status": "official_verified", + "official_raw_response_path": str(artifact), + "artifact_hash": hashlib.sha256(artifact.read_bytes()).hexdigest(), + "settings": {"ayanamsa": PUBLIC_CASE["ayanamsa"], "node_mode": PUBLIC_CASE["node_mode"]}, + "reason": "imported_latest_official_full_snapshot_artifact", + } + return { + "status": "blocked", + "official_raw_response_path": "", + "reason": "official_runner_requires_explicit_raw_capture_workflow", + } + + +def _latest_vedastro_official_raw_artifact() -> Path | None: + if not VEDASTRO_ARTIFACT_DIR.exists(): + return None + candidates: list[Path] = [] + for path in VEDASTRO_ARTIFACT_DIR.glob("official_full_snapshot-*.json"): + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + continue + raw = payload.get("official_raw_response") or payload.get("raw_response") + source = str(raw.get("source") or "") if isinstance(raw, dict) else "" + if ( + payload.get("status") == "ok" + and source.startswith("vedastro_official") + and _artifact_matches_public_case(payload) + ): + candidates.append(path) + if not candidates: + return None + return max(candidates, key=lambda item: item.stat().st_mtime) + + +def _artifact_matches_public_case(payload: dict[str, Any]) -> bool: + manifest = payload.get("request_manifest") if isinstance(payload.get("request_manifest"), dict) else {} + text = json.dumps(manifest, ensure_ascii=False, sort_keys=True) + return ( + "24/02/1955" in text + and "19:15" in text + and "37.7749" in text + and "-122.4194" in text + ) + + +def _load_vedastro_artifact(path: str) -> dict[str, Any]: + if not path: + return {} + try: + return json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + + +def _vedastro_d1(artifact: dict[str, Any], planet: str) -> dict[str, Any]: + try: + payload = artifact["snapshot_sections"]["chart_core"][planet]["Payload"]["AllPlanetData"] + return { + "sign": payload["PlanetRasiD1Sign"]["Name"], + "longitude": float(payload["PlanetNirayanaLongitude"]["TotalDegrees"]), + } + except (KeyError, TypeError, ValueError): + return {} + + +def _jyotishganit_d1(raw: dict[str, Any], planet: str) -> dict[str, Any]: + for house in (raw.get("d1Chart") or {}).get("houses") or []: + for occupant in house.get("occupants") or []: + if occupant.get("celestialBody") != planet: + continue + sign = occupant.get("sign") + degree = occupant.get("signDegrees") + if sign not in SIGNS or degree is None: + return {} + return {"sign": sign, "longitude": SIGNS.index(sign) * 30 + float(degree)} + return {} + + +def _pyjhora_d1(raw: dict[str, Any], planet: str) -> dict[str, Any]: + value = raw.get("planets", {}).get(planet) + return value if isinstance(value, dict) else {} + + +def _d1_comparison_rows( + local: dict[str, Any], + vedastro_artifact: dict[str, Any], + jyotishganit_raw: dict[str, Any], + pyjhora_raw: dict[str, Any], +) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for planet in PLANETS: + local_planet = local.get("planets", {}).get(planet) or {} + vedastro = _vedastro_d1(vedastro_artifact, planet) + jyotishganit = _jyotishganit_d1(jyotishganit_raw, planet) + pyjhora = _pyjhora_d1(pyjhora_raw, planet) + local_sign = local_planet.get("sign") + sign_values = { + "VedAstro": vedastro.get("sign"), + "PyJHora_JHora": pyjhora.get("sign"), + "jyotishganit": jyotishganit.get("sign"), + } + comparable_signs = [value for value in sign_values.values() if value is not None] + rows.append({ + "section": "D1", + "field": f"{planet}.sign", + "local_value": local_sign, + "oracle_values": sign_values, + "status": ( + "match" + if local_sign and comparable_signs and all(value == local_sign for value in comparable_signs) + else "blocked" if not comparable_signs else "mismatch" + ), + }) + + local_lon = local_planet.get("lon") + longitude_values = { + "VedAstro": vedastro.get("longitude"), + "PyJHora_JHora": pyjhora.get("longitude"), + "jyotishganit": jyotishganit.get("longitude"), + } + comparable = [value for value in longitude_values.values() if isinstance(value, (int, float))] + rows.append({ + "section": "D1", + "field": f"{planet}.longitude", + "local_value": local_lon, + "oracle_values": longitude_values, + "status": ( + "match" + if isinstance(local_lon, (int, float)) + and comparable + and all(abs(value - local_lon) <= LONGITUDE_TOLERANCE_DEGREES for value in comparable) + else "blocked" if not comparable else "mismatch" + ), + }) + return rows + + +def build_public_case_replay(*, output_dir: Path, allow_vedastro_network: bool = False) -> dict[str, Any]: + output_dir.mkdir(parents=True, exist_ok=True) + local = compute_chart(PUBLIC_CASE) + jyotishganit_raw, jyotishganit_path = _capture_jyotishganit_raw(output_dir) + pyjhora_raw, pyjhora_path = _capture_pyjhora_structured_d1(output_dir) + pyjhora_available = PYJHORA_ARTIFACT.is_file() + vedastro = _vedastro_state(allow_network=allow_vedastro_network) + vedastro_artifact = _load_vedastro_artifact(vedastro.get("official_raw_response_path", "")) + rows = _d1_comparison_rows(local, vedastro_artifact, jyotishganit_raw, pyjhora_raw) + [ + { + "section": "Panchanga", + "field": "raw_capture", + "local_value": None, + "oracle_values": { + "VedAstro": None, + "PyJHora_JHora": "structured_d1_captured" if pyjhora_path else "dasha_only_artifact", + "jyotishganit": "captured" if jyotishganit_path else None, + }, + "status": "not_comparable", + "reason": "three_engine_scope_does_not_share_this_normalized_field", + }, + ] + has_blocked = any(row.get("status") == "blocked" for row in rows) + has_mismatch = any(row.get("status") == "mismatch" for row in rows) + has_required_raw = vedastro.get("status") == "official_verified" and bool(pyjhora_path) and bool(jyotishganit_path) + blocked_reason = ( + "official_vedastro_raw_missing_or_unverified" + if vedastro.get("status") != "official_verified" + else "some_comparison_rows_blocked" + if has_blocked + else "comparison_rows_mismatch" + if has_mismatch + else "none" + ) + report_status = ( + "blocked" + if not bool(jyotishganit_path) + else "mismatch" + if has_mismatch + else "partial" + if not has_required_raw or has_blocked + else "pass" + ) + pyjhora_artifact_path = pyjhora_path or (str(PYJHORA_ARTIFACT) if pyjhora_available else "") + report = { + "case_id": PUBLIC_CASE["case_id"], + "birth_data_policy": "public_case_only", + "status": report_status, + "tested": vedastro.get("status") == "official_verified" and bool(rows), + "blocked_reason": blocked_reason, + "engines": { + "VedAstro": vedastro, + "PyJHora_JHora": { + "status": "structured_captured" if pyjhora_path else "raw_imported" if pyjhora_available else "blocked", + "raw_output_path": pyjhora_artifact_path, + "artifact_hash": hashlib.sha256(Path(pyjhora_artifact_path).read_bytes()).hexdigest() if pyjhora_artifact_path else "", + "settings": {"ayanamsa": "LAHIRI", "node_mode": "PyJHora default"}, + }, + "jyotishganit": { + "status": "raw_captured" if jyotishganit_path else "blocked", + "raw_output_path": jyotishganit_path, + "error": jyotishganit_raw.get("error") if isinstance(jyotishganit_raw, dict) else None, + }, + }, + "local": { + "result_hash": local["result_hash"], + "calculation_contract": local["calculation_contract"], + }, + "comparison_rows": rows, + "runtime_boundary": ( + "This packet has real public raw artifacts and may include VedAstro official raw evidence, " + "and normalized D1 planet sign/longitude parity is tested. Non-D1 scopes still require separate gates." + ), + } + _write_json(output_dir / "three_engine_parity_replay.json", report) + return report + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--output-dir", default="scratch/local/three_engine_parity") + parser.add_argument("--allow-vedastro-network", action="store_true") + args = parser.parse_args() + report = build_public_case_replay( + output_dir=ROOT / args.output_dir, + allow_vedastro_network=args.allow_vedastro_network, + ) + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_external_oracle_raw_import.py b/tests/test_external_oracle_raw_import.py new file mode 100644 index 00000000..58fb6e74 --- /dev/null +++ b/tests/test_external_oracle_raw_import.py @@ -0,0 +1,35 @@ +from pathlib import Path + +import pytest + +from scripts.external_oracle_raw_import import build_raw_oracle_import + + +def _metadata(**overrides): + value = { + "case_id": "public_case", + "license_boundary": "external benchmark only", + "collection_method": "manual export", + "birth_data_policy": "public_case_only", + } + value.update(overrides) + return value + + +def test_raw_import_requires_reviewable_public_case_artifact(tmp_path: Path): + artifact = tmp_path / "oracle.json" + artifact.write_text('{"raw": true}', encoding="utf-8") + + result = build_raw_oracle_import("VedAstro", artifact, _metadata()) + + assert result["status"] == "raw_imported_uncompared" + assert result["source_artifact_sha256"] + assert result["comparison_ready"] is False + + +def test_raw_import_rejects_non_public_birth_policy(tmp_path: Path): + artifact = tmp_path / "oracle.json" + artifact.write_text("{}", encoding="utf-8") + + with pytest.raises(ValueError, match="public_case_only"): + build_raw_oracle_import("VedAstro", artifact, _metadata(birth_data_policy="private")) diff --git a/tests/test_three_engine_parity_artifact_contract.py b/tests/test_three_engine_parity_artifact_contract.py new file mode 100644 index 00000000..c55e5013 --- /dev/null +++ b/tests/test_three_engine_parity_artifact_contract.py @@ -0,0 +1,21 @@ +import json + +from scripts.three_engine_parity_replay_validator import validate_manifest + + +def test_verified_oracle_requires_raw_artifact_hash_and_settings(tmp_path): + manifest = { + "engines": { + "VedAstro": {"status": "official_verified"}, + "PyJHora_JHora": {"status": "blocked"}, + "jyotishganit": {"status": "blocked"}, + }, + "comparison_rows": [], + } + path = tmp_path / "manifest.json" + path.write_text(json.dumps(manifest), encoding="utf-8") + + result = validate_manifest(path) + + assert result["status"] == "invalid" + assert any(error["error"] == "required_for_verified_status" for error in result["errors"]) diff --git a/tests/test_three_engine_parity_runner.py b/tests/test_three_engine_parity_runner.py new file mode 100644 index 00000000..34aebc7e --- /dev/null +++ b/tests/test_three_engine_parity_runner.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + +import three_engine_parity_runner # noqa: E402 +from three_engine_parity_runner import build_public_case_replay # noqa: E402 + + +def test_public_same_chart_replay_never_promotes_missing_vedastro_raw(tmp_path: Path) -> None: + report = build_public_case_replay(output_dir=tmp_path, allow_vedastro_network=False) + + assert report["case_id"] == "steve_jobs_public_1955_lahiri" + assert report["birth_data_policy"] == "public_case_only" + assert report["engines"]["PyJHora_JHora"]["status"] == "structured_captured" + assert report["engines"]["jyotishganit"]["status"] == "raw_captured" + assert report["engines"]["VedAstro"]["status"] == "blocked" + assert report["status"] in {"partial", "blocked"} + assert report["tested"] is False + assert report["comparison_rows"] + assert any(row["status"] == "match" for row in report["comparison_rows"]) + assert all(row["status"] in {"blocked", "match", "not_comparable"} for row in report["comparison_rows"]) + + +def test_public_same_chart_replay_imports_verified_vedastro_artifact(tmp_path: Path, monkeypatch) -> None: + artifact_root = tmp_path / "vedastro_adapter" + artifact_root.mkdir() + artifact = artifact_root / "official_full_snapshot-abc-def.json" + artifact.write_text( + json.dumps( + { + "status": "ok", + "operation": "official_full_snapshot", + "raw_response": {"source": "vedastro_official_full_snapshot", "sections": {"chart_core": {}}}, + "request_manifest": { + "settings": {"ayanamsa": "lahiri", "node_mode": "mean"}, + "requests": [ + { + "body": { + "BirthTime": { + "StdTime": "19:15 24/02/1955 -08:00", + "Location": {"Latitude": 37.7749, "Longitude": -122.4194}, + } + } + } + ], + }, + "snapshot_sections": {"chart_core": {"Status": "Pass"}}, + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(three_engine_parity_runner, "VEDASTRO_ARTIFACT_DIR", artifact_root) + + report = build_public_case_replay(output_dir=tmp_path / "out", allow_vedastro_network=True) + + assert report["engines"]["VedAstro"]["status"] == "official_verified" + assert report["engines"]["VedAstro"]["official_raw_response_path"] == str(artifact) + assert len(report["engines"]["VedAstro"]["artifact_hash"]) == 64 + assert report["blocked_reason"] == "none" + + +def test_public_same_chart_replay_rejects_wrong_vedastro_artifact(tmp_path: Path, monkeypatch) -> None: + artifact_root = tmp_path / "vedastro_adapter" + artifact_root.mkdir() + artifact = artifact_root / "official_full_snapshot-wrong-chart.json" + artifact.write_text( + json.dumps( + { + "status": "ok", + "raw_response": {"source": "vedastro_official_full_snapshot"}, + "request_manifest": { + "requests": [ + { + "body": { + "BirthTime": { + "StdTime": "12:00 01/01/1990 +05:30", + "Location": {"Latitude": 28.6139, "Longitude": 77.209}, + } + } + } + ] + }, + "snapshot_sections": {"chart_core": {"Status": "Pass"}}, + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(three_engine_parity_runner, "VEDASTRO_ARTIFACT_DIR", artifact_root) + + report = build_public_case_replay(output_dir=tmp_path / "out", allow_vedastro_network=True) + + assert report["engines"]["VedAstro"]["status"] == "blocked" + assert report["engines"]["VedAstro"]["reason"] == "official_runner_requires_explicit_raw_capture_workflow" + + +def test_public_same_chart_replay_adds_normalized_d1_rows(tmp_path: Path, monkeypatch) -> None: + artifact_root = tmp_path / "vedastro_adapter" + artifact_root.mkdir() + artifact = artifact_root / "official_full_snapshot-abc-def.json" + artifact.write_text( + json.dumps( + { + "status": "ok", + "raw_response": {"source": "vedastro_official_full_snapshot"}, + "request_manifest": { + "requests": [ + { + "body": { + "BirthTime": { + "StdTime": "19:15 24/02/1955 -08:00", + "Location": {"Latitude": 37.7749, "Longitude": -122.4194}, + } + } + } + ] + }, + "snapshot_sections": { + "chart_core": { + "Sun": { + "Payload": { + "AllPlanetData": { + "PlanetNirayanaLongitude": {"TotalDegrees": "312.5122"}, + "PlanetRasiD1Sign": {"Name": "Aquarius"}, + } + } + } + } + }, + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(three_engine_parity_runner, "VEDASTRO_ARTIFACT_DIR", artifact_root) + + report = build_public_case_replay(output_dir=tmp_path / "out", allow_vedastro_network=True) + rows = {(row["section"], row["field"]): row for row in report["comparison_rows"]} + + assert rows[("D1", "Sun.sign")]["status"] == "match" + assert rows[("D1", "Sun.longitude")]["status"] == "match"