root/Jyotisha
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Guard PyJHora artifacts and VedAstro HTTP stub

Browse Source
This commit is contained in:
732642856
2026-06-27 14:16:47 +08:00
parent 4265c66763
commit 6f46805fab
3 changed files with 231 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/vedastro_service_adapter.py
+97 -15
View File
@@ -13,6 +13,7 @@ import json
import os
from pathlib import Path
from typing import Any
from urllib import request, error
ROOT = Path(__file__).resolve().parents[1]
@@ -142,6 +143,26 @@ def _request_preview(case: dict[str, Any]) -> dict[str, Any]:
}
def _normalize_success(payload: dict[str, Any], endpoint: str) -> dict[str, Any]:
return {
"backend": "vedastro_service_adapter_candidate",
"available": True,
"status": "ok",
"ayanamsa_value": payload.get("ayanamsa_value"),
"node_policy": payload.get("node_policy"),
"body_list": payload.get("body_list"),
"bodies": payload.get("bodies"),
"source_metadata": {
"transport": "http_json_service_boundary",
"endpoint": endpoint,
"provenance_mode": "external_service_candidate",
"timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
"retry_policy": RETRY_POLICY,
**(payload.get("source_metadata") or {}),
},
}
def run_case(case_id: str) -> dict[str, Any]:
if case_id not in PARITY_CASES:
return {
@@ -173,21 +194,82 @@ def run_case(case_id: str) -> dict[str, Any]:
"network_execution_env": ALLOW_NETWORK_ENV,
},
}
return {
"backend": "vedastro_service_adapter_candidate",
"available": False,
"status": "service_execution_not_implemented",
"reason": "Endpoint is configured, but the real HTTP execution/normalization layer is intentionally not implemented yet.",
"request_preview": request_preview,
"source_metadata": {
"transport": "http_json_service_boundary",
"endpoint": endpoint,
"provenance_mode": "external_service_candidate",
"timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
"retry_policy": RETRY_POLICY,
"network_execution_env": ALLOW_NETWORK_ENV,
},
}
try:
req = request.Request(
endpoint,
data=json.dumps(request_preview).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="POST",
)
with request.urlopen(req, timeout=DEFAULT_TIMEOUT_SECONDS) as resp:
raw = resp.read().decode("utf-8")
payload = json.loads(raw)
except error.HTTPError as exc:
return {
"backend": "vedastro_service_adapter_candidate",
"available": False,
"status": "http_error",
"reason": f"VedAstro adapter HTTP error: {exc.code}",
"request_preview": request_preview,
"source_metadata": {
"transport": "http_json_service_boundary",
"endpoint": endpoint,
"provenance_mode": "external_service_candidate",
"timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
"retry_policy": RETRY_POLICY,
"network_execution_env": ALLOW_NETWORK_ENV,
},
}
except error.URLError as exc:
return {
"backend": "vedastro_service_adapter_candidate",
"available": False,
"status": "network_error",
"reason": f"VedAstro adapter network error: {exc.reason}",
"request_preview": request_preview,
"source_metadata": {
"transport": "http_json_service_boundary",
"endpoint": endpoint,
"provenance_mode": "external_service_candidate",
"timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
"retry_policy": RETRY_POLICY,
"network_execution_env": ALLOW_NETWORK_ENV,
},
}
except TimeoutError:
return {
"backend": "vedastro_service_adapter_candidate",
"available": False,
"status": "timeout",
"reason": "VedAstro adapter timed out",
"request_preview": request_preview,
"source_metadata": {
"transport": "http_json_service_boundary",
"endpoint": endpoint,
"provenance_mode": "external_service_candidate",
"timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
"retry_policy": RETRY_POLICY,
"network_execution_env": ALLOW_NETWORK_ENV,
},
}
except json.JSONDecodeError:
return {
"backend": "vedastro_service_adapter_candidate",
"available": False,
"status": "invalid_json",
"reason": "VedAstro adapter received non-JSON response",
"request_preview": request_preview,
"source_metadata": {
"transport": "http_json_service_boundary",
"endpoint": endpoint,
"provenance_mode": "external_service_candidate",
"timeout_seconds": DEFAULT_TIMEOUT_SECONDS,
"retry_policy": RETRY_POLICY,
"network_execution_env": ALLOW_NETWORK_ENV,
},
}
return _normalize_success(payload, endpoint)
def main() -> int:
tests/test_pyjhora_oracle_artifacts_presence.py
+33
View File
@@ -0,0 +1,33 @@
from __future__ import annotations
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
ARTIFACTS = ROOT / "references" / "oracle" / "artifacts"
def test_pyjhora_blackbox_artifacts_exist_for_current_hard_fronts() -> None:
required = [
"pyjhora_steve_jobs_dasha_stdout_20260627.txt",
"pyjhora_steve_jobs_shadbala_lahiri_stdout_20260627.txt",
"pyjhora_steve_jobs_varshaphala_1984_lahiri_stdout_20260627.txt",
"pyjhora_user_REDACTED_YEAR_shadbala_lahiri_stdout_20260627.txt",
"pyjhora_historical_epoch_dasha_stdout_20260627.txt",
]
for name in required:
path = ARTIFACTS / name
assert path.exists(), name
assert path.read_text(encoding="utf-8").strip(), name
def test_pyjhora_pending_packets_exist_for_replayable_external_capture() -> None:
pending = ARTIFACTS / "pending_packets"
required = [
"external_template_steve_jobs_dasha_lahiri_pyjhora_20260627.json",
"external_template_user_REDACTED_YEAR_shadbala_lahiri_pyjhora_20260627.json",
"external_template_steve_jobs_varshaphala_1984_lahiri_pyjhora_20260627.json",
"external_template_historical_epoch_lahiri_pyjhora_20260627.json",
]
for name in required:
assert (pending / name).exists(), name
tests/test_vedastro_service_adapter_executor.py
+101
View File
@@ -4,7 +4,9 @@ import json
import os
import subprocess
import sys
import threading
from pathlib import Path
from http.server import BaseHTTPRequestHandler, HTTPServer
ROOT = Path(__file__).resolve().parents[1]
@@ -97,3 +99,102 @@ def test_vedastro_service_adapter_builds_request_preview_before_real_network_use
assert report["request_preview"]["ayanamsa_policy"] == "lahiri"
assert report["source_metadata"]["endpoint"] == "https://example.invalid/vedastro"
assert report["source_metadata"]["provenance_mode"] == "external_service_candidate"
def test_vedastro_service_adapter_can_normalize_mock_http_response() -> None:
class Handler(BaseHTTPRequestHandler):
def do_POST(self) -> None: # noqa: N802
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode("utf-8"))
assert payload["year"] == 1990
response = {
"ayanamsa_value": 23.717413,
"node_policy": "mean",
"body_list": ["Sun", "Moon", "Ascendant", "Rahu", "Ketu"],
"bodies": {
"Sun": {"sidereal_longitude": 256.757014, "sign": "Sagittarius"},
"Moon": {"sidereal_longitude": 305.071255, "sign": "Aquarius"},
"Ascendant": {"sidereal_longitude": 348.1199, "sign": "Pisces"},
"Rahu": {"sidereal_longitude": 294.735223, "sign": "Capricorn"},
"Ketu": {"sidereal_longitude": 114.735223, "sign": "Cancer"},
},
"source_metadata": {
"service": "mock-vedastro",
"version": "test",
},
}
body = json.dumps(response).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, format: str, *args) -> None: # noqa: A003
return
server = HTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
env = os.environ.copy()
env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro"
env["VEDASTRO_ENABLE_NETWORK"] = "1"
completed = subprocess.run(
[sys.executable, "scripts/vedastro_service_adapter.py", "--case", "beijing_first_use_demo"],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env=env,
)
finally:
server.shutdown()
thread.join(timeout=5)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["backend"] == "vedastro_service_adapter_candidate"
assert report["available"] is True
assert report["status"] == "ok"
assert report["ayanamsa_value"] == 23.717413
assert report["bodies"]["Sun"]["sign"] == "Sagittarius"
assert report["source_metadata"]["transport"] == "http_json_service_boundary"
assert report["source_metadata"]["endpoint"].startswith("http://127.0.0.1:")
def test_vedastro_service_adapter_classifies_http_error() -> None:
class Handler(BaseHTTPRequestHandler):
def do_POST(self) -> None: # noqa: N802
self.send_response(503)
self.send_header("Content-Type", "application/json")
self.end_headers()
def log_message(self, format: str, *args) -> None: # noqa: A003
return
server = HTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
env = os.environ.copy()
env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro"
env["VEDASTRO_ENABLE_NETWORK"] = "1"
completed = subprocess.run(
[sys.executable, "scripts/vedastro_service_adapter.py", "--case", "beijing_first_use_demo"],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env=env,
)
finally:
server.shutdown()
thread.join(timeout=5)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["status"] == "http_error"
assert "503" in report["reason"]