732642856
260 lines
11 KiB
Python
260 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_vedastro_evidence_orchestrator_routes_to_minimal_domain_set(monkeypatch) -> None:
|
|
from scripts import vedastro_evidence_orchestrator as orchestrator
|
|
|
|
calls = []
|
|
|
|
def fake_snapshot(case, *, case_id="user_chart"):
|
|
return {
|
|
"backend": "vedastro_service_adapter_candidate",
|
|
"available": True,
|
|
"status": "ok",
|
|
"operation": "official_full_snapshot",
|
|
"primary_source": "vedastro_official",
|
|
"snapshot_sections": {"chart_core": {}, "house_core": {}},
|
|
"source_metadata": {
|
|
"official_python_path": "vedastro_official_capability_runner",
|
|
"official_python_bundle": {
|
|
"status": "ok",
|
|
"coverage": {"source_mode": "official_capability_runner_bundle"},
|
|
},
|
|
"official_full_capability_catalog": {
|
|
"status": "partial",
|
|
"summary": {"catalog_method_count": 641, "executed_method_count": 80},
|
|
"domain_routing": {
|
|
"marriage": {
|
|
"method_count": 8,
|
|
"auto_method_count": 5,
|
|
"needs_user_context_count": 1,
|
|
"needs_user_text_count": 0,
|
|
"blocked_method_count": 2,
|
|
"high_priority_methods": ["SearchEvents", "DasaAtRange"],
|
|
}
|
|
},
|
|
"dynamic_selection": {
|
|
"marriage": {
|
|
"requested_theme": "marriage",
|
|
"selected_methods": [
|
|
{"method": "SearchEvents", "citation_id": "vedastro:marriage:SearchEvents", "execution_policy": "auto"},
|
|
],
|
|
"needs_user_context_methods": [
|
|
{"method": "MatchReport", "citation_id": "vedastro:marriage:MatchReport", "execution_policy": "needs_user_context"},
|
|
],
|
|
"report_reference": {
|
|
"theme": "marriage",
|
|
"citation_ids": ["vedastro:marriage:SearchEvents"],
|
|
"auto_count": 1,
|
|
"needs_user_context_count": 1,
|
|
"blocked_count": 0,
|
|
},
|
|
}
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
def fake_scan(case, domain, start_date, end_date, case_id):
|
|
calls.append((domain, start_date, end_date, case_id, case["year"]))
|
|
return {
|
|
"backend": "vedastro_service_adapter_candidate",
|
|
"available": True,
|
|
"status": "ok",
|
|
"operation": "range_scan",
|
|
"domain": domain,
|
|
"event_count": 1,
|
|
"evidence_ledger": [{"domain": domain, "event_id": f"{domain}_event"}],
|
|
"source_metadata": {"endpoint_host": "api.vedastro.org"},
|
|
}
|
|
|
|
monkeypatch.setattr(orchestrator, "run_official_full_snapshot_for_case", fake_snapshot)
|
|
monkeypatch.setattr(orchestrator, "run_range_scan_for_case", fake_scan)
|
|
|
|
result = orchestrator.orchestrate_vedastro_evidence(
|
|
{
|
|
"year": 1955,
|
|
"month": 2,
|
|
"day": 24,
|
|
"hour": 19,
|
|
"minute": 15,
|
|
"lat": 37.7749,
|
|
"lon": -122.4194,
|
|
"tz": 8.0,
|
|
},
|
|
route="relationship",
|
|
reference_date="2026-06-29",
|
|
)
|
|
|
|
assert [call[0] for call in calls] == ["marriage"]
|
|
assert result["source_metadata"]["auto_ingested_by"] == "VedAstroEvidenceOrchestrator"
|
|
assert result["source_metadata"]["node_coverage"]["strategy"] == "domain_scoped_range_scan"
|
|
assert result["source_metadata"]["official_python_path"] == "vedastro_official_capability_runner"
|
|
assert result["source_metadata"]["official_python_bundle_status"] == "ok"
|
|
assert result["source_metadata"]["official_full_capability_catalog_status"] == "partial"
|
|
assert result["source_metadata"]["official_full_capability_catalog_summary"]["catalog_method_count"] == 641
|
|
assert result["source_metadata"]["official_full_capability_domain_routing"]["marriage"]["auto_method_count"] == 5
|
|
assert result["source_metadata"]["official_full_capability_dynamic_selection"]["marriage"]["report_reference"]["auto_count"] == 1
|
|
assert result["source_metadata"]["official_report_references"]["marriage"]["citation_ids"] == ["vedastro:marriage:SearchEvents"]
|
|
assert result["source_metadata"]["node_coverage"]["official_full_capability_theme_routing"] is True
|
|
assert result["source_metadata"]["node_coverage"]["official_full_capability_dynamic_selection"] is True
|
|
assert result["event_count"] == 1
|
|
|
|
|
|
def test_api_and_mcp_use_shared_vedastro_evidence_orchestrator() -> None:
|
|
api = (ROOT / "scripts" / "jyotish_api_server.py").read_text(encoding="utf-8")
|
|
mcp = (ROOT / "mcp_server.py").read_text(encoding="utf-8")
|
|
|
|
assert "vedastro_evidence_orchestrator" in api
|
|
assert "orchestrate_vedastro_evidence" in api
|
|
assert "vedastro_evidence_orchestrator" in mcp
|
|
assert "orchestrate_vedastro_evidence" in mcp
|
|
|
|
|
|
def test_vedastro_orchestrator_surfaces_official_section_statuses_and_theme_requirements(monkeypatch) -> None:
|
|
from scripts import vedastro_evidence_orchestrator as orchestrator
|
|
|
|
monkeypatch.setattr(
|
|
orchestrator,
|
|
"run_official_full_snapshot_for_case",
|
|
lambda *args, **kwargs: {
|
|
"status": "partial",
|
|
"available": True,
|
|
"official_chart": {"planets": {"Sun": {}}, "ascendant": {"sign": "Leo"}},
|
|
"section_statuses": {"chart_core": "ok", "dasha_all": "ok", "events_overview": "partial"},
|
|
"source_metadata": {},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
orchestrator,
|
|
"run_range_scan_for_case",
|
|
lambda *args, **kwargs: {
|
|
"status": "ok",
|
|
"available": True,
|
|
"event_count": 1,
|
|
"evidence_ledger": [],
|
|
},
|
|
)
|
|
|
|
result = orchestrator.orchestrate_vedastro_evidence(
|
|
{
|
|
"year": 1955,
|
|
"month": 2,
|
|
"day": 24,
|
|
"hour": 19,
|
|
"minute": 15,
|
|
"lat": 37.7749,
|
|
"lon": -122.4194,
|
|
"tz": 8,
|
|
},
|
|
route="relationship",
|
|
reference_date="2026-06-29",
|
|
)
|
|
|
|
assert result["source_metadata"]["official_section_statuses"]["dasha_all"] == "ok"
|
|
assert result["source_metadata"]["theme_requirements"]["route"] == "relationship"
|
|
assert result["source_metadata"]["theme_requirements"]["requires_dual_dasha"] is True
|
|
|
|
|
|
def test_vedastro_orchestrator_surfaces_daily_windows_by_domain(monkeypatch) -> None:
|
|
from scripts import vedastro_evidence_orchestrator as orchestrator
|
|
|
|
monkeypatch.setattr(
|
|
orchestrator,
|
|
"run_official_full_snapshot_for_case",
|
|
lambda *args, **kwargs: {"status": "ok", "source_metadata": {}},
|
|
)
|
|
monkeypatch.setattr(
|
|
orchestrator,
|
|
"run_range_scan_for_case",
|
|
lambda *args, **kwargs: {
|
|
"status": "ok",
|
|
"available": True,
|
|
"event_count": 2,
|
|
"daily_windows": [{"date": "2026-07-18", "domain": "career", "score": 5, "event_count": 2}],
|
|
"top_daily_window": {"date": "2026-07-18", "domain": "career", "score": 5, "event_count": 2},
|
|
"evidence_ledger": [],
|
|
},
|
|
)
|
|
|
|
result = orchestrator.orchestrate_vedastro_evidence(
|
|
{"year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15, "lat": 37.7749, "lon": -122.4194, "tz": 8},
|
|
route="career",
|
|
reference_date="2026-06-30",
|
|
)
|
|
|
|
assert result["daily_windows_by_domain"]["career"][0]["date"] == "2026-07-18"
|
|
assert result["top_daily_window_by_domain"]["career"]["score"] == 5
|
|
|
|
|
|
def test_vedastro_orchestrator_passes_non_core_themes_to_official_catalog(monkeypatch) -> None:
|
|
from scripts import vedastro_evidence_orchestrator as orchestrator
|
|
|
|
seen_snapshot_cases = []
|
|
seen_scan_domains = []
|
|
|
|
def fake_snapshot(case, *, case_id="user_chart"):
|
|
seen_snapshot_cases.append(case)
|
|
return {
|
|
"status": "partial",
|
|
"available": True,
|
|
"section_statuses": {},
|
|
"source_metadata": {
|
|
"official_full_capability_catalog": {
|
|
"status": "partial",
|
|
"summary": {"catalog_method_count": 641, "unknown_method_count": 0},
|
|
"domain_routing": {
|
|
"health": {"method_count": 3, "auto_method_count": 1, "high_priority_methods": ["HealthProblemEvent"]},
|
|
},
|
|
"dynamic_selection": {
|
|
"health": {
|
|
"requested_theme": "health",
|
|
"selected_methods": [
|
|
{
|
|
"method": "HealthProblemEvent",
|
|
"citation_id": "vedastro:health:HealthProblemEvent",
|
|
"execution_policy": "auto",
|
|
}
|
|
],
|
|
"report_reference": {
|
|
"theme": "health",
|
|
"citation_ids": ["vedastro:health:HealthProblemEvent"],
|
|
"auto_count": 1,
|
|
},
|
|
}
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
def fake_scan(case, domain, start_date, end_date, case_id):
|
|
seen_scan_domains.append(domain)
|
|
return {
|
|
"status": "unsupported_range_scan_domain",
|
|
"available": False,
|
|
"reason": f"Unsupported range scan domain: {domain}",
|
|
"event_count": 0,
|
|
"evidence_ledger": [],
|
|
}
|
|
|
|
monkeypatch.setattr(orchestrator, "run_official_full_snapshot_for_case", fake_snapshot)
|
|
monkeypatch.setattr(orchestrator, "run_range_scan_for_case", fake_scan)
|
|
|
|
result = orchestrator.orchestrate_vedastro_evidence(
|
|
{"year": 1955, "month": 2, "day": 24, "hour": 19, "minute": 15, "lat": 37.7749, "lon": -122.4194, "tz": 8},
|
|
route="health",
|
|
reference_date="2026-06-30",
|
|
)
|
|
|
|
assert seen_snapshot_cases[0]["themes"] == ["health"]
|
|
assert seen_scan_domains == ["health"]
|
|
assert result["source_metadata"]["official_report_references"]["health"]["citation_ids"] == [
|
|
"vedastro:health:HealthProblemEvent"
|
|
]
|
|
assert result["source_metadata"]["domain_statuses"]["health"] == "unsupported_range_scan_domain"
|