from __future__ import annotations import json import os import subprocess import sys import threading import types from pathlib import Path from http.server import BaseHTTPRequestHandler, HTTPServer ROOT = Path(__file__).resolve().parents[1] def test_vedastro_service_adapter_executor_schema_is_declared() -> None: completed = subprocess.run( [sys.executable, "scripts/vedastro_service_adapter.py", "--print-schema"], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, ) assert completed.returncode == 0, completed.stderr or completed.stdout report = json.loads(completed.stdout) assert report["adapter"] == "vedastro_service_adapter" assert report["transport"] == "http_json_service_boundary" assert 3 <= report["default_timeout_seconds"] <= 5 assert report["retry_policy"]["max_attempts"] >= 1 assert report["retry_policy"]["backoff_seconds"] >= 0 assert "endpoint" in report["required_env"] assert "ayanamsa_policy" in report["request_contract"] assert "bodies" in report["response_contract"] assert "source_metadata" in report["response_contract"] assert "request_example" in report assert report["provenance_contract"]["external_service"] is True assert "endpoint" in report["provenance_contract"]["required_fields"] assert "range_scan_request_contract" in report assert "domain" in report["range_scan_request_contract"] assert "vedastro_event_method" in report["range_scan_request_contract"] assert "start_date" in report["range_scan_request_contract"] assert report["vedastro_calculation_coverage"] == { "official_python_library_calculations": "596+", "official_api_builder_calculators": "600+", "official_events_builder_events": "400+", "official_events_builder_methods": ["SearchEvents", "GetEventTiming", "ListEventTypes"], "range_scan_role": "high_frequency_life_event_radar", "intended_use": "external_timing_evidence_for_strict_workflow", } assert "range_scan_response_contract" in report assert "evidence_ledger" in report["range_scan_response_contract"] assert report["range_scan_event_allowlist"]["marriage"]["event_ids"] assert "marriage" in report["range_scan_event_allowlist"]["marriage"]["tags"] def test_vedastro_official_subprocesses_use_adapter_timeout(monkeypatch) -> None: from scripts import vedastro_service_adapter as adapter monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "7") seen_timeouts = [] def fake_run(*args, **kwargs): seen_timeouts.append(kwargs.get("timeout")) return types.SimpleNamespace(returncode=1, stdout="", stderr="simulated failure") monkeypatch.setattr(adapter.subprocess, "run", fake_run) case = { "year": 1990, "month": 4, "day": 17, "hour": 14, "minute": 59, "lat": 36.42, "lon": 114.2, "tz": 8, } bridge = adapter._call_vedastro_python_bridge_high_value("official_full_snapshot_bundle", {}) runner = adapter._try_official_capability_runner_snapshot_bundle(case) catalog = adapter._try_official_full_capability_catalog_bundle(case) assert seen_timeouts == [7.0, 7.0, 7.0] assert bridge["status"] == "python_bridge_runtime_error" assert runner["status"] == "official_capability_runner_runtime_error" assert catalog["status"] == "official_full_capability_catalog_runtime_error" def test_temporary_timeout_is_scoped_to_the_current_execution(monkeypatch) -> None: from scripts import vedastro_service_adapter as adapter from scripts.vedastro_runtime_context import temporary_timeout_seconds monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "7") assert adapter._timeout_seconds() == 7.0 with temporary_timeout_seconds(90): assert adapter._timeout_seconds() == 90.0 assert adapter._timeout_seconds() == 7.0 def test_vedastro_official_subprocess_timeouts_are_controlled(monkeypatch) -> None: from scripts import vedastro_service_adapter as adapter monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "3") def fake_timeout(*args, **kwargs): raise subprocess.TimeoutExpired(cmd=args[0], timeout=kwargs.get("timeout")) monkeypatch.setattr(adapter.subprocess, "run", fake_timeout) case = { "year": 1990, "month": 4, "day": 17, "hour": 14, "minute": 59, "lat": 36.42, "lon": 114.2, "tz": 8, } bridge = adapter._call_vedastro_python_bridge_high_value("official_full_snapshot_bundle", {}) runner = adapter._try_official_capability_runner_snapshot_bundle(case) catalog = adapter._try_official_full_capability_catalog_bundle(case) assert bridge["status"] == "python_bridge_timeout" assert runner["status"] == "official_capability_runner_timeout" assert catalog["status"] == "official_full_capability_catalog_timeout" def test_vedastro_official_snapshot_stops_when_foreground_budget_is_exhausted(monkeypatch) -> None: from scripts import vedastro_service_adapter as adapter monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api") monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1") monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "4") monkeypatch.setattr( adapter, "_try_official_full_capability_catalog_bundle", lambda case: { "available": False, "status": "official_full_capability_catalog_timeout", "source": "vedastro_official_capability_runner", "bundle": "official_full_capability_catalog", "summary": {}, "coverage": {}, "domain_routing": {}, "dynamic_selection": {}, }, ) runner_called = {"value": False} bridge_called = {"value": False} def fail_runner(case): runner_called["value"] = True raise AssertionError("snapshot runner should not run after foreground budget is exhausted") def fail_bridge(case): bridge_called["value"] = True raise AssertionError("python bridge fallback should not run after foreground budget is exhausted") monkeypatch.setattr(adapter, "_try_official_capability_runner_snapshot_bundle", fail_runner) monkeypatch.setattr(adapter, "_try_official_python_bridge_snapshot_bundle", fail_bridge) ticks = iter([100.0, 104.1]) monkeypatch.setattr(adapter.time, "monotonic", lambda: next(ticks)) result = adapter._run_official_full_snapshot_case({ "year": 1990, "month": 4, "day": 17, "hour": 14, "minute": 59, "lat": 36.42, "lon": 114.2, "tz": 8, }) assert runner_called["value"] is False assert bridge_called["value"] is False assert result["status"] == "official_snapshot_budget_exhausted" assert result["available"] is False assert result["source_metadata"]["official_python_bundle"]["status"] == "official_snapshot_budget_exhausted" assert result["source_metadata"]["official_full_capability_catalog"]["status"] == "official_full_capability_catalog_skipped_budget_exhausted" def test_vedastro_official_snapshot_skips_bridge_after_runner_consumes_budget(monkeypatch) -> None: from scripts import vedastro_service_adapter as adapter monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api") monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1") monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "4") monkeypatch.setattr( adapter, "_try_official_full_capability_catalog_bundle", lambda case: { "available": True, "status": "partial", "source": "vedastro_official_capability_runner", "bundle": "official_full_capability_catalog", "summary": {"catalog_method_count": 641}, "coverage": {}, "domain_routing": {}, "dynamic_selection": {}, }, ) monkeypatch.setattr( adapter, "_try_official_capability_runner_snapshot_bundle", lambda case: { "available": False, "status": "official_capability_runner_timeout", "source": "vedastro_official_capability_runner", "bundle": "official_full_snapshot", "snapshot_sections": {}, "section_statuses": {}, "coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []}, }, ) bridge_called = {"value": False} def fail_bridge(case): bridge_called["value"] = True raise AssertionError("python bridge fallback should not run after snapshot runner exhausts budget") monkeypatch.setattr(adapter, "_try_official_python_bridge_snapshot_bundle", fail_bridge) ticks = iter([100.0, 100.1, 104.2]) monkeypatch.setattr(adapter.time, "monotonic", lambda: next(ticks)) result = adapter._run_official_full_snapshot_case({ "year": 1990, "month": 4, "day": 17, "hour": 14, "minute": 59, "lat": 36.42, "lon": 114.2, "tz": 8, }) assert bridge_called["value"] is False assert result["status"] == "official_snapshot_budget_exhausted" assert result["source_metadata"]["official_python_bundle"]["status"] == "official_snapshot_budget_exhausted" assert result["source_metadata"]["official_full_capability_catalog"]["status"] == "official_full_capability_catalog_skipped_budget_exhausted" def test_vedastro_official_snapshot_budget_does_not_mask_mock_rest_endpoint(monkeypatch) -> None: from scripts import vedastro_service_adapter as adapter monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "http://127.0.0.1:12345/api") monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1") monkeypatch.setenv("VEDASTRO_TIMEOUT_SECONDS", "4") monkeypatch.setattr( adapter, "_try_official_full_capability_catalog_bundle", lambda case: { "available": False, "status": "official_full_capability_catalog_timeout", "source": "vedastro_official_capability_runner", "bundle": "official_full_capability_catalog", "summary": {}, "coverage": {}, "domain_routing": {}, "dynamic_selection": {}, }, ) monkeypatch.setattr( adapter, "_try_official_capability_runner_snapshot_bundle", lambda case: { "available": False, "status": "official_capability_runner_timeout", "source": "vedastro_official_capability_runner", "bundle": "official_full_snapshot", "snapshot_sections": {}, "section_statuses": {}, "coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []}, }, ) calls = [] def fake_post(endpoint, request_item): calls.append((endpoint, request_item["section"], request_item.get("fanout_value"))) return {"Status": "Pass", "Payload": {"ok": True}}, 1, [] monkeypatch.setattr(adapter, "_try_official_python_bridge_snapshot_bundle", lambda case: {"available": False, "status": "blocked", "snapshot_sections": {}, "section_statuses": {}, "coverage": {"source_mode": "official_python_bridge_bundle", "filled_sections": []}}) monkeypatch.setattr(adapter, "_post_official_snapshot_section", fake_post) ticks = iter([100.0, 100.1, 104.2]) monkeypatch.setattr(adapter.time, "monotonic", lambda: next(ticks)) result = adapter._run_official_full_snapshot_case({ "year": 1990, "month": 4, "day": 17, "hour": 14, "minute": 59, "lat": 36.42, "lon": 114.2, "tz": 8, }) assert calls assert result["status"] in {"ok", "partial"} assert result["source_metadata"]["official_python_bundle"]["status"] == "official_snapshot_budget_exhausted" def test_vedastro_service_adapter_returns_controlled_unconfigured_status() -> None: env = os.environ.copy() env.pop("VEDASTRO_API_ENDPOINT", None) env.pop("VEDASTRO_ENABLE_NETWORK", None) env["JYOTISH_SKIP_LOCAL_ENV"] = "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, ) 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 False assert report["status"] == "service_endpoint_not_configured" assert "VEDASTRO" in report["reason"] assert report["source_metadata"]["transport"] == "http_json_service_boundary" assert report["source_metadata"]["retry_policy"]["max_attempts"] >= 1 assert report["source_metadata"]["provenance_mode"] == "external_service_candidate" def test_ephemeris_adapter_contract_can_call_vedastro_candidate_stub() -> None: env = os.environ.copy() env.pop("VEDASTRO_API_ENDPOINT", None) env.pop("VEDASTRO_ENABLE_NETWORK", None) env["JYOTISH_SKIP_LOCAL_ENV"] = "1" completed = subprocess.run( [sys.executable, "scripts/ephemeris_adapter_contract.py", "vedastro_service_adapter_candidate"], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, env=env, ) assert completed.returncode == 0, completed.stderr or completed.stdout report = json.loads(completed.stdout) assert report["candidate_backend"] == "vedastro_service_adapter_candidate" first = report["rows"][0]["candidate_backend"] assert first["backend"] == "vedastro_service_adapter_candidate" assert first["status"] == "service_endpoint_not_configured" def test_vedastro_service_adapter_builds_request_preview_before_real_network_use() -> None: env = os.environ.copy() env["VEDASTRO_API_ENDPOINT"] = "https://example.invalid/vedastro" env.pop("VEDASTRO_ENABLE_NETWORK", None) env["JYOTISH_SKIP_LOCAL_ENV"] = "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, ) assert completed.returncode == 0, completed.stderr or completed.stdout report = json.loads(completed.stdout) assert report["backend"] == "vedastro_service_adapter_candidate" assert report["status"] == "network_execution_disabled" assert report["request_preview"]["year"] == 1990 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_builds_range_scan_preview_before_real_network_use() -> None: env = os.environ.copy() env["VEDASTRO_API_ENDPOINT"] = "https://example.invalid/vedastro" env.pop("VEDASTRO_ENABLE_NETWORK", None) env["JYOTISH_SKIP_LOCAL_ENV"] = "1" completed = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2031-01-01", ], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, env=env, ) assert completed.returncode == 0, completed.stderr or completed.stdout report = json.loads(completed.stdout) assert report["backend"] == "vedastro_service_adapter_candidate" assert report["status"] == "network_execution_disabled" assert report["request_preview"]["operation"] == "range_scan" assert report["request_preview"]["vedastro_event_method"] == "SearchEvents" assert report["request_preview"]["domain"] == "marriage" assert report["request_preview"]["start_date"] == "2026-01-01" assert report["request_preview"]["end_date"] == "2031-01-01" assert report["request_preview"]["event_model"] == "vedastro_events_at_range_candidate" assert report["source_metadata"]["endpoint"] == "https://example.invalid/vedastro" def test_vedastro_range_scan_unconfigured_still_returns_official_search_events_preview() -> None: env = os.environ.copy() env.pop("VEDASTRO_API_ENDPOINT", None) env.pop("VEDASTRO_ENABLE_NETWORK", None) env["JYOTISH_SKIP_LOCAL_ENV"] = "1" completed = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "wealth", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2031-01-01", ], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, env=env, ) assert completed.returncode == 0, completed.stderr or completed.stdout report = json.loads(completed.stdout) assert report["status"] == "service_endpoint_not_configured" assert report["request_preview"]["operation"] == "range_scan" assert report["request_preview"]["vedastro_event_method"] == "SearchEvents" assert report["request_preview"]["domain"] == "wealth" assert report["request_preview"]["official_request_profile"]["endpoint_path"] == "/Calculate/SearchEvents" assert report["request_preview"]["official_request_profile"]["method"] == "POST" assert report["request_preview"]["official_request_profile"]["headers"] == {"Content-Type": "application/json"} assert report["request_preview"]["official_request_profile"]["body"]["Ayanamsa"] == "lahiri" assert report["request_preview"]["official_request_profile"]["body"]["EventTagList"] == ["LendingMoney", "BorrowingMoney", "BuyingSelling", "General"] assert "AtTime" not in report["request_preview"]["official_request_profile"]["body"] assert report["request_preview"]["official_request_profile"]["body"]["StartTime"]["StdTime"] == "12:00 01/01/2026 +08:00" assert report["request_preview"]["official_request_profile"]["body"]["EndTime"]["StdTime"] == "12:00 01/01/2031 +08:00" assert report["request_preview"]["official_request_profile"]["body"]["PrecisionHours"] == 100 assert report["request_preview"]["live_sampling_request_profile"]["body"]["AtTime"]["StdTime"] == "12:00 01/01/2026 +08:00" def test_vedastro_schema_declares_official_search_events_live_contract() -> None: completed = subprocess.run( [sys.executable, "scripts/vedastro_service_adapter.py", "--print-schema"], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, ) assert completed.returncode == 0, completed.stderr or completed.stdout report = json.loads(completed.stdout) contract = report["official_search_events_profile_contract"] assert contract["route_template"] == "/Calculate/SearchEvents" assert contract["method"] == "POST" assert contract["optional_auth_header"] == "x-api-key" assert "BirthTime" in contract["body_fields"] assert "EventTagList" in contract["body_fields"] assert "AtTime | StartTime + EndTime + PrecisionHours" in contract["range_mode_fields"] def test_vedastro_service_adapter_posts_official_search_events_contract() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 assert self.path == "/api/Calculate/SearchEvents" assert self.headers.get("Content-Type") == "application/json" assert self.headers.get("x-api-key") == "sk_live_test" length = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(length).decode("utf-8")) assert payload["Ayanamsa"] == "lahiri" assert payload["EventTagList"] == ["Marriage", "Personal", "General"] assert payload["BirthTime"]["StdTime"] == "12:00 01/01/1990 +08:00" assert payload["AtTime"]["StdTime"] == "12:00 01/01/2026 +08:00" assert "StartTime" not in payload response = { "Status": "Pass", "Payload": [ { "Name": "JupiterSupportsMarriageAxis", "StartTime": "2026-05-01", "EndTime": "2026-06-01", "EventTags": ["marriage", "transit"], "Nature": "Good", } ], } 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}/api" env["VEDASTRO_ENABLE_NETWORK"] = "1" env["VEDASTRO_API_KEY"] = "sk_live_test" completed = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2026-01-01", ], 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"] == "ok" assert report["source_metadata"]["endpoint"].endswith("/api") assert report["source_metadata"]["official_endpoint_path"] == "/Calculate/SearchEvents" assert report["source_metadata"]["official_request_profile"]["method"] == "POST" metadata_headers = report["source_metadata"]["official_request_profile"]["headers"] assert metadata_headers == {"Content-Type": "application/json"} assert report["source_metadata"]["official_request_profile_hash"] assert report["source_metadata"]["request_hash"] != report["source_metadata"]["official_request_profile_hash"] 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_can_normalize_mock_range_scan_response() -> None: seen_at_times: list[str] = [] 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["Ayanamsa"] == "lahiri" assert payload["EventTagList"] == ["Marriage", "Personal", "General"] assert payload["BirthTime"]["StdTime"] == "12:00 01/01/1990 +08:00" seen_at_times.append(payload["AtTime"]["StdTime"]) response = { "events": [ { "id": "GocharJupiterIn7th", "name": "Jupiter supports marriage axis", "start": "2026-05-01", "end": "2026-06-01", "score": 72, "tags": ["marriage", "transit"], } ], "source_metadata": { "service": "mock-vedastro", "version": "range-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", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2031-01-01", ], 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["operation"] == "range_scan" assert report["domain"] == "marriage" assert report["event_count"] == 1 assert report["top_event"] == { "event_id": "GocharJupiterIn7th", "signal_key": "gochar_jupiter_7th_marriage", "signal_label": "Jupiter in 7th marriage window", "signal_family": "marriage_trigger", "score": 72, "start": "2026-05-01", "end": "2026-06-01", "tags": ["marriage", "transit"], } assert report["evidence_ledger"][0]["event_id"] == "GocharJupiterIn7th" assert report["evidence_ledger"][0]["signal_key"] == "gochar_jupiter_7th_marriage" assert report["evidence_ledger"][0]["signal_label"] == "Jupiter in 7th marriage window" assert report["evidence_ledger"][0]["signal_family"] == "marriage_trigger" assert report["evidence_ledger"][0]["domain"] == "marriage" assert report["evidence_ledger"][0]["score"] == 72 assert report["evidence_ledger"][0]["raw"]["name"] == "Jupiter supports marriage axis" assert report["source_metadata"]["endpoint"].startswith("http://127.0.0.1:") assert report["source_metadata"]["sampling_mode"] == "at_time_sweep" assert report["source_metadata"]["sample_count"] >= 2 assert seen_at_times[0] == "12:00 01/01/2026 +08:00" assert len(set(seen_at_times)) >= 2 def test_vedastro_range_scan_records_hashes_and_artifact_path() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 response = { "events": [ { "id": "GocharJupiterIn7th", "name": "Jupiter enters 7th house", "start": "2026-05-01", "end": "2026-06-01", "score": 72, "tags": ["marriage", "transit"], } ], "source_metadata": { "service": "mock-vedastro", "version": "artifact-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", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2031-01-01", ], 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) metadata = report["source_metadata"] assert len(metadata["request_hash"]) == 64 assert len(metadata["response_hash"]) == 64 assert metadata["method"] == "POST" assert metadata["operation"] == "range_scan" assert metadata["vedastro_event_method"] == "SearchEvents" assert metadata["official_endpoint_path"] == "/Calculate/SearchEvents" assert metadata["official_request_profile"]["method"] == "POST" assert metadata["official_request_profile"]["body"]["EventTagList"] == ["Marriage", "Personal", "General"] assert metadata["official_request_profile_hash"] assert metadata["allowlist_domain"] == "marriage" assert metadata["allowlist_event_count"] == 1 assert metadata["filtered_event_count"] == 1 assert metadata["attempt_count"] >= 1 assert metadata["sample_count"] >= 1 assert metadata["sampling_mode"] == "at_time_sweep" artifact_path = ROOT / metadata["artifact_path"] assert artifact_path.exists() artifact = json.loads(artifact_path.read_text(encoding="utf-8")) assert artifact["source_metadata"]["request_hash"] == metadata["request_hash"] assert artifact["source_metadata"]["response_hash"] == metadata["response_hash"] assert artifact["evidence_ledger"][0]["event_id"] == "GocharJupiterIn7th" def test_vedastro_range_scan_retries_transient_http_error() -> None: class Handler(BaseHTTPRequestHandler): attempts = 0 def do_POST(self) -> None: # noqa: N802 Handler.attempts += 1 if Handler.attempts == 1: self.send_response(503) self.send_header("Content-Type", "application/json") self.end_headers() return response = { "events": [ { "id": "GocharJupiterIn7th", "name": "Jupiter enters 7th house", "start": "2026-05-01", "end": "2026-06-01", "score": 72, "tags": ["marriage", "transit"], } ] } 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" env["VEDASTRO_RETRY_BACKOFF_SECONDS"] = "0" completed = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2031-01-01", ], 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"] == "ok" assert report["event_count"] == 1 assert report["source_metadata"]["attempt_count"] >= 2 assert 503 in report["source_metadata"]["retry_error_codes"] assert report["source_metadata"]["sampling_mode"] == "at_time_sweep" def test_vedastro_range_scan_reuses_cached_live_response() -> None: class Handler(BaseHTTPRequestHandler): calls = 0 def do_POST(self) -> None: # noqa: N802 Handler.calls += 1 response = { "events": [ { "id": "GocharJupiterIn7th", "name": "Jupiter enters 7th house", "start": "2026-05-01", "end": "2026-06-01", "score": 72, "tags": ["marriage", "transit"], } ] } 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" env["VEDASTRO_CACHE_TTL_SECONDS"] = "600" env["JYOTISH_SKIP_LOCAL_ENV"] = "1" first = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2026-01-01", ], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, env=env, ) second = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2026-01-01", ], cwd=ROOT, text=True, capture_output=True, timeout=120, check=False, env=env, ) finally: server.shutdown() thread.join(timeout=5) assert first.returncode == 0, first.stderr or first.stdout assert second.returncode == 0, second.stderr or second.stdout first_report = json.loads(first.stdout) second_report = json.loads(second.stdout) assert Handler.calls == 1 assert first_report["source_metadata"].get("cache_hit") is False assert second_report["source_metadata"].get("cache_hit") is True assert second_report["event_count"] == first_report["event_count"] == 1 def test_vedastro_service_adapter_applies_domain_allowlist_to_range_scan_noise() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 response = { "events": [ { "id": "GocharJupiterIn7th", "name": "Jupiter enters 7th house", "start": "2026-05-01", "end": "2026-06-01", "score": 72, "tags": ["marriage", "transit"], }, { "id": "RandomMoonMoodShift", "name": "Random moon mood shift", "start": "2026-05-08", "end": "2026-05-09", "score": 91, "tags": ["emotion", "noise"], }, ] } 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", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2031-01-01", ], 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["event_count"] == 1 assert report["top_event"]["event_id"] == "GocharJupiterIn7th" assert report["evidence_ledger"][0]["event_id"] == "GocharJupiterIn7th" def test_vedastro_service_adapter_preserves_match_metadata_for_official_tag_and_alias_hits() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 response = { "Status": "Pass", "Payload": [ { "Name": "GoodForMarriage", "Nature": "Good", "Description": "Marriage event support.", "StartTime": "2026-05-01", "EndTime": "2026-05-02", "EventTags": ["Marriage"], }, { "Name": "PartnershipBlessingWindow", "Nature": "Good", "Description": "Spouse alignment and relationship blessing.", "StartTime": "2026-05-03", "EndTime": "2026-05-04", "EventTags": ["Personal"], }, ], } 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}/api" env["VEDASTRO_ENABLE_NETWORK"] = "1" completed = subprocess.run( [ sys.executable, "scripts/vedastro_service_adapter.py", "--range-scan", "--domain", "marriage", "--case", "beijing_first_use_demo", "--start-date", "2026-01-01", "--end-date", "2026-12-31", ], 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["event_count"] == 2 exact = {item["event_id"]: item for item in report["evidence_ledger"]} assert exact["GoodForMarriage"]["matched_by"] == "official_tag" assert exact["PartnershipBlessingWindow"]["matched_by"] == "alias" assert exact["GoodForMarriage"]["confidence"] == "medium_high" assert exact["PartnershipBlessingWindow"]["confidence"] == "low" assert report["source_metadata"]["mapping_replay"]["match_counts"]["official_tag"] == 1 assert report["source_metadata"]["mapping_replay"]["match_counts"]["alias"] == 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"] def test_vedastro_service_adapter_classifies_invalid_json_response() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 body = b"not-json" 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["status"] == "invalid_json" def test_vedastro_service_adapter_classifies_timeout() -> None: class Handler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 import time time.sleep(0.2) body = b"{}" 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" env["VEDASTRO_TIMEOUT_SECONDS"] = "0.05" 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"] == "timeout"