Harden high-rigor Jyotish timing workflow

This commit is contained in:
732642856
2026-07-01 19:15:35 +08:00
parent 6276309d41
commit 5a76abf8d6
87 changed files with 17685 additions and 382 deletions
File diff suppressed because it is too large Load Diff
+65
View File
@@ -0,0 +1,65 @@
#!/usr/bin/env python3
"""Runtime checks for BPHS validation and D9 entrypoint consistency."""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
ENGINE = ROOT / "scripts" / "jyotish_engine.py"
BASE_BIRTH_ARGS = [
"--year",
"1990",
"--month",
"1",
"--day",
"1",
"--hour",
"12",
"--minute",
"0",
"--lat",
"39.9",
"--lon",
"116.4",
"--tz",
"8",
]
def _run_engine(*args: str) -> dict:
completed = subprocess.run(
[sys.executable, str(ENGINE), *args],
cwd=ROOT,
text=True,
capture_output=True,
check=True,
)
return json.loads(completed.stdout)
def test_validate_bphs_invariants_uses_repo_varga_and_reports_zero_failures() -> None:
completed = subprocess.run(
[sys.executable, "scripts/validate_bphs_invariants.py"],
cwd=ROOT,
text=True,
capture_output=True,
check=False,
)
assert completed.returncode == 0, completed.stdout + completed.stderr
assert "失败: 0" in completed.stdout
assert ".workbuddy/skills/jyotish-vedic-astrology/scripts" not in completed.stdout
def test_varga_cli_d9_uses_same_navamsa_mapping_as_varga_full() -> None:
short = _run_engine("varga", *BASE_BIRTH_ARGS, "--d9")["divisional_charts"]["D9_Navamsa"]
full = _run_engine("varga-full", *BASE_BIRTH_ARGS, "--divisions", "9")["D9_Navamsa"]
assert short["ascendant"] == full["Ascendant"]["sign"]
for planet in ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu"]:
assert short[planet]["sign"] == full[planet]["sign"]
+211
View File
@@ -310,9 +310,19 @@ def test_full_reading_reports_ayanamsa_metadata_and_ai_prompt_pack() -> None:
assert isinstance(relationship_narrative["strengths"], list)
assert isinstance(relationship_narrative["risks"], list)
assert isinstance(relationship_narrative["boundaries"], list)
assert relationship_narrative["monthly_frame"]["primary_state"]["value"]
assert relationship_narrative["monthly_frame"]["manifestation_mode"]["value"]
assert relationship_narrative["monthly_frame"]["friction_source"]["value"]
assert relationship_narrative["monthly_frame"]["time_confidence"]["value"]
assert relationship_narrative["markdown"]
assert "D9" in "".join(relationship_narrative["boundaries"])
assert "dual dasha" in relationship_narrative["markdown"]
career_narrative = prompt_pack["evidence_snapshot"]["career_narrative"]
finance_narrative = prompt_pack["evidence_snapshot"]["finance_narrative"]
assert career_narrative["headline"]
assert finance_narrative["headline"]
assert career_narrative["monthly_frame"]["primary_state"]["value"]
assert finance_narrative["monthly_frame"]["primary_state"]["value"]
vimsopaka_summary = prompt_pack["evidence_snapshot"]["vimsopaka_semantic_summary"]
assert vimsopaka_summary["status"] == "used"
assert isinstance(vimsopaka_summary["highlights"], list)
@@ -329,6 +339,22 @@ def test_full_reading_reports_ayanamsa_metadata_and_ai_prompt_pack() -> None:
assert vedastro_overview["source"] == "vedastro_service_adapter_candidate"
assert vedastro_overview["ingestion_profile"] == "main_entry_overview"
assert vedastro_overview["visibility"] == "user_visible_overview_only"
vedastro_official_snapshot = prompt_pack["evidence_snapshot"]["vedastro_official_full_snapshot"]
assert "official_primary_evidence" in vedastro_official_snapshot
assert "local_supplemental_evidence" in vedastro_official_snapshot
assert "fallback_used" in vedastro_official_snapshot
assert "blocked_items" in vedastro_official_snapshot
assert "conflicts" in vedastro_official_snapshot
assert "strict_workflow_contracts" in vedastro_official_snapshot
assert "strict_workflow_routes_available" in vedastro_official_snapshot
assert "relationship" in vedastro_official_snapshot["strict_workflow_contracts"]
assert "career" in vedastro_official_snapshot["strict_workflow_contracts"]
assert "finance" in vedastro_official_snapshot["strict_workflow_contracts"]
career_contract = vedastro_official_snapshot["strict_workflow_contracts"]["career"]
assert "adjudication_stages" in career_contract
assert "multi_reference_reading_summary" in career_contract
assert "modifier_frame" in career_contract["multi_reference_reading_summary"]
assert vedastro_official_snapshot["strict_workflow_primary_route"] in {"relationship", "career", "finance", None}
vedastro_rows = [row for row in audit_table if row["technique"] == "VedAstro Main Entry Overview"]
assert vedastro_rows
assert vedastro_rows[0]["status"] in {"used", "blocked"}
@@ -341,6 +367,79 @@ def test_full_reading_reports_ayanamsa_metadata_and_ai_prompt_pack() -> None:
assert "后台备选证据池" in prompt_pack["prompt_zh"]
def test_full_reading_summary_exposes_stage_timing_contract() -> None:
result = run_engine(
"full-reading",
"--year", "1990",
"--month", "1",
"--day", "1",
"--hour", "12",
"--minute", "0",
"--lat", "39.9",
"--lon", "116.4",
"--tz", "8",
"--today", "2026-01-01",
"--transit-date", "2026-01-01",
)
summary = result["summary"]
assert "stage_timings" in summary
assert isinstance(summary["stage_timings"], list)
assert summary["stage_timings"]
first = summary["stage_timings"][0]
assert "stage" in first
assert "elapsed_seconds" in first
assert "status" in first
assert isinstance(first["elapsed_seconds"], (int, float))
assert "slowest_stages" in summary
assert isinstance(summary["slowest_stages"], list)
assert summary["stage_timing_enabled"] is True
def test_full_reading_summary_exposes_unified_stage_groups() -> None:
result = run_engine(
"full-reading",
"--year", "1990",
"--month", "1",
"--day", "1",
"--hour", "12",
"--minute", "0",
"--lat", "39.9",
"--lon", "116.4",
"--tz", "8",
"--today", "2026-01-01",
"--transit-date", "2026-01-01",
)
summary = result["summary"]
assert summary["stage_contract_version"] == 1
assert isinstance(summary["stage_groups"], list)
assert any(group["group"] == "official_evidence" for group in summary["stage_groups"])
assert summary["cache_recommendations"]["api_chart_response"] == "recommended"
assert summary["async_recommendations"]["chart_async_optional"] is True
def test_full_reading_prompt_pack_carries_compact_technique_audit_summary() -> None:
result = run_engine(
"full-reading",
"--year", "REDACTED_YEAR",
"--month", "4",
"--day", "17",
"--hour", "14",
"--minute", "49",
"--lat", "36.42",
"--lon", "114.2",
"--tz", "8",
"--today", "2026-06-09",
"--transit-date", "2026-06-09",
)
career = result["ai_prompt_pack"]["evidence_snapshot"]["strict_workflow_contracts"]["career"]
assert "technique_audit_summary" in career
assert career["technique_audit_summary"]["functional_benefic_malefic"]["gate"] == "hard"
assert "audit_gate_frame" in career["multi_reference_reading_summary"]
def test_full_reading_generates_guided_topics_from_real_evidence() -> None:
result = run_engine(
"full-reading",
@@ -358,6 +457,7 @@ def test_full_reading_generates_guided_topics_from_real_evidence() -> None:
topics = result["modules"]["guided_topics"]
assert len(topics) >= 3
assert result["summary"]["guided_topics"] == topics
assert result["ai_prompt_pack"]["evidence_snapshot"]["guided_topics"] == topics
for topic in topics[:3]:
@@ -366,6 +466,10 @@ def test_full_reading_generates_guided_topics_from_real_evidence() -> None:
assert topic["reality_value"]
assert topic["why_worth_exploring"]
assert topic["evidence"]
assert "strict_audit_gate" in topic
assert topic["strict_audit_gate"]["functional_benefic_malefic"]["gate"] == "hard"
assert topic["strict_audit_gate"]["relevant_vargas"]["gate"] == "hard"
assert topic["strict_audit_gate"]["vimshottari_narayana_crosscheck"]["gate"] == "hard"
assert topic["confidence"] in {"high", "medium", "low"}
assert topic["vedastro"]["status"] in {"used", "blocked", "not_available"}
assert topic["suggested_questions"]
@@ -376,6 +480,113 @@ def test_full_reading_generates_guided_topics_from_real_evidence() -> None:
assert any(topic["id"] == "birth_time_rectification" for topic in topics)
def test_full_reading_guided_topics_can_carry_official_day_signal_summary() -> None:
result = {
"modules": {
"dasha": {
"current_dasha": {
"lord": "Mercury",
"antardasha": {"lord": "Sun"},
"start": "2026-01-01",
"end": "2026-12-31",
}
},
"dasa_convergence": {
"domain_activations": {
"career_status": {"convergence_level": "L2"},
"marriage_partnership": {"convergence_level": "L1"},
}
},
"chart": {"planets": {"Ketu": {"house": 10}}},
"functional_benefic_malefic": {
"status": "used",
"functional_benefics": ["Mars", "Jupiter", "Sun"],
"functional_malefics": ["Venus", "Mercury", "Saturn"],
},
"career_strict_evidence": {
"technique_audit_summary": {
"functional_benefic_malefic": {"gate": "hard", "used": True},
"relevant_vargas": {"gate": "hard", "present_keys": ["d10_dasamsa", "a10_karma_pada"]},
"vimshottari_narayana_crosscheck": {"gate": "hard", "used": True},
"source_priority_boundary": {"gate": "boundary", "official": {}, "local": {}, "fallback_used": [], "blocked_items": [], "conflicts": []},
},
"monthly_adjudication_summary": {
"route": "career",
"primary_state": {"value": "推进"},
"manifestation_mode": {"value": "项目/合作推进"},
"friction_source": {"value": "结构调整"},
"time_confidence": {"value": "day_supported"},
"supporting_days": [
{"date": "2026-07-18", "summary": "事业机会进入日", "confidence": "high"}
],
},
"present_evidence": {
"external_activation": {
"official_day_signals": [
{"date": "2026-07-18", "day_type": "opportunity_entry", "summary": "事业机会进入日", "confidence": "high"}
]
}
},
},
"relationship_strict_evidence": {
"technique_audit_summary": {
"functional_benefic_malefic": {"gate": "hard", "used": True},
"relevant_vargas": {"gate": "hard", "present_keys": ["d9_navamsa", "upapada_lagna"]},
"vimshottari_narayana_crosscheck": {"gate": "hard", "used": True},
"source_priority_boundary": {"gate": "boundary", "official": {}, "local": {}, "fallback_used": [], "blocked_items": [], "conflicts": []},
},
},
"finance_strict_evidence": {
"technique_audit_summary": {
"functional_benefic_malefic": {"gate": "hard", "used": True},
"relevant_vargas": {"gate": "hard", "present_keys": ["d2_hora"]},
"vimshottari_narayana_crosscheck": {"gate": "hard", "used": True},
"source_priority_boundary": {"gate": "boundary", "official": {}, "local": {}, "fallback_used": [], "blocked_items": [], "conflicts": []},
},
},
"vedastro_range_scan_result": {
"status": "ok",
"source_metadata": {"domain_statuses": {"career": "ok", "marriage": "blocked", "wealth": "blocked"}, "domain_event_counts": {"career": 1}},
"top_events_by_domain": {},
},
},
"chart": {"planets": {"Ketu": {"house": 10}}},
"ai_prompt_pack": {"evidence_snapshot": {"strict_workflow_contracts": {}}},
"birth_info": {"time": "REDACTED_TIME"},
}
from guided_topic_discovery import build_guided_topics
topics = build_guided_topics(result)
career = next(topic for topic in topics if topic["id"] == "career_direction")
assert career["official_day_signal_summary"]["top_day"]["date"] == "2026-07-18"
assert career["official_day_signal_summary"]["top_day"]["summary"] == "事业机会进入日"
assert career["monthly_adjudication_summary"]["primary_state"]["value"] == "推进"
assert career["monthly_adjudication_summary"]["supporting_days"][0]["date"] == "2026-07-18"
assert career["strict_adjudication_bundle"]["monthly_adjudication_summary"] == career["monthly_adjudication_summary"]
assert career["strict_adjudication_bundle"]["strict_audit_gate"] == career["strict_audit_gate"]
def test_full_reading_preserves_official_daily_window_fields_in_range_scan_result() -> None:
result = run_engine(
"full-reading",
"--year", "REDACTED_YEAR",
"--month", "4",
"--day", "17",
"--hour", "14",
"--minute", "49",
"--lat", "36.42",
"--lon", "114.2",
"--tz", "8",
"--today", "2026-06-30",
"--transit-date", "2026-06-30",
)
vedastro = result["modules"]["vedastro_range_scan_result"]
assert "daily_windows" in vedastro
assert "top_daily_window" in vedastro
def test_full_reading_auto_attaches_vedastro_main_entry_boundary() -> None:
env = dict(**__import__("os").environ)
env["VEDASTRO_API_ENDPOINT"] = "https://example.invalid/api"
+152
View File
@@ -307,10 +307,34 @@ def test_ai_prompt_pack_panel_exposes_copyable_audit_context() -> None:
assert "AI Prompt Pack 审计上下文已复制" in main
assert "vedastro_overview" in main
assert "renderVedAstroOverviewPromptCard" in main
assert "renderStrictWorkflowContractPanel" in main
assert "renderStrictWorkflowContractCard" in main
assert "strict_workflow_contracts" in main
assert "strict_workflow_routes_available" in main
assert "Official-first Strict Contract" in main
assert "primary route" in main
assert "VedAstro Overview" in main
assert "overview only,不替代长周期精扫" in main
assert ".ai-prompt-pack-actions" in style
assert ".ayanamsa-runtime-status" in style
assert ".ai-prompt-pack-contracts" in style
def test_frontend_consumes_top_reader_contract_in_prompt_pack_and_ai_chat() -> None:
main = read("main.js")
ai_chat = read("ai-chat.js")
style = read("style.css")
assert "adjudication_stages" in main
assert "multi_reference_reading_summary" in main
assert "technique_audit_summary" in main
assert "main_conflicts" in main
assert "Top-reader adjudication" in main
assert "multi_reference_reading_summary" in ai_chat
assert "adjudication_stages" in ai_chat
assert "technique_audit_summary" in ai_chat
assert "【Top Reader Contract】" in ai_chat
assert ".ai-prompt-pack-contract-card" in style
def test_api_bridge_variants_prefer_backend_prompt_pack_context() -> None:
@@ -1039,6 +1063,17 @@ def test_complete_reading_and_ai_prompt_pack_surface_career_wealth_vedastro_over
for token in [
"buildVedAstroOverviewHighlights",
"renderVedAstroOfficialSnapshotPromptCard",
"renderStrictWorkflowContractPanel",
"VedAstro Official Snapshot",
"official_python_path",
"official_bundle_status",
"official_chart_available",
"official_full_capability_catalog_status",
"official_full_capability_catalog_summary",
"strict_workflow_contracts",
"strict_workflow_primary_route",
"strict_workflow_routes_available",
"Career VedAstro overview",
"Wealth VedAstro overview",
"vedastro_overview",
@@ -1501,11 +1536,15 @@ def test_rectification_ui_guides_yes_no_interview_from_existing_event_assets() -
"EVENT_COLLECTION_GUIDE",
"EVENT_CATEGORIES",
"guided_rectification_interview",
"buildRecommendedRectificationQuestions",
"RECTIFICATION_RECOMMENDED_EVENT_QUESTION_MAP",
]:
assert token in rect_engine
for token in [
"rect-interview",
"renderRectificationInterview",
"recommended_events",
'data-rect-answer="yes"',
'data-rect-answer="no"',
'data-rect-answer="other"',
@@ -1516,6 +1555,61 @@ def test_rectification_ui_guides_yes_no_interview_from_existing_event_assets() -
assert token in rect_ui
def test_web_entry_prefers_unified_consultation_workflow() -> None:
bridge = read("api-bridge.js")
public_bridge = read("public/api-bridge.js")
main = read("main.js")
for source in (bridge, public_bridge):
assert "computeConsultationWorkflow" in source
assert "/api/consultation_workflow" in source
assert "entryMode = options.entryMode || 'direct_chart'" in main
assert "entryMode: 'rectification'" in main
def test_result_page_surfaces_workflow_summary_and_provenance_detail() -> None:
main = read("main.js")
style = read("style.css")
html = read("index.html")
for token in [
"workflow-summary-panel",
"renderWorkflowSummaryPanel",
"renderWorkflowSummaryCard",
"renderWorkflowProvenancePanel",
"runtime_planner",
"renderRuntimePlannerPills",
"UnifiedConsultationRuntimePlanner",
"Workflow Provenance",
"本次解读工作流",
"VedAstro official",
"local supplemental",
"fallback",
"routing.focus_techniques",
"unified_orchestrator",
"source_priority",
]:
assert token in main
for token in [
'id="workflow-summary-panel"',
'id="chart-summary"',
'id="provenance-panel"',
]:
assert token in html
for token in [
".workflow-summary-panel",
".workflow-summary-card",
".workflow-summary-grid",
".workflow-summary-pill",
".workflow-provenance-panel",
".workflow-provenance-list",
]:
assert token in style
def test_remedies_ui_keeps_evidence_boundary_and_hidden_json() -> None:
main = read("main.js")
style = read("style.css")
@@ -1532,6 +1626,21 @@ def test_remedies_ui_keeps_evidence_boundary_and_hidden_json() -> None:
assert token in main or token in style
assert "不能替代医疗、法律、投资或心理咨询" in main
def test_ai_chat_context_carries_runtime_planner_summary() -> None:
ai_chat = read("ai-chat.js")
main = read("main.js")
for token in [
"_consultationWorkflow",
"runtime_planner",
"【Runtime Planner】",
"UnifiedConsultationRuntimePlanner",
"sync_steps",
"async_candidates",
]:
assert token in ai_chat
assert "低风险优先" in main
assert "需要谨慎确认" in main
@@ -1992,12 +2101,16 @@ def test_provenance_panchanga_workspace_panel_is_productized() -> None:
assert "_relationshipBoundary" in export_js
assert "_relationshipStrictNarrativeSection" in export_js
assert "relationship_report" in export_js
assert "career_narrative" in export_js
assert "finance_narrative" in export_js
assert "relationship_narrative" in export_js
assert "vimsopaka_semantic_summary" in export_js
assert "functional_benefic_malefic" in export_js
assert "vedastro_overview" in export_js
assert "technique_audit_table" in export_js
assert "relationship_narrative" in main
assert "career_narrative" in main
assert "finance_narrative" in main
assert "vimsopaka_semantic_summary" in main
assert "functional_benefic_malefic:" in main
assert "vedastro_overview:" in main
@@ -2096,6 +2209,12 @@ def test_synastry_relationship_report_template_keeps_public_formalization_candid
assert "/api/report_artifact" in read("api-bridge.js")
assert "_exportInProgress" in main
assert "setExportBusy" in main
assert "const careerNarrative = extras.career_narrative" in export_js
assert "const financeNarrative = extras.finance_narrative" in export_js
assert "sourceChart?.ai_prompt_pack?.evidence_snapshot?.career_narrative" in main
assert "sourceChart?.ai_prompt_pack?.evidence_snapshot?.finance_narrative" in main
assert "career_narrative: careerNarrative" in main
assert "finance_narrative: financeNarrative" in main
for token in [
"TERMINOLOGY_MODE_LABELS",
@@ -2358,8 +2477,10 @@ def test_complete_reading_surfaces_guided_topic_discovery() -> None:
assert "renderGuidedTopicDiscovery(chartData)" in main
assert "function renderGuidedTopicDiscovery" in main
assert "guided_topics" in main
assert "strict_audit_gate" in main
assert "继续深入" in main
assert "数据依据" in main
assert "严谨门槛" in main
assert "适合继续问" in main
assert "data-guided-topic-question" in main
assert ".guided-topic-discovery" in style
@@ -2373,6 +2494,37 @@ def test_guided_topic_questions_reuse_ai_chat_entry() -> None:
assert "openAIChatWithPrompt" in ai_chat
assert "data-guided-topic-question" in main
assert "openAIChatWithPrompt(question)" in main
assert "strict_adjudication_bundle" in main
assert "strict_audit_gate" in main
assert "guided_topic_context" in ai_chat
assert "strict_adjudication_bundle" in ai_chat
assert "official_day_signal_summary" in main
assert "official_day_signal_summary" in ai_chat
assert "monthly_adjudication_summary" in main
assert "monthly_adjudication_summary" in ai_chat
def test_complete_reading_surfaces_career_quick_questions() -> None:
main = read("main.js")
style = read("style.css")
for token in [
"renderCareerQuickQuestions",
"career-quick-questions",
"career-quick-question",
"事业先看这几个",
"我现在适合换方向还是继续深耕?",
"2026 年事业吉利在哪里,不利在哪里?",
"哪些月份适合推进项目、发布产品或谈合作?",
]:
assert token in main
for token in [
".career-quick-questions",
".career-quick-question-list",
".career-quick-question",
]:
assert token in style
def test_real_case_revalidation_is_release_gate_and_accuracy_boundary() -> None:
+294
View File
@@ -0,0 +1,294 @@
#!/usr/bin/env python3
"""Regression tests for reusable historical event backtest reporting."""
from __future__ import annotations
from scripts import historical_event_backtest as backtest
def _payload(events: list[dict]) -> dict:
return {
"subject": {
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8.0,
"node_mode": "mean",
},
"events": events,
}
def _strict_packet(
route: str,
*,
verdict: str,
dominant_label: str | None,
score: int,
blocked: bool = False,
confidence_cap: str = "medium-high",
missing_evidence: list[str] | None = None,
official_level: str = "primary",
source_priority_mode: str = "vedastro_official_primary",
) -> dict:
return {
"routing": {"question_type": route},
"strict_workflow": {
"question_type": route,
"blocked": blocked,
"confidence_cap": confidence_cap,
"missing_evidence": missing_evidence or [],
"present_evidence": {
"vedastro_official_snapshot": {
"level": official_level,
"source": "vedastro_official",
"status": "partial" if official_level == "primary" else "fallback",
},
"source_priority": {
"mode": source_priority_mode,
"priority": [
"vedastro_official_snapshot",
"local_supplemental_modules",
"local_engine_fallback_when_official_blocked",
],
},
},
"event_judgement": {
"event_family": route,
"verdict": verdict,
"dominant_label": dominant_label,
"score": score,
"primary_drivers": ["vimshottari_current", "narayana_current"],
"secondary_context": ["functional_benefic_malefic_used"],
},
"technique_audit": [
{
"technique": "VedAstro Official Full Snapshot",
"status": "used" if official_level == "primary" else "blocked",
"role": "primary_raw_evidence",
}
],
"life_event_graph": {
"version": "life_event_graph_v1",
"route": route,
"dominant_label": dominant_label,
"verdict": verdict,
"event_nodes": [],
},
},
}
def test_backtest_reports_strong_hit_with_official_snapshot_priority(monkeypatch) -> None:
def fake_strict_workflow(**kwargs):
assert kwargs["transit_date"] == "2019-12-15"
return _strict_packet(
"career",
verdict="high_probability_window",
dominant_label="career_status",
score=84,
)
monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow)
report = backtest.build_report(
_payload(
[
{
"id": "career_turn_2019",
"date": "2019-12-15",
"domain": "career",
"expected_label": "career_status",
"summary": "事业逐渐好转",
}
]
)
)
assert report["scope"] == "historical_event_backtest"
assert report["summary"]["total_events"] == 1
assert report["summary"]["strong_hits"] == 1
assert report["summary"]["official_primary_events"] == 1
event = report["events"][0]
assert event["route"] == "career"
assert event["result_class"] == "strong_hit"
assert event["official_snapshot"]["level"] == "primary"
assert event["evidence"]["source_priority_mode"] == "vedastro_official_primary"
assert event["matched_expected_label"] is True
def test_backtest_marks_blocked_and_unsupported_domains(monkeypatch) -> None:
def fake_strict_workflow(**kwargs):
return _strict_packet(
"relationship",
verdict="insufficient_evidence",
dominant_label=None,
score=28,
blocked=True,
confidence_cap="low",
missing_evidence=["d9_navamsa", "upapada_lagna"],
official_level="fallback",
source_priority_mode="local_fallback_only",
)
monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow)
report = backtest.build_report(
_payload(
[
{
"id": "marriage_probe",
"date": "2026-06-09",
"domain": "marriage",
"expected_label": "legal_marriage",
},
{
"id": "move_1999",
"date": "1999-08-01",
"domain": "move",
"summary": "搬家",
},
]
)
)
assert report["summary"]["blocked_events"] == 1
assert report["summary"]["unsupported_domain_events"] == 1
blocked, unsupported = report["events"]
assert blocked["result_class"] == "blocked"
assert blocked["boundary"]["reason"] == "strict_workflow_blocked"
assert unsupported["result_class"] == "unsupported_domain"
assert unsupported["boundary"]["reason"] == "route_not_yet_implemented_for_event_backtest"
def test_backtest_distinguishes_weak_hit_and_miss(monkeypatch) -> None:
def fake_strict_workflow(**kwargs):
if kwargs["transit_date"] == "2025-02-28":
return _strict_packet(
"finance",
verdict="moderate_probability_window",
dominant_label="income_growth",
score=66,
)
return _strict_packet(
"career",
verdict="insufficient_evidence",
dominant_label=None,
score=22,
confidence_cap="low",
)
monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow)
report = backtest.build_report(
_payload(
[
{
"id": "project_end_cashflow",
"date": "2025-02-28",
"domain": "wealth",
"expected_label": "public_wealth_status",
"summary": "项目结束但收到小额定金",
},
{
"id": "career_false_start",
"date": "2026-01-10",
"domain": "career",
"expected_label": "project_manifestation",
"summary": "短期项目未成",
},
]
)
)
assert report["summary"]["weak_hits"] == 1
assert report["summary"]["misses"] == 1
first, second = report["events"]
assert first["result_class"] == "weak_hit"
assert first["matched_expected_label"] is False
assert first["boundary"]["reason"] == "label_mismatch_under_supported_route"
assert second["result_class"] == "miss"
assert second["boundary"]["reason"] == "insufficient_evidence"
def test_backtest_carries_conflicts_and_blocked_items_from_strict_contract(monkeypatch) -> None:
def fake_strict_workflow(**kwargs):
packet = _strict_packet(
"career",
verdict="high_probability_window",
dominant_label="career_status",
score=84,
)
packet["strict_workflow"]["blocked_items"] = ["official_event_radar_partial"]
packet["strict_workflow"]["conflicts"] = [{"type": "official_local_dasha_conflict"}]
return packet
monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow)
report = backtest.build_report(
_payload(
[
{
"id": "career_turn_2019",
"date": "2019-12-15",
"domain": "career",
"expected_label": "career_status",
"summary": "事业逐渐好转",
}
]
)
)
assert report["events"][0]["evidence"]["blocked_items"] == ["official_event_radar_partial"]
assert report["events"][0]["evidence"]["conflicts"] == [{"type": "official_local_dasha_conflict"}]
def test_backtest_carries_top_reader_contract_summary_from_strict_contract(monkeypatch) -> None:
def fake_strict_workflow(**kwargs):
packet = _strict_packet(
"career",
verdict="high_probability_window",
dominant_label="career_status",
score=84,
)
packet["strict_workflow"]["adjudication_stages"] = {
"promise": {"status": "present"},
"activation": {
"status": "present",
"required_timing_systems": ["Vimshottari", "Narayana"],
},
}
packet["strict_workflow"]["multi_reference_reading_summary"] = {
"root_frame": {"signal": "career_promise"},
"modifier_frame": {"functional_benefic_malefic": {"used": True}},
}
packet["strict_workflow"]["main_conflicts"] = [{"type": "official_local_dasha_conflict"}]
return packet
monkeypatch.setattr(backtest.mcp_server, "strict_workflow", fake_strict_workflow)
report = backtest.build_report(
_payload(
[
{
"id": "career_turn_2019",
"date": "2019-12-15",
"domain": "career",
"expected_label": "career_status",
"summary": "事业逐渐好转",
}
]
)
)
event = report["events"][0]
assert event["evidence"]["adjudication_stages"]["activation"]["required_timing_systems"] == [
"Vimshottari",
"Narayana",
]
assert event["evidence"]["multi_reference_reading_summary"]["root_frame"]["signal"] == "career_promise"
assert event["evidence"]["main_conflicts"] == [{"type": "official_local_dasha_conflict"}]
+31
View File
@@ -262,3 +262,34 @@ def test_life_event_graph_surfaces_main_entry_overview_boundary_node() -> None:
"reference_date": "2026-06-29",
"source": "vedastro_service_adapter_candidate",
} in graph["event_nodes"]
def test_life_event_graph_surfaces_ranked_official_day_window_nodes() -> None:
strict = {
"event_judgement": {"event_family": "career", "verdict": "moderate_probability_window", "score": 74},
"present_evidence": {
"external_activation": {
"level": "moderate",
"source": "vedastro_service_adapter_candidate",
"daily_windows": [
{
"date": "2026-07-18",
"domain": "career",
"score": 5,
"confidence": "medium_high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["GocharJupiterAspect10th", "CareerExpansionWindow"],
"top_signal_label": "Career expansion window",
}
],
}
},
"confidence_cap": "medium",
"missing_evidence": [],
"blocked": False,
}
graph = _build_life_event_graph("career", strict)
assert any(node["kind"] == "official_day_window" and node["date"] == "2026-07-18" for node in graph["event_nodes"])
+301
View File
@@ -3,12 +3,23 @@
from __future__ import annotations
import jyotish_engine
from mcp_server import _collect_strict_evidence
def _base_career_result() -> dict:
return {
"modules": {
"chart": {
"ascendant": {"sign": "Leo"},
"planets": {
"Moon": {"status": "中性(Neutral)"},
"Venus": {"status": "中性(Neutral)"},
"Saturn": {"status": "中性(Neutral)"},
"Sun": {"status": "中性(Neutral)"},
},
},
"varga_full": {"D10_Dasamsa": {"summary": "career varga present"}},
"special_lagnas": {"A10_Karma_Pada": {"sign": "Capricorn", "lord": "Saturn"}},
"jaimini": {
@@ -55,6 +66,7 @@ def test_career_collects_a10_amk_karakamsha_as_strict_evidence() -> None:
"a10_active",
"amk_active",
"karakamsha_context",
"functional_benefic_malefic_used",
"argala_support",
"vedastro_range_scan_missing",
]
@@ -73,11 +85,50 @@ def test_career_blocks_label_when_d10_is_missing_but_preserves_jaimini_context()
"a10_active",
"amk_active",
"karakamsha_context",
"functional_benefic_malefic_used",
"argala_support",
"vedastro_range_scan_missing",
]
def test_career_dignity_guardrail_uses_career_relevant_planets_only() -> None:
result = _base_career_result()
result["modules"]["chart"] = {
"ascendant": {"sign": "Leo"},
"planets": {
"Venus": {"status": "落陷取消(Neecha Bhanga)"},
"Saturn": {"status": "中性(Neutral)"},
"Moon": {"status": "中性(Neutral)"},
"Mars": {"status": "极敌(Great Enemy)"},
},
}
strict = _collect_strict_evidence("career", result)
assert strict["present_evidence"]["dignity_guardrail"]["status"] == "caution"
assert strict["present_evidence"]["dignity_guardrail"]["score_delta"] == 5
assert "dignity_supportive_recovery" in strict["event_judgement"]["secondary_context"]
assert "dignity_high_friction" not in strict["event_judgement"]["secondary_context"]
def test_career_dignity_guardrail_detects_conflict_across_career_significators() -> None:
result = _base_career_result()
result["modules"]["chart"] = {
"ascendant": {"sign": "Leo"},
"planets": {
"Venus": {"status": "落陷取消(Neecha Bhanga)"},
"Saturn": {"status": "极敌(Great Enemy)"},
"Moon": {"status": "中性(Neutral)"},
},
}
strict = _collect_strict_evidence("career", result)
assert strict["present_evidence"]["dignity_guardrail"]["status"] == "conflict"
assert strict["present_evidence"]["dignity_guardrail"]["score_delta"] == 0
assert "dignity_conflict" in strict["event_judgement"]["secondary_context"]
def test_career_argala_bridge_uses_tenth_house_as_modifier_only() -> None:
result = _base_career_result()
@@ -175,3 +226,253 @@ def test_career_accepts_complete_shadbala_components_without_cap_penalty() -> No
assert strict["present_evidence"]["shadbala_component_audit"]["status"] == "complete"
assert strict["confidence_cap"] == "medium-high"
assert "shadbala_component_gap" not in strict["event_judgement"]["secondary_context"]
def test_career_strict_contract_marks_a10_as_local_supplement_to_official_primary() -> None:
result = _base_career_result()
result["modules"]["source_priority"] = {"mode": "vedastro_official_primary"}
result["modules"]["vedastro_official_full_snapshot"] = {
"status": "partial",
"available": True,
"official_chart": {"planets": {"Sun": {}}, "ascendant": {"sign": "Leo"}},
"section_statuses": {"chart_core": "ok", "dasha_all": "ok"},
}
strict = _collect_strict_evidence("career", result)
assert strict["official_primary_evidence"]["chart_core"]["status"] == "ok"
assert strict["official_primary_evidence"]["dasha"]["status"] == "ok"
assert strict["local_supplemental_evidence"]["a10_karma_pada"]["role"] == "required_local_supplement"
assert strict["local_supplemental_evidence"]["narayana_current"]["role"] == "required_local_supplement"
def test_career_strict_contract_exposes_adjudication_stages_and_multi_reference_summary() -> None:
strict = _collect_strict_evidence("career", _base_career_result())
assert strict["adjudication_stages"]["promise"]["status"] in {"present", "weak", "missing"}
assert strict["adjudication_stages"]["activation"]["required_timing_systems"] == ["Vimshottari", "Narayana"]
assert strict["adjudication_stages"]["label"]["value"] == strict["event_judgement"]["dominant_label"]
assert "multi_reference_reading_summary" in strict
summary = strict["multi_reference_reading_summary"]
assert "root_frame" in summary
assert "divisional_frame" in summary
assert "visibility_frame" in summary
assert "karaka_frame" in summary
assert "timing_frame" in summary
assert "modifier_frame" in summary
assert "conflict_frame" in summary
def test_career_strict_contract_exposes_monthly_adjudication_summary() -> None:
result = _base_career_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "career",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-07-18",
"domain": "career",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["GocharJupiterAspect10th", "CareerExpansionWindow"],
"top_signal_label": "Career expansion window",
},
{
"date": "2026-07-28",
"domain": "career",
"score": 6,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["TravelForWork", "GocharJupiterAspect10th"],
"top_signal_label": "Travel for work expansion window",
},
],
"top_daily_window": {
"date": "2026-07-28",
"domain": "career",
"score": 6,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["TravelForWork", "GocharJupiterAspect10th"],
"top_signal_label": "Travel for work expansion window",
},
"source_metadata": {},
}
strict = _collect_strict_evidence("career", result)
summary = strict["monthly_adjudication_summary"]
assert summary["route"] == "career"
assert summary["primary_state"]["value"] in {"推进", "启动", "重组", "收束", "观察"}
assert summary["manifestation_mode"]["value"]
assert summary["friction_source"]["value"]
assert summary["time_confidence"]["value"] in {"day_supported", "month_supported", "month_only", "blocked"}
assert isinstance(summary["supporting_days"], list)
assert summary["supporting_days"][0]["date"] == "2026-07-18"
def test_career_strict_contract_exposes_compact_technique_audit_summary() -> None:
strict = _collect_strict_evidence("career", _base_career_result())
audit = strict["technique_audit_summary"]
assert audit["functional_benefic_malefic"]["gate"] == "hard"
assert audit["relevant_vargas"]["gate"] == "hard"
assert audit["vimshottari_narayana_crosscheck"]["gate"] == "hard"
assert audit["source_priority_boundary"]["fallback_used"] == strict["fallback_used"]
def test_career_external_activation_derives_user_readable_day_signals() -> None:
result = _base_career_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "career",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-07-18",
"domain": "career",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["GocharJupiterAspect10th", "CareerExpansionWindow"],
"top_signal_label": "Career expansion window",
},
{
"date": "2026-07-26",
"domain": "career",
"score": 2,
"confidence": "medium",
"event_count": 1,
"signal_families": ["career_pressure"],
"event_ids": ["SaturnIn10thCareerWindow"],
"top_signal_label": "Saturn in 10th career window",
},
],
"top_daily_window": {
"date": "2026-07-18",
"domain": "career",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["GocharJupiterAspect10th", "CareerExpansionWindow"],
"top_signal_label": "Career expansion window",
},
"source_metadata": {},
}
strict = _collect_strict_evidence("career", result)
external = strict["present_evidence"]["external_activation"]
assert external["official_day_signals"][0]["date"] == "2026-07-18"
assert external["official_day_signals"][0]["day_type"] == "opportunity_entry"
assert external["official_day_signals"][0]["summary"] == "事业机会进入日"
assert external["official_day_signals"][1]["day_type"] == "pressure_opportunity"
def test_career_official_day_signals_distinguish_motion_and_closure_risk() -> None:
result = _base_career_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "career",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-07-28",
"domain": "career",
"score": 6,
"confidence": "high",
"event_count": 2,
"signal_families": [],
"event_ids": ["GoodLunarDayForTravel", "GoodSunSignForBuilding"],
"top_signal_label": "Good lunar day for travel",
},
{
"date": "2025-02-28",
"domain": "career",
"score": 4,
"confidence": "medium",
"event_count": 2,
"signal_families": [],
"event_ids": ["BadLunarDayForTravel", "BadForSellingForProfit"],
"top_signal_label": "Bad lunar day for travel",
},
],
"top_daily_window": {
"date": "2026-07-28",
"domain": "career",
"score": 6,
"confidence": "high",
"event_count": 2,
"signal_families": [],
"event_ids": ["GoodLunarDayForTravel", "GoodSunSignForBuilding"],
"top_signal_label": "Good lunar day for travel",
},
"source_metadata": {},
}
strict = _collect_strict_evidence("career", result)
signals = strict["present_evidence"]["external_activation"]["official_day_signals"]
assert signals[0]["day_type"] == "relocation_motion"
assert signals[0]["summary"] == "事业迁移动作日"
assert signals[1]["day_type"] == "closure_risk"
assert signals[1]["summary"] == "事业真正收尾风险日"
def test_career_narrative_payload_forces_monthly_adjudication_layers_into_final_chinese_conclusion() -> None:
result = _base_career_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "career",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-07-18",
"domain": "career",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["GocharJupiterAspect10th", "CareerExpansionWindow"],
"top_signal_label": "Career expansion window",
}
],
"top_daily_window": {
"date": "2026-07-18",
"domain": "career",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["career_trigger"],
"event_ids": ["GocharJupiterAspect10th", "CareerExpansionWindow"],
"top_signal_label": "Career expansion window",
},
"source_metadata": {},
}
strict = _collect_strict_evidence("career", result)
payload = jyotish_engine._build_career_narrative_payload(strict)
assert "事业" in payload["headline"]
assert payload["monthly_frame"]["primary_state"]["value"]
assert payload["monthly_frame"]["manifestation_mode"]["value"]
assert payload["monthly_frame"]["friction_source"]["value"]
assert payload["monthly_frame"]["time_confidence"]["value"]
assert any("月度主状态" in item for item in payload["strengths"])
assert any("阻力来源" in item for item in payload["risks"])
assert "时间置信度" in payload["markdown"]
+316
View File
@@ -3,6 +3,8 @@
from __future__ import annotations
import jyotish_engine
from mcp_server import (
_collect_strict_evidence,
_derive_event_judgement,
@@ -144,6 +146,238 @@ def test_finance_source_diversity_adds_small_bump_without_changing_verdict_band(
assert high_diversity["score"] == low_diversity["score"] + 5
def test_finance_strict_contract_surfaces_official_block_and_local_fallback_usage() -> None:
strict = _collect_strict_evidence("finance", {"modules": {"source_priority": {"mode": "local_fallback_official_blocked"}}})
assert "official_primary_chart_blocked" in strict["blocked_items"]
assert isinstance(strict["fallback_used"], list)
assert isinstance(strict["conflicts"], list)
def test_finance_strict_contract_exposes_adjudication_stages_and_multi_reference_summary() -> None:
strict = _collect_strict_evidence("finance", {"modules": {"source_priority": {"mode": "local_fallback_official_blocked"}}})
assert strict["adjudication_stages"]["promise"]["status"] in {"present", "weak", "missing"}
assert strict["adjudication_stages"]["activation"]["required_timing_systems"] == ["Vimshottari", "Narayana"]
assert "multi_reference_reading_summary" in strict
summary = strict["multi_reference_reading_summary"]
assert "root_frame" in summary
assert "divisional_frame" in summary
assert "visibility_frame" in summary
assert "karaka_frame" in summary
assert "timing_frame" in summary
assert "modifier_frame" in summary
assert "conflict_frame" in summary
def test_finance_strict_contract_exposes_monthly_adjudication_summary() -> None:
strict = _collect_strict_evidence(
"finance",
{
"modules": {
"source_priority": {"mode": "vedastro_official_primary"},
"varga_full": {
"D2_Hora": {"summary": "hora ready"},
"D10_Dasamsa": {"summary": "dasamsa ready"},
},
"shadbala": {
"planets": {
"Venus": {
"components": {
"sthana": 1,
"dig": 1,
"kala": 1,
"chesta": 1,
"naisargika": 1,
"drik": 1,
}
}
}
},
"ashtakavarga": {"house_scores": {"2": {"sav_score": 33}, "11": {"sav_score": 35}}},
"dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Mercury"}},
"narayana_dasha": {"current_dasha": {"sign": "Taurus", "lord": "Venus"}},
"dasa_convergence": {
"domain_activations": {
"wealth_family": {"convergence_level": "L2", "probability": "35-50%"},
"gains_wishes": {"convergence_level": "L1", "probability": "+15-20%"},
}
},
"chart": {"ascendant": {"sign": "Leo"}},
"dhana_yogas": [{"name": "Dhana Yoga"}],
"lakshmi_yoga": {"present": True},
"vedastro_range_scan_result": {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "wealth",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-10-04",
"domain": "wealth",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["wealth_trigger"],
"event_ids": ["GoodForBorrowingMoneyForBusiness", "GoodForLendingMoney"],
"top_signal_label": "Good for borrowing money for business",
},
{
"date": "2026-10-20",
"domain": "wealth",
"score": 2,
"confidence": "medium",
"event_count": 1,
"signal_families": ["wealth_pressure"],
"event_ids": ["BadForSellingForProfit"],
"top_signal_label": "Bad for selling for profit",
},
],
"top_daily_window": {
"date": "2026-10-04",
"domain": "wealth",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["wealth_trigger"],
"event_ids": ["GoodForBorrowingMoneyForBusiness", "GoodForLendingMoney"],
"top_signal_label": "Good for borrowing money for business",
},
"source_metadata": {},
},
}
},
)
summary = strict["monthly_adjudication_summary"]
assert summary["route"] == "finance"
assert summary["primary_state"]["value"] in {"推进", "启动", "整固", "收束", "观察"}
assert summary["manifestation_mode"]["value"]
assert summary["friction_source"]["value"]
assert summary["time_confidence"]["value"] in {"day_supported", "month_supported", "month_only", "blocked"}
assert isinstance(summary["supporting_days"], list)
assert summary["supporting_days"][0]["date"] == "2026-10-04"
def test_finance_summary_modifier_frame_includes_yogi_and_ashtakavarga_only_as_modifiers() -> None:
modules = {
"source_priority": {"mode": "vedastro_official_primary"},
"varga_full": {
"D2_Hora": {"summary": "hora ready"},
"D10_Dasamsa": {"summary": "dasamsa ready"},
},
"shadbala": {"planets": {"Venus": {"components": {"sthana": 1, "dig": 1, "kala": 1, "chesta": 1, "naisargika": 1, "drik": 1}}}},
"ashtakavarga": {"house_scores": {"2": {"sav_score": 33}, "11": {"sav_score": 35}}},
"dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Mercury"}},
"narayana_dasha": {"current_dasha": {"sign": "Taurus", "lord": "Venus"}},
"dasa_convergence": {"domain_activations": {"wealth_family": {"convergence_level": "L1"}}},
"chart": {"ascendant": {"sign": "Leo"}},
"source_priority": {"mode": "vedastro_official_primary"},
"dhana_yogas": [{"name": "Dhana Yoga"}],
"lakshmi_yoga": {"present": True},
}
strict = _collect_strict_evidence("finance", {"modules": modules})
modifier = strict["multi_reference_reading_summary"]["modifier_frame"]
assert "functional_benefic_malefic" in modifier
assert modifier["ashtakavarga_finance_support"]["source"] == "ashtakavarga_house_scores_bridge_v1"
assert modifier["yogi_support"]["role"] == "modifier_only"
def test_finance_strict_contract_compact_audit_marks_dual_dasha_gate() -> None:
strict = _collect_strict_evidence("finance", {"modules": {"source_priority": {"mode": "local_fallback_official_blocked"}}})
audit = strict["technique_audit_summary"]
assert audit["vimshottari_narayana_crosscheck"]["gate"] == "hard"
assert "official_primary_chart_blocked" in audit["source_priority_boundary"]["blocked_items"]
def test_finance_external_activation_derives_wealth_day_signals() -> None:
strict = _collect_strict_evidence(
"finance",
{
"modules": {
"source_priority": {"mode": "vedastro_official_primary"},
"varga_full": {
"D2_Hora": {"summary": "hora ready"},
"D10_Dasamsa": {"summary": "dasamsa ready"},
},
"shadbala": {
"planets": {
"Venus": {
"components": {
"sthana": 1,
"dig": 1,
"kala": 1,
"chesta": 1,
"naisargika": 1,
"drik": 1,
}
}
}
},
"ashtakavarga": {"house_scores": {"2": {"sav_score": 33}, "11": {"sav_score": 35}}},
"dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Mercury"}},
"narayana_dasha": {"current_dasha": {"sign": "Taurus", "lord": "Venus"}},
"dasa_convergence": {
"domain_activations": {
"wealth_family": {"convergence_level": "L2", "probability": "35-50%"},
"gains_wishes": {"convergence_level": "L1", "probability": "+15-20%"},
}
},
"chart": {"ascendant": {"sign": "Leo"}},
"vedastro_range_scan_result": {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "wealth",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-10-04",
"domain": "wealth",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["wealth_trigger"],
"event_ids": ["GoodForBorrowingMoneyForBusiness", "GoodForLendingMoney"],
"top_signal_label": "Good for borrowing money for business",
},
{
"date": "2026-10-20",
"domain": "wealth",
"score": 2,
"confidence": "medium",
"event_count": 1,
"signal_families": ["wealth_pressure"],
"event_ids": ["BadForSellingForProfit"],
"top_signal_label": "Bad for selling for profit",
},
],
"top_daily_window": {
"date": "2026-10-04",
"domain": "wealth",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["wealth_trigger"],
"event_ids": ["GoodForBorrowingMoneyForBusiness", "GoodForLendingMoney"],
"top_signal_label": "Good for borrowing money for business",
},
"source_metadata": {},
},
}
},
)
external = strict["present_evidence"]["external_activation"]
assert external["official_day_signals"][0]["date"] == "2026-10-04"
assert external["official_day_signals"][0]["day_type"] == "opportunity"
assert external["official_day_signals"][0]["summary"] == "财富机会日"
assert external["official_day_signals"][1]["day_type"] == "risk"
def test_finance_deep_ashtakavarga_supports_add_small_score_bump_without_label_override() -> None:
base = _derive_event_judgement(
"finance",
@@ -1164,3 +1398,85 @@ def test_collect_strict_evidence_finance_does_not_add_avayogi_without_external_t
assert strict["present_evidence"].get("avayogi_risk") is None
assert strict["event_judgement"]["score"] == 95
assert "avayogi_active" not in strict["event_judgement"]["secondary_context"]
def test_finance_narrative_payload_forces_monthly_adjudication_layers_into_final_chinese_conclusion() -> None:
strict = _collect_strict_evidence(
"finance",
{
"modules": {
"source_priority": {"mode": "vedastro_official_primary"},
"varga_full": {
"D2_Hora": {"summary": "hora ready"},
"D10_Dasamsa": {"summary": "dasamsa ready"},
},
"shadbala": {
"planets": {
"Venus": {
"components": {
"sthana": 1,
"dig": 1,
"kala": 1,
"chesta": 1,
"naisargika": 1,
"drik": 1,
}
}
}
},
"ashtakavarga": {"house_scores": {"2": {"sav_score": 33}, "11": {"sav_score": 35}}},
"dasha": {"current_dasha": {"mahadasha": "Venus", "antardasha": "Mercury"}},
"narayana_dasha": {"current_dasha": {"sign": "Taurus", "lord": "Venus"}},
"dasa_convergence": {
"domain_activations": {
"wealth_family": {"convergence_level": "L2", "probability": "35-50%"},
"gains_wishes": {"convergence_level": "L1", "probability": "+15-20%"},
}
},
"chart": {"ascendant": {"sign": "Leo"}},
"dhana_yogas": [{"name": "Dhana Yoga"}],
"lakshmi_yoga": {"present": True},
"vedastro_range_scan_result": {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "wealth",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-10-04",
"domain": "wealth",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["wealth_trigger"],
"event_ids": ["GoodForBorrowingMoneyForBusiness", "GoodForLendingMoney"],
"top_signal_label": "Good for borrowing money for business",
}
],
"top_daily_window": {
"date": "2026-10-04",
"domain": "wealth",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["wealth_trigger"],
"event_ids": ["GoodForBorrowingMoneyForBusiness", "GoodForLendingMoney"],
"top_signal_label": "Good for borrowing money for business",
},
"source_metadata": {},
},
}
},
)
payload = jyotish_engine._build_finance_narrative_payload(strict)
assert "财富" in payload["headline"]
assert payload["monthly_frame"]["primary_state"]["value"]
assert payload["monthly_frame"]["manifestation_mode"]["value"]
assert payload["monthly_frame"]["friction_source"]["value"]
assert payload["monthly_frame"]["time_confidence"]["value"]
assert any("月度主状态" in item for item in payload["strengths"])
assert any("阻力来源" in item for item in payload["risks"])
assert "时间置信度" in payload["markdown"]
@@ -176,6 +176,187 @@ def test_relationship_collects_vedastro_range_scan_as_external_activation_contex
assert "external_activation_support" in strict["event_judgement"]["secondary_context"]
def test_relationship_external_activation_exposes_top_daily_window() -> None:
result = _base_relationship_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "marriage",
"evidence_ledger": [],
"daily_windows": [{"date": "2026-08-02", "domain": "marriage", "score": 5, "event_count": 2}],
"top_daily_window": {"date": "2026-08-02", "domain": "marriage", "score": 5, "event_count": 2},
"source_metadata": {},
}
strict = _collect_strict_evidence("relationship", result)
external = strict["present_evidence"]["external_activation"]
assert external["top_daily_window"]["date"] == "2026-08-02"
assert external["daily_windows"][0]["score"] == 5
def test_relationship_external_activation_derives_progress_day_signals() -> None:
result = _base_relationship_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "marriage",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-08-02",
"domain": "marriage",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["marriage_trigger"],
"event_ids": ["GocharJupiterIn7th", "JupiterSupportsMarriageAxis"],
"top_signal_label": "Jupiter in 7th marriage window",
},
{
"date": "2026-08-19",
"domain": "marriage",
"score": 2,
"confidence": "medium",
"event_count": 1,
"signal_families": ["relationship_pressure"],
"event_ids": ["SaturnAspect7thRelationshipWindow"],
"top_signal_label": "Saturn aspecting 7th relationship window",
},
],
"top_daily_window": {
"date": "2026-08-02",
"domain": "marriage",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["marriage_trigger"],
"event_ids": ["GocharJupiterIn7th", "JupiterSupportsMarriageAxis"],
"top_signal_label": "Jupiter in 7th marriage window",
},
"source_metadata": {},
}
strict = _collect_strict_evidence("relationship", result)
external = strict["present_evidence"]["external_activation"]
assert external["official_day_signals"][0]["date"] == "2026-08-02"
assert external["official_day_signals"][0]["day_type"] == "progress"
assert external["official_day_signals"][0]["summary"] == "婚恋推进日"
assert external["official_day_signals"][1]["day_type"] == "risk"
def test_relationship_strict_contract_exposes_official_primary_and_local_supplemental_layers() -> None:
result = _base_relationship_result()
result["modules"]["source_priority"] = {"mode": "vedastro_official_primary"}
result["modules"]["vedastro_official_full_snapshot"] = {
"status": "partial",
"available": True,
"official_chart": {"planets": {"Sun": {}}, "ascendant": {"sign": "Leo"}},
"section_statuses": {"chart_core": "ok", "dasha_all": "ok", "events_overview": "partial"},
}
strict = _collect_strict_evidence("relationship", result)
assert strict["official_primary_evidence"]["chart_core"]["status"] == "ok"
assert strict["official_primary_evidence"]["dasha"]["status"] == "ok"
assert strict["local_supplemental_evidence"]["upapada_lagna"]["role"] == "required_local_supplement"
assert strict["local_supplemental_evidence"]["darakaraka"]["role"] == "required_local_supplement"
assert isinstance(strict["fallback_used"], list)
assert isinstance(strict["blocked_items"], list)
assert isinstance(strict["conflicts"], list)
def test_relationship_strict_contract_exposes_adjudication_stages_and_multi_reference_summary() -> None:
strict = _collect_strict_evidence("relationship", _base_relationship_result())
assert strict["adjudication_stages"]["promise"]["status"] in {"present", "weak", "missing"}
assert strict["adjudication_stages"]["activation"]["required_timing_systems"] == ["Vimshottari", "Narayana"]
assert strict["adjudication_stages"]["label"]["value"] == strict["event_judgement"]["dominant_label"]
summary = strict["multi_reference_reading_summary"]
assert "root_frame" in summary
assert "divisional_frame" in summary
assert "visibility_frame" in summary
assert "karaka_frame" in summary
assert "timing_frame" in summary
assert "modifier_frame" in summary
assert "conflict_frame" in summary
def test_relationship_strict_contract_exposes_monthly_adjudication_summary() -> None:
result = _base_relationship_result()
result["modules"]["vedastro_range_scan_result"] = {
"backend": "vedastro_service_adapter_candidate",
"status": "ok",
"operation": "range_scan",
"domain": "marriage",
"evidence_ledger": [],
"daily_windows": [
{
"date": "2026-08-02",
"domain": "marriage",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["marriage_trigger"],
"event_ids": ["GocharJupiterIn7th", "JupiterSupportsMarriageAxis"],
"top_signal_label": "Jupiter in 7th marriage window",
},
{
"date": "2026-08-19",
"domain": "marriage",
"score": 2,
"confidence": "medium",
"event_count": 1,
"signal_families": ["relationship_pressure"],
"event_ids": ["SaturnAspect7thRelationshipWindow"],
"top_signal_label": "Saturn aspecting 7th relationship window",
},
],
"top_daily_window": {
"date": "2026-08-02",
"domain": "marriage",
"score": 5,
"confidence": "high",
"event_count": 2,
"signal_families": ["marriage_trigger"],
"event_ids": ["GocharJupiterIn7th", "JupiterSupportsMarriageAxis"],
"top_signal_label": "Jupiter in 7th marriage window",
},
"source_metadata": {},
}
strict = _collect_strict_evidence("relationship", result)
summary = strict["monthly_adjudication_summary"]
assert summary["route"] == "relationship"
assert summary["primary_state"]["value"] in {"推进", "启动", "筛选", "收束", "观察"}
assert summary["manifestation_mode"]["value"]
assert summary["friction_source"]["value"]
assert summary["time_confidence"]["value"] in {"day_supported", "month_supported", "month_only", "blocked"}
assert isinstance(summary["supporting_days"], list)
assert summary["supporting_days"][0]["date"] == "2026-08-02"
def test_relationship_summary_modifier_frame_surfaces_label_lift_related_modifiers() -> None:
strict = _collect_strict_evidence("relationship", _base_relationship_result())
modifier = strict["multi_reference_reading_summary"]["modifier_frame"]
assert "functional_benefic_malefic" in modifier
assert modifier["manifestation_split"]["role"] == "modifier_only"
assert "legal_marriage" in modifier["manifestation_split"]["signals"]
def test_relationship_multi_reference_summary_carries_audit_gate_frame() -> None:
strict = _collect_strict_evidence("relationship", _base_relationship_result())
frame = strict["multi_reference_reading_summary"]["audit_gate_frame"]
assert frame["functional_benefic_malefic"]["used"] in {True, False}
assert frame["source_priority_boundary"]["blocked_items"] == strict["blocked_items"]
def test_relationship_synastry_bridge_adds_context_and_small_score_bump_without_overriding_label_gate() -> None:
base = _collect_strict_evidence("relationship", _base_relationship_result())
result = _base_relationship_result()
@@ -425,11 +606,19 @@ def test_relationship_narrative_payload_turns_synastry_taxonomy_into_user_readab
payload = jyotish_engine._build_relationship_narrative_payload(strict)
assert payload["monthly_frame"]["primary_state"]["value"]
assert payload["monthly_frame"]["manifestation_mode"]["value"]
assert payload["monthly_frame"]["friction_source"]["value"]
assert payload["monthly_frame"]["time_confidence"]["value"]
assert "婚恋" in payload["headline"]
assert any("合盘" in item for item in payload["strengths"])
assert any("protective kuta" in item.lower() for item in payload["strengths"])
assert any("dual dasha" in item.lower() for item in payload["risks"])
assert any("legal_marriage" in item for item in payload["boundaries"])
assert "月度主状态" in payload["markdown"]
assert "落地形式" in payload["markdown"]
assert "阻力来源" in payload["markdown"]
assert "时间置信度" in payload["markdown"]
assert "D9" in payload["markdown"]
assert "dual dasha" in payload["markdown"]
+37
View File
@@ -0,0 +1,37 @@
#!/usr/bin/env python3
"""Narayana Dasha regression tests."""
import os
import sys
SCRIPT_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts")
sys.path.insert(0, SCRIPT_DIR)
from narayana_dasha import calc_narayana_antardasha, get_current_narayana_dasha
def _sample_mahadashas():
return [
{"sign": "Aries", "sign_idx": 0, "lord": "Mars", "years": 4, "start_age": 0.0, "end_age": 4.0},
{"sign": "Taurus", "sign_idx": 1, "lord": "Venus", "years": 8, "start_age": 4.0, "end_age": 12.0},
{"sign": "Gemini", "sign_idx": 2, "lord": "Mercury", "years": 12, "start_age": 12.0, "end_age": 24.0},
]
def test_narayana_antardasha_uses_parent_absolute_age_axis():
ads = calc_narayana_antardasha(_sample_mahadashas(), 1)
assert ads[0]["sign"] == "Taurus"
assert ads[0]["start_age"] == 4.0
assert ads[-1]["end_age"] == 12.0
assert round(sum(ad["years"] for ad in ads), 6) == 8.0
def test_current_narayana_dasha_finds_ad_and_pd_inside_md():
current = get_current_narayana_dasha(_sample_mahadashas(), 5.0)
assert current["md"]["sign"] == "Taurus"
assert current["ad"] is not None
assert current["ad"]["start_age"] <= 5.0 < current["ad"]["end_age"]
assert current["pd"] is not None
assert current["pd"]["start_age"] <= 5.0 < current["pd"]["end_age"]
+113
View File
@@ -0,0 +1,113 @@
#!/usr/bin/env python3
"""Regression tests for preflight fragment scan discipline."""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_preflight_fragment_scan_reports_authority_layers_and_risk_buckets() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/preflight_fragment_scan.py",
],
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["scope"] == "preflight_fragment_scan"
assert report["summary"]["authority_layers_scanned"] == 4
assert report["summary"]["production_truth_layer"] == "main_repo_truth"
assert report["summary"]["real_capability_status"] in {
"engineering_surfaces_covered_but_external_accuracy_not_closed",
"needs_manual_review",
}
layers = report["layers"]
assert layers["main_repo_truth"]["status"] == "authoritative"
assert layers["repo_local_drafts"]["status"] == "draft_reference_only"
assert layers["external_work_brain"]["status"] == "recovery_reference_only"
assert layers["distribution_mirror"]["status"] == "mirror_do_not_reverse_sync"
findings = report["findings"]
assert findings["high_value_unpromoted_count"] >= 1
assert findings["redundant_or_mirror_count"] >= 1
assert findings["workspace_residue_count"] >= 1
assert findings["real_capability_risk_count"] >= 1
categories = {item["category"] for item in report["high_value_unpromoted"]}
assert "repo_local_draft" in categories or "external_work_brain" in categories
mirror_paths = [item["path"] for item in report["redundant_or_mirror"]]
assert any(".workbuddy/skills/jyotish-vedic-astrology" in path for path in mirror_paths)
risk_ids = {item["id"] for item in report["real_capability_risks"]}
assert "external_oracle_not_closed" in risk_ids
assert "historical_event_accuracy_not_proven" in risk_ids
def test_preflight_fragment_scan_preserves_audit_capability_and_oracle_boundaries() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/preflight_fragment_scan.py",
],
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)
audit = report["upstream_audits"]
assert audit["capability_audit"]["valid"] is True
assert audit["capability_audit"]["technique_count"] >= 89
assert audit["fragment_audit"]["valid"] is True
assert audit["fragment_audit"]["candidate_count"] == 0
oracle_boundary = report["real_capability_boundary"]["oracle_boundary"]
assert oracle_boundary["scope"] == "external_oracle_boundary_audit"
assert oracle_boundary["production_tuning_recommended"] is False
assert oracle_boundary["dasha_cases"] >= 1
assert oracle_boundary["shadbala_cases"] >= 1
assert oracle_boundary["longitude_cases"] >= 1
def test_preflight_fragment_scan_emits_repo_cleanup_priorities() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/preflight_fragment_scan.py",
],
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)
priorities = report["cleanup_priorities"]
assert len(priorities) >= 2
assert all(item["id"] != "remove_mirror_runtime_dependency" for item in priorities)
assert all("validate_bphs_invariants.py" not in str(item.get("evidence", {})) for item in priorities)
assert any(item["id"] == "triage_workspace_residue" for item in priorities)
assert any(item["id"] == "promote_or_archive_high_value_drafts" for item in priorities)
+50
View File
@@ -0,0 +1,50 @@
#!/usr/bin/env python3
"""Regression tests for research governance and main-chain documentation."""
from __future__ import annotations
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_local_drafts_disposition_table_covers_all_june_drafts() -> None:
draft_dir = ROOT / "docs" / "research" / "local_drafts" / "2026-06"
disposition_doc = ROOT / "docs" / "research" / "local_drafts_2026_06_disposition.md"
text = disposition_doc.read_text(encoding="utf-8")
draft_names = sorted(path.name for path in draft_dir.glob("*.md"))
assert len(draft_names) == 93
missing = [name for name in draft_names if name not in text]
assert missing == []
assert text.count("| promote |") >= 20
assert "| archive |" in text
assert "| reference-only |" in text
assert ".workbuddy" in text
assert "Do not move or delete files in this pass" in text
def test_unique_main_chain_map_names_runtime_entrypoints_and_boundaries() -> None:
doc = ROOT / "docs" / "research" / "unique_main_chain_map_2026_07_01.md"
text = doc.read_text(encoding="utf-8")
required_tokens = [
"mcp_server.py",
"scripts/jyotish_api_server.py",
"scripts/unified_consultation_orchestrator.py",
"scripts/vedastro_service_adapter.py",
"scripts/historical_event_backtest.py",
"SKILL.md",
"references/strict-workflow-router.md",
"VedAstro official snapshot",
"rectification gate",
"historical event backtest",
".workbuddy",
]
for token in required_tokens:
assert token in text
assert "source of truth" in text.lower()
assert "must not import from `.workbuddy`" in text
+45
View File
@@ -0,0 +1,45 @@
#!/usr/bin/env python3
"""Regression tests for runtime import-source boundaries."""
from __future__ import annotations
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
MIRROR_SCRIPTS_TOKEN = ".workbuddy/skills/jyotish-vedic-astrology/scripts"
def test_bphs_validator_does_not_import_varga_from_distribution_mirror() -> None:
script = ROOT / "scripts" / "validate_bphs_invariants.py"
text = script.read_text(encoding="utf-8")
assert MIRROR_SCRIPTS_TOKEN not in text
def test_runtime_entrypoints_do_not_put_distribution_mirror_on_import_path() -> None:
runtime_entrypoints = [
ROOT / "mcp_server.py",
ROOT / "scripts" / "jyotish_api_server.py",
ROOT / "scripts" / "unified_consultation_orchestrator.py",
ROOT / "scripts" / "vedastro_service_adapter.py",
ROOT / "scripts" / "historical_event_backtest.py",
ROOT / "scripts" / "validate_bphs_invariants.py",
]
offenders = [
str(path.relative_to(ROOT))
for path in runtime_entrypoints
if path.exists() and MIRROR_SCRIPTS_TOKEN in path.read_text(encoding="utf-8", errors="ignore")
]
assert offenders == []
def test_mcp_docstring_marks_workbuddy_as_distribution_mirror_not_runtime_source() -> None:
text = (ROOT / "mcp_server.py").read_text(encoding="utf-8", errors="ignore")
assert "Add to ~/.workbuddy/mcp.json" not in text
assert "/.workbuddy/skills/jyotish-vedic-astrology/mcp_server.py" not in text
assert "distribution mirror" in text
assert "reference only" in text
+17
View File
@@ -15,6 +15,7 @@ from shadbala import (
NAISARGIKA_BALA, MIN_REQUIRED, DIG_BALA_HOUSE,
EXALTATION_DEG, DEBILITATION_DEG, FRIENDSHIP, SHADBALA_CONSTANTS_PATH,
)
from scripts.shadbala_oracle_comparison import compare_case
# Standard test chart data
def _sample_planets():
@@ -70,6 +71,16 @@ class TestSthanaBala:
result = calc_sthana_bala('Sun', 5.0, 'Aries', 1)
assert result['drekkana_bala'] in (0, 15)
def test_sapta_d3_exaltation_branch_is_capped_below_d1_exaltation(self):
result = calc_sthana_bala('Sun', 0.0, 'Aries', 1)
assert result['sapta_d1'] == 50.0
assert result['sapta_d3'] == 45.0
def test_sapta_d3_own_sign_branch_is_capped_below_d1_own_sign(self):
result = calc_sthana_bala('Venus', 30.0, 'Taurus', 2)
assert result['sapta_d1'] == 45.0
assert result['sapta_d3'] == 30.0
def test_total_positive(self):
result = calc_sthana_bala('Jupiter', 105.0, 'Cancer', 5)
assert result['total'] > 0
@@ -124,6 +135,12 @@ class TestDigBala:
def test_midrange_positive(self):
assert 0 < calc_dig_bala('Sun', 7) < 60
def test_dig_bala_jupiter_redacted_place_case_needs_better_than_house_only_linear_model(self):
comparison = compare_case("references/oracle/dasha_shadbala_oracle_cases.json", "template_redacted_place_shadbala_raman")
jupiter_dig_gap = comparison["comparison"]["Jupiter"]["components"]["dig"]["abs_diff_rupa"]
assert jupiter_dig_gap < 3.6348
# ── Kala Bala Tests ─────────────────────────────────────────────────
+19 -5
View File
@@ -4,23 +4,37 @@
from __future__ import annotations
from scripts.shadbala_d3_branch_audit import build_report
from scripts.shadbala_oracle_comparison import compare_case
def test_shadbala_d3_branch_audit_points_to_calc_sthana_bala_d3_branch() -> None:
report = build_report("references/oracle/dasha_shadbala_oracle_cases.json")
assert report["scope"] == "shadbala_d3_branch_audit"
assert report["summary"]["row_count"] >= 4
assert report["summary"]["row_count"] >= 1
assert report["summary"]["global_closure_blocked"] is True
assert report["branch_counts"]
assert "direct_exaltation_branch" in report["branch_counts"] or "direct_own_sign_branch" in report["branch_counts"]
assert "direct_own_sign_branch" in report["branch_counts"]
assert all(row["suspected_function"] == "calc_sthana_bala" for row in report["rows"])
def test_shadbala_d3_branch_audit_splits_exaltation_vs_own_drift() -> None:
def test_shadbala_d3_branch_audit_after_d3_exaltation_cap_retargets_to_own_branch() -> None:
report = build_report("references/oracle/dasha_shadbala_oracle_cases.json")
assert report["branch_hotspots"]["direct_exaltation_branch"]["row_count"] >= 1
assert report["branch_hotspots"]["direct_own_sign_branch"]["row_count"] >= 1
assert report["branch_hotspots"]["direct_exaltation_branch"]["avg_abs_component_diff_rupa"] is not None
assert report["branch_hotspots"]["direct_own_sign_branch"]["avg_abs_component_diff_rupa"] is not None
assert "direct_exaltation_branch" not in report["branch_hotspots"]
def test_shadbala_d3_exaltation_cap_reduces_sun_sthana_gap_for_redacted_place_case() -> None:
comparison = compare_case("references/oracle/dasha_shadbala_oracle_cases.json", "template_redacted_place_shadbala_raman")
sun_sthana_gap = comparison["comparison"]["Sun"]["components"]["sthana"]["abs_diff_rupa"]
assert sun_sthana_gap < 3.5207
def test_shadbala_d3_own_sign_cap_reduces_venus_sthana_gap_for_redacted_place_case() -> None:
comparison = compare_case("references/oracle/dasha_shadbala_oracle_cases.json", "template_redacted_place_shadbala_raman")
venus_sthana_gap = comparison["comparison"]["Venus"]["components"]["sthana"]["abs_diff_rupa"]
assert venus_sthana_gap < 1.7084
+2 -2
View File
@@ -10,9 +10,9 @@ def test_shadbala_d3_mapping_audit_distinguishes_mapping_vs_dignity_path() -> No
report = build_report("references/oracle/dasha_shadbala_oracle_cases.json")
assert report["scope"] == "shadbala_d3_mapping_audit"
assert report["summary"]["case_count"] >= 3
assert report["summary"]["case_count"] >= 1
assert report["summary"]["global_closure_blocked"] is True
assert report["rows"]
assert any(row["mapping_matches_engine_sign"] is True for row in report["rows"])
assert any(row["d3_dignity_bucket"] in {"exalted", "own"} for row in report["rows"])
assert any(row["d3_dignity_bucket"] == "own" for row in report["rows"])
assert report["suspected_fault_split"]
@@ -0,0 +1,21 @@
#!/usr/bin/env python3
"""Regression tests for Dig Bala source-of-truth audit."""
from __future__ import annotations
from scripts.shadbala_dig_source_of_truth_audit import build_report
def test_shadbala_dig_source_of_truth_audit_compares_three_candidate_models() -> None:
report = build_report("references/oracle/dasha_shadbala_oracle_cases.json")
assert report["scope"] == "shadbala_dig_source_of_truth_audit"
assert report["summary"]["case_count"] >= 1
assert set(report["candidate_models"]) == {
"current_linear_house_model",
"house_midpoint_angular_model",
"bhava_madhya_angular_model",
}
assert report["summary"]["best_model_by_avg_abs_diff"] in report["candidate_models"]
assert report["rows"]
@@ -0,0 +1,85 @@
#!/usr/bin/env python3
"""Tests for the shared consultation orchestrator contract."""
from __future__ import annotations
from scripts.unified_consultation_orchestrator import UnifiedConsultationOrchestrator
def test_unified_consultation_orchestrator_normalizes_themes_and_route() -> None:
orchestrator = UnifiedConsultationOrchestrator()
themes = orchestrator.normalize_themes(["事业", "relationship", "money"])
route = orchestrator.resolve_route("我想看事业和工作机会", themes)
assert themes == ["career", "marriage", "wealth"]
assert route["question_type"] == "career"
assert route["primary_theme"] == "career"
assert "D10" in route["focus_techniques"]
def test_unified_consultation_orchestrator_exposes_surface_agnostic_contract() -> None:
orchestrator = UnifiedConsultationOrchestrator()
route = orchestrator.resolve_route("When will I marry?", ["marriage"])
contract = orchestrator.shared_contract(
entry_mode="direct_chart",
question="When will I marry?",
themes=["marriage"],
route_packet=route,
surface="skill_mcp",
)
assert contract["name"] == "UnifiedConsultationOrchestrator"
assert contract["surface"] == "skill_mcp"
assert contract["entry_mode"] == "direct_chart"
assert contract["route"]["question_type"] == "relationship"
assert contract["themes"] == ["marriage"]
assert contract["source_priority"]["mode"] == "vedastro_official_snapshot_first"
assert contract["source_priority"]["priority"][0] == "vedastro_official_snapshot"
def test_unified_consultation_orchestrator_builds_runtime_planner() -> None:
orchestrator = UnifiedConsultationOrchestrator()
themes = orchestrator.normalize_themes(["career", "wealth"])
route = orchestrator.resolve_route("请直接排盘并重点看事业收入", themes)
planner = orchestrator.runtime_planner(
entry_mode="direct_chart",
question="请直接排盘并重点看事业收入",
themes=themes,
route_packet=route,
events=[{"id": "career_turn_2019"}],
surface="api_web",
high_rigor=False,
)
assert planner["planner_name"] == "UnifiedConsultationRuntimePlanner"
assert planner["entry_mode"] == "direct_chart"
assert planner["route"]["question_type"] == "career"
assert planner["sync_steps"][0] == "compute_chart"
assert "run_thematic_report" in planner["sync_steps"]
assert planner["async_candidates"][0] == "historical_event_backtest"
assert planner["source_priority"]["priority"][0] == "vedastro_official_snapshot"
assert planner["question_context"]["event_count"] == 1
def test_unified_consultation_orchestrator_rectification_entry_runs_gate_first() -> None:
orchestrator = UnifiedConsultationOrchestrator()
themes = orchestrator.normalize_themes(["marriage"])
route = orchestrator.resolve_route("我想先校正出生时间再看婚恋", themes)
planner = orchestrator.runtime_planner(
entry_mode="rectification",
question="我想先校正出生时间再看婚恋",
themes=themes,
route_packet=route,
events=[{"id": "relationship_turn_2015"}],
surface="api_web",
high_rigor=False,
)
assert planner["entry_mode"] == "rectification"
assert planner["sync_steps"][0] == "run_rectification_gate"
assert "compute_chart" in planner["sync_steps"]
assert "run_thematic_report" in planner["sync_steps"]
+139
View File
@@ -0,0 +1,139 @@
from __future__ import annotations
import pytest
from scripts.varga import calc_all_vargas
PLANET_LONS = {
"Sun": 3.5082636623988996,
"Moon": 311.80885383130334,
"Mars": 91.31729371926605,
"Mercury": 338.53260292309903,
"Jupiter": 163.8229749564531,
"Venus": 340.5440654310959,
"Saturn": 304.28763628893995,
"Rahu": 231.03374080496383,
"Ketu": 51.03374080496383,
}
ASC_LON = 133.0814
EXPECTED_VEDASTRO = {
"D2_Hora": {
"Sun": ("Aries", 10.51),
"Moon": ("Gemini", 5.4125),
"Mars": ("Cancer", 3.9383),
"Mercury": ("Pisces", 25.5842),
"Jupiter": ("Capricorn", 11.4542),
"Venus": ("Cancer", 1.6172),
"Saturn": ("Aquarius", 12.8489),
"Rahu": ("Cancer", 3.0872),
"Ketu": ("Capricorn", 3.0867),
"Ascendant": ("Leo", 26.1517),
},
"D4_Turyamsa": {
"Sun": ("Aries", 14.0133),
"Moon": ("Taurus", 17.2167),
"Mars": ("Cancer", 5.2511),
"Mercury": ("Gemini", 4.1122),
"Jupiter": ("Sagittarius", 25.2722),
"Venus": ("Gemini", 12.1567),
"Saturn": ("Aquarius", 17.1322),
"Rahu": ("Taurus", 24.1164),
"Ketu": ("Scorpio", 24.1156),
"Ascendant": ("Scorpio", 22.3033),
},
"D7_Saptamsa": {
"Sun": ("Aries", 24.5233),
"Moon": ("Aries", 22.6289),
"Mars": ("Cancer", 9.1894),
"Mercury": ("Aries", 29.6964),
"Jupiter": ("Sagittarius", 6.7264),
"Venus": ("Taurus", 13.7742),
"Saturn": ("Aquarius", 29.9814),
"Rahu": ("Pisces", 27.2039),
"Ketu": ("Pisces", 27.2019),
"Ascendant": ("Scorpio", 1.5308),
},
"D16_Shodasamsa": {
"Sun": ("Taurus", 26.0533),
"Moon": ("Aquarius", 8.8667),
"Mars": ("Aries", 21.0044),
"Mercury": ("Aries", 16.4489),
"Jupiter": ("Cancer", 11.0889),
"Venus": ("Taurus", 18.6267),
"Saturn": ("Libra", 8.5286),
"Rahu": ("Cancer", 6.4664),
"Ketu": ("Cancer", 6.4622),
"Ascendant": ("Aquarius", 29.2133),
},
"D20_Vimsamsa": {
"Sun": ("Gemini", 10.0664),
"Moon": ("Cancer", 26.0833),
"Mars": ("Aries", 26.2556),
"Mercury": ("Capricorn", 20.5608),
"Jupiter": ("Taurus", 6.3611),
"Venus": ("Pisces", 0.7831),
"Saturn": ("Aquarius", 25.6611),
"Rahu": ("Aquarius", 0.5831),
"Ketu": ("Aquarius", 0.5778),
"Ascendant": ("Leo", 21.5164),
},
"D27_Bhamsa": {
"Sun": ("Cancer", 4.59),
"Moon": ("Leo", 18.7125),
"Mars": ("Aquarius", 5.445),
"Mercury": ("Leo", 20.2572),
"Jupiter": ("Cancer", 13.0875),
"Venus": ("Libra", 14.5575),
"Saturn": ("Capricorn", 25.6425),
"Rahu": ("Cancer", 27.7872),
"Ketu": ("Capricorn", 27.7797),
"Ascendant": ("Pisces", 23.0475),
},
"D30_Trimsamsa": {
"Sun": ("Scorpio", 15.1),
"Moon": ("Sagittarius", 24.125),
"Mars": ("Taurus", 9.3833),
"Mercury": ("Gemini", 15.8417),
"Jupiter": ("Sagittarius", 24.5417),
"Venus": ("Gemini", 16.175),
"Saturn": ("Scorpio", 8.4917),
"Rahu": ("Capricorn", 0.875),
"Ketu": ("Capricorn", 0.8667),
"Ascendant": ("Sagittarius", 2.275),
},
"D45_Akshavedamsa": {
"Sun": ("Virgo", 7.65),
"Moon": ("Capricorn", 21.1875),
"Mars": ("Taurus", 29.075),
"Mercury": ("Sagittarius", 23.7622),
"Jupiter": ("Leo", 21.8125),
"Venus": ("Pisces", 24.2622),
"Saturn": ("Aquarius", 12.7372),
"Rahu": ("Pisces", 16.3122),
"Ketu": ("Pisces", 16.3),
"Ascendant": ("Pisces", 18.4125),
},
"D60_Shashtyamsa": {
"Sun": ("Scorpio", 0.2),
"Moon": ("Capricorn", 18.25),
"Mars": ("Virgo", 18.7667),
"Mercury": ("Leo", 1.6833),
"Jupiter": ("Sagittarius", 19.0833),
"Venus": ("Sagittarius", 2.35),
"Saturn": ("Libra", 16.9833),
"Rahu": ("Taurus", 1.75),
"Ketu": ("Scorpio", 1.7333),
"Ascendant": ("Libra", 4.55),
},
}
@pytest.mark.parametrize("varga_key, expected_rows", EXPECTED_VEDASTRO.items())
def test_varga_vedastro_mode_matches_official_golden_case(varga_key, expected_rows):
result = calc_all_vargas(PLANET_LONS, ASC_LON, mode="vedastro")
chart = result[varga_key]
for body, (sign, degree) in expected_rows.items():
assert chart[body]["sign"] == sign, (varga_key, body)
assert chart[body]["degree_in_sign"] == pytest.approx(degree, abs=0.02), (varga_key, body)
@@ -11,6 +11,55 @@ def test_vedastro_evidence_orchestrator_routes_to_minimal_domain_set(monkeypatch
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 {
@@ -24,6 +73,7 @@ def test_vedastro_evidence_orchestrator_routes_to_minimal_domain_set(monkeypatch
"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(
@@ -44,6 +94,15 @@ def test_vedastro_evidence_orchestrator_routes_to_minimal_domain_set(monkeypatch
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
@@ -55,3 +114,79 @@ def test_api_and_mcp_use_shared_vedastro_evidence_orchestrator() -> None:
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": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"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": REDACTED_YEAR, "month": 4, "day": 17, "hour": 14, "minute": 49, "lat": 36.42, "lon": 114.2, "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
@@ -381,3 +381,92 @@ def test_strict_workflow_auto_ingests_live_vedastro_external_technique_as_contex
assert strict["present_evidence"]["external_technique_evidence"]["level"] == "context_only"
assert strict["present_evidence"]["external_technique_evidence"]["methods"] == ["CalculateShadbala"]
assert "external_technique_evidence" in strict["event_judgement"]["secondary_context"]
def test_strict_workflow_reports_unified_orchestrator_metadata() -> None:
with mock.patch.object(mcp_server, "_run_engine") as run_engine, mock.patch.object(
mcp_server,
"_maybe_attach_vedastro_evidence",
) as attach_vedastro:
run_engine.return_value = {"modules": _finance_modules()}
attach_vedastro.return_value = {"modules": _finance_modules()}
result = mcp_server.strict_workflow(
question="我的财务今年如何?",
year=REDACTED_YEAR,
month=4,
day=17,
hour=14,
minute=49,
lat=36.42,
lon=114.2,
tz=8.0,
age=33,
transit_date="2026-06-29",
node_mode="mean",
)
unified = result["unified_orchestrator"]
assert unified["name"] == "UnifiedConsultationOrchestrator"
assert unified["surface"] == "skill_mcp"
assert unified["route"]["question_type"] == "finance"
assert unified["source_priority"]["priority"][0] == "vedastro_official_snapshot"
planner = result["runtime_planner"]
assert planner["planner_name"] == "UnifiedConsultationRuntimePlanner"
assert planner["surface"] == "skill_mcp"
assert planner["route"]["question_type"] == "finance"
assert planner["sync_steps"][0] == "compute_chart"
def test_strict_workflow_uses_shared_consultation_executor(monkeypatch) -> None:
fake_result = {
"success": True,
"endpoint": "consultation_workflow",
"entry_mode": "direct_chart",
"routing": {"question_type": "finance"},
"unified_orchestrator": {
"name": "UnifiedConsultationOrchestrator",
"surface": "skill_mcp",
"route": {"question_type": "finance"},
},
"runtime_planner": {
"planner_name": "UnifiedConsultationRuntimePlanner",
"surface": "skill_mcp",
"entry_mode": "direct_chart",
"route": {"question_type": "finance"},
"sync_steps": ["compute_chart", "run_rectification_gate", "run_thematic_report"],
"executed_steps": ["compute_chart", "run_rectification_gate", "run_thematic_report"],
"skipped_steps": ["run_historical_event_backtest"],
},
"chart": {"modules": _finance_modules()},
"rectification": {"success": True, "endpoint": "rectification_gate"},
"thematic_report": {"success": True, "endpoint": "thematic_report"},
"vedastro_official": {"available": True},
}
seen = {}
def fake_executor(**kwargs):
seen.update(kwargs)
return fake_result
monkeypatch.setattr(mcp_server, "_execute_mcp_consultation_workflow", fake_executor)
result = mcp_server.strict_workflow(
question="我的财务今年如何?",
year=REDACTED_YEAR,
month=4,
day=17,
hour=14,
minute=49,
lat=36.42,
lon=114.2,
tz=8.0,
age=33,
transit_date="2026-06-29",
node_mode="mean",
)
assert seen["entry_mode"] == "direct_chart"
assert seen["question"] == "我的财务今年如何?"
assert result["runtime_planner"]["surface"] == "skill_mcp"
assert result["routing"]["question_type"] == "finance"
@@ -26,6 +26,7 @@ def test_vedastro_method_catalog_sync_schema_is_declared() -> None:
assert report["scope"] == "official_vedastro_method_catalog"
assert "sync_tags" in report["operations"]
assert "write_snapshot" in report["operations"]
assert "sync_python_capabilities" in report["operations"]
def test_vedastro_method_catalog_sync_can_write_stubbed_snapshot() -> None:
@@ -65,3 +66,55 @@ def test_vedastro_method_catalog_sync_can_write_stubbed_snapshot() -> None:
snapshot = json.loads(output_path.read_text(encoding="utf-8"))
assert "Marriage" in snapshot["tag_groups"]
assert snapshot["summary"]["method_count"] == 2
def test_vedastro_method_catalog_sync_can_build_stubbed_python_capability_registry() -> None:
env = os.environ.copy()
env["VEDASTRO_METHOD_CATALOG_STUB"] = json.dumps(
{
"source": "stubbed_catalog",
"tag_groups": {},
"python_capabilities": [
{
"method": "AllPlanetData",
"signature": "(planetName, time)",
"bucket": "planet_time",
"parameter_names": ["planetName", "time"],
"callable": True,
},
{
"method": "AllHouseData",
"signature": "(houseName, time)",
"bucket": "house_name_time",
"parameter_names": ["houseName", "time"],
"callable": True,
},
{
"method": "GetAllEventDataGroupedByTag",
"signature": "()",
"bucket": "zero_arg",
"parameter_names": [],
"callable": True,
},
],
}
)
completed = subprocess.run(
[sys.executable, "scripts/vedastro_method_catalog_sync.py"],
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["summary"]["python_capability_count"] == 3
assert report["summary"]["python_callable_count"] == 3
assert report["summary"]["python_signature_bucket_count"] >= 2
assert any(item["method"] == "AllPlanetData" for item in report["python_capabilities"])
buckets = report["python_signature_buckets"]
assert buckets["planet_time"]["count"] == 1
assert "AllPlanetData" in buckets["planet_time"]["examples"]
@@ -0,0 +1,376 @@
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_official_capability_runner_schema_is_declared() -> None:
completed = subprocess.run(
[sys.executable, "scripts/vedastro_official_capability_runner.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["runner"] == "vedastro_official_capability_runner"
assert report["primary_source"] == "vedastro_python_bridge"
assert "run_bucket" in report["operations"]
assert "run_selected_methods" in report["operations"]
assert "run_snapshot_bundle" in report["operations"]
assert "run_full_capability_catalog" in report["operations"]
def test_official_capability_runner_can_execute_stubbed_selected_methods() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_official_capability_runner.py",
"--methods-json",
json.dumps(["AllPlanetData", "AllHouseData", "GetAllEventDataGroupedByTag"]),
"--birth-json",
json.dumps(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
}
),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**dict(**__import__("os").environ),
"VEDASTRO_OFFICIAL_CAPABILITY_RUNNER_STUB": json.dumps(
{
"AllPlanetData": {"available": True, "status": "ok", "result": {"PlanetRasiD1Sign": {"Name": "Aries"}}},
"AllHouseData": {"available": True, "status": "ok", "result": {"HouseBhavaChalitSign": {"Name": "Leo"}}},
"GetAllEventDataGroupedByTag": {"available": True, "status": "ok", "result": {"Marriage": []}},
}
),
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["summary"]["requested_method_count"] == 3
assert report["summary"]["executed_method_count"] == 3
assert report["summary"]["ok_count"] == 3
assert report["results"]["AllPlanetData"]["status"] == "ok"
assert report["results"]["AllHouseData"]["status"] == "ok"
def test_official_capability_runner_can_execute_stubbed_snapshot_bundle() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_official_capability_runner.py",
"--bundle",
"official_full_snapshot",
"--birth-json",
json.dumps(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
}
),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**dict(**__import__("os").environ),
"VEDASTRO_OFFICIAL_CAPABILITY_RUNNER_STUB": json.dumps(
{
"AllPlanetData:Sun": {
"available": True,
"status": "ok",
"result": {"PlanetRasiD1Sign": {"Name": "Aries"}},
},
"AllHouseData:House1": {
"available": True,
"status": "ok",
"result": {"HouseBhavaChalitSign": {"Name": "Leo"}},
},
"DasaAtRange": {"available": True, "status": "ok", "result": [{"Lord": "Venus"}]},
"DasaAtTime": {"available": True, "status": "ok", "result": {"Lord": "Venus"}},
"GetCharaDasaAtTime": {"available": True, "status": "ok", "result": {"Sign": "Leo"}},
"AllPlanetStrength": {"available": True, "status": "ok", "result": {"Sun": 527.36}},
"AshtakvargaLifeMap": {"available": True, "status": "ok", "result": {"TotalBindus": 337}},
}
),
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["bundle"] == "official_full_snapshot"
assert report["summary"]["ok_count"] >= 6
assert report["result"]["section_statuses"]["chart_core"] in {"ok", "partial"}
assert report["result"]["section_statuses"]["house_core"] in {"ok", "partial"}
assert report["result"]["snapshot_sections"]["chart_core"]["Sun"]["Payload"]["AllPlanetData"]["PlanetRasiD1Sign"]["Name"] == "Aries"
assert report["result"]["snapshot_sections"]["house_core"]["House1"]["Payload"]["AllHouseData"]["HouseBhavaChalitSign"]["Name"] == "Leo"
def test_official_capability_runner_can_compile_stubbed_full_catalog_without_heavy_calls() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_official_capability_runner.py",
"--bundle",
"official_full_capability_catalog",
"--birth-json",
json.dumps(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
}
),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**dict(**__import__("os").environ),
"VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB": json.dumps(
{
"available": True,
"status": "ok",
"capabilities": [
{
"method": "GetAllEventDataGroupedByTag",
"signature": "()",
"bucket": "zero_arg",
"parameter_names": [],
"callable": True,
},
{
"method": "AllPlanetData",
"signature": "(planetName, time)",
"bucket": "planet_time",
"parameter_names": ["planetName", "time"],
"callable": True,
},
{
"method": "SearchEvents",
"signature": "(birthTime, atTime, eventTagList)",
"bucket": "event_time",
"parameter_names": ["birthTime", "atTime", "eventTagList"],
"callable": True,
},
{
"method": "DasaAtRange",
"signature": "(birthTime, startTime, endTime, levels, precisionHours)",
"bucket": "dasha_at_range",
"parameter_names": ["birthTime", "startTime", "endTime", "levels", "precisionHours"],
"callable": True,
},
{
"method": "AllPlanetDashamamshaSign",
"signature": "(planetName, time)",
"bucket": "planet_time",
"parameter_names": ["planetName", "time"],
"callable": True,
},
{
"method": "MatchReport",
"signature": "(maleBirthTime, femaleBirthTime)",
"bucket": "(maleBirthTime, femaleBirthTime)",
"parameter_names": ["maleBirthTime", "femaleBirthTime"],
"callable": True,
},
],
"buckets": {
"zero_arg": {"count": 1, "examples": ["GetAllEventDataGroupedByTag"]},
"planet_time": {"count": 2, "examples": ["AllPlanetData", "AllPlanetDashamamshaSign"]},
"event_time": {"count": 1, "examples": ["SearchEvents"]},
"dasha_at_range": {"count": 1, "examples": ["DasaAtRange"]},
"(maleBirthTime, femaleBirthTime)": {"count": 1, "examples": ["MatchReport"]},
},
}
),
"VEDASTRO_OFFICIAL_CAPABILITY_RUNNER_STUB": json.dumps(
{
"GetAllEventDataGroupedByTag": {
"available": True,
"status": "ok",
"result": {"Marriage": []},
},
"AllPlanetData": {
"available": True,
"status": "ok",
"result": {"PlanetRasiD1Sign": {"Name": "Aries"}},
},
"SearchEvents": {
"available": True,
"status": "ok",
"result": {"Events": []},
},
"DasaAtRange": {
"available": True,
"status": "ok",
"result": {"Periods": []},
},
"AllPlanetDashamamshaSign": {
"available": True,
"status": "ok",
"result": {"Name": "Capricorn"},
},
}
),
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
assert report["bundle"] == "official_full_capability_catalog"
assert report["summary"]["catalog_method_count"] == 6
assert report["summary"]["executed_method_count"] == 5
assert report["summary"]["unsupported_method_count"] == 1
assert report["coverage"]["source_mode"] == "official_full_capability_catalog"
assert report["method_statuses"]["AllPlanetData"]["status"] == "ok"
assert report["method_statuses"]["SearchEvents"]["domains"] == ["career", "marriage", "wealth", "rectification", "timing"]
assert report["method_statuses"]["SearchEvents"]["execution_policy"] == "auto"
assert report["method_statuses"]["SearchEvents"]["priority"] == "high"
assert report["method_statuses"]["DasaAtRange"]["domains"] == ["career", "marriage", "wealth", "rectification", "timing"]
assert "career" in report["method_statuses"]["AllPlanetDashamamshaSign"]["domains"]
assert "timing" not in report["method_statuses"]["AllPlanetDashamamshaSign"]["domains"]
assert report["domain_routing"]["marriage"]["high_priority_methods"] == ["SearchEvents", "DasaAtRange"]
assert report["domain_routing"]["career"]["auto_method_count"] >= 2
assert report["method_statuses"]["MatchReport"]["status"] == "requires_user_context"
assert report["method_statuses"]["MatchReport"]["execution_policy"] == "needs_user_context"
def test_official_capability_runner_builds_dynamic_theme_selection_and_citations() -> None:
completed = subprocess.run(
[
sys.executable,
"scripts/vedastro_official_capability_runner.py",
"--bundle",
"official_full_capability_catalog",
"--birth-json",
json.dumps(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
"themes": ["career", "marriage"],
}
),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
env={
**dict(**__import__("os").environ),
"VEDASTRO_OFFICIAL_CAPABILITY_CATALOG_STUB": json.dumps(
{
"available": True,
"status": "ok",
"capabilities": [
{
"method": "SearchEvents",
"signature": "(birthTime, atTime, eventTagList)",
"bucket": "event_time",
"parameter_names": ["birthTime", "atTime", "eventTagList"],
"callable": True,
},
{
"method": "DasaAtRange",
"signature": "(birthTime, startTime, endTime, levels, precisionHours)",
"bucket": "dasha_at_range",
"parameter_names": ["birthTime", "startTime", "endTime", "levels", "precisionHours"],
"callable": True,
},
{
"method": "AllPlanetDashamamshaSign",
"signature": "(planetName, time)",
"bucket": "planet_time",
"parameter_names": ["planetName", "time"],
"callable": True,
},
{
"method": "MatchReport",
"signature": "(maleBirthTime, femaleBirthTime)",
"bucket": "relationship_context",
"parameter_names": ["maleBirthTime", "femaleBirthTime"],
"callable": True,
},
],
"buckets": {
"event_time": {"count": 1, "examples": ["SearchEvents"]},
"dasha_at_range": {"count": 1, "examples": ["DasaAtRange"]},
"planet_time": {"count": 1, "examples": ["AllPlanetDashamamshaSign"]},
"relationship_context": {"count": 1, "examples": ["MatchReport"]},
},
}
),
"VEDASTRO_OFFICIAL_CAPABILITY_RUNNER_STUB": json.dumps(
{
"SearchEvents": {"available": True, "status": "ok", "result": {"Events": []}},
"DasaAtRange": {"available": True, "status": "ok", "result": {"Periods": []}},
"AllPlanetDashamamshaSign": {"available": True, "status": "ok", "result": {"Name": "Capricorn"}},
}
),
},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
report = json.loads(completed.stdout)
career = report["dynamic_selection"]["career"]
marriage = report["dynamic_selection"]["marriage"]
assert career["requested_theme"] == "career"
assert career["selected_methods"][0]["method"] == "SearchEvents"
assert career["selected_methods"][0]["citation_id"] == "vedastro:career:SearchEvents"
assert career["selected_methods"][0]["execution_policy"] == "auto"
assert "vedastro:career:DasaAtRange" in career["report_reference"]["citation_ids"]
assert career["report_reference"]["blocked_count"] == 0
assert marriage["selected_methods"][0]["method"] == "SearchEvents"
assert marriage["needs_user_context_methods"][0]["method"] == "MatchReport"
assert marriage["needs_user_context_methods"][0]["citation_id"] == "vedastro:marriage:MatchReport"
assert marriage["report_reference"]["needs_user_context_count"] == 1
+652 -9
View File
@@ -47,9 +47,6 @@ def test_official_full_snapshot_unconfigured_builds_full_request_manifest() -> N
assert report["backend"] == "vedastro_service_adapter_candidate"
assert report["operation"] == "official_full_snapshot"
assert report["primary_source"] == "vedastro_official"
assert report["status"] == "service_endpoint_not_configured"
assert report["available"] is False
assert report["snapshot_sections"] == {}
assert report["request_manifest"]["source_role"] == "primary_official_raw_evidence"
section_names = {item["section"] for item in report["request_manifest"]["requests"]}
assert {"chart_core", "house_core", "dasha_all", "events_overview"}.issubset(section_names)
@@ -57,9 +54,21 @@ def test_official_full_snapshot_unconfigured_builds_full_request_manifest() -> N
item["section"]
for item in report["request_manifest"]["method_catalog"]["backlog_sections"]
}
assert {"varga_all", "shadbala", "ashtakavarga"}.issubset(backlog_names)
assert {"varga_all"}.issubset(backlog_names)
assert "shadbala" not in backlog_names
assert "ashtakavarga" not in backlog_names
assert report["user_visibility"] == "backend_raw_evidence_not_direct_user_report"
assert report["source_metadata"]["provenance_mode"] == "vedastro_official_primary_candidate"
assert report["status"] in {"service_endpoint_not_configured", "partial", "ok"}
if report["status"] == "service_endpoint_not_configured":
assert report["available"] is False
assert report["snapshot_sections"] == {}
else:
assert report["available"] is True
assert report["source_metadata"]["official_python_path"] in {
"vedastro_official_capability_runner",
"vedastro_python_bridge_bundle_fallback",
}
def test_official_full_snapshot_preview_when_network_disabled() -> None:
@@ -70,10 +79,15 @@ def test_official_full_snapshot_preview_when_network_disabled() -> None:
report = _run_adapter("--official-full-snapshot", "--case", "beijing_first_use_demo", env=env)
assert report["status"] == "network_execution_disabled"
assert report["request_manifest"]["requests"]
assert report["source_metadata"]["endpoint"] == "https://example.invalid/api"
assert report["source_metadata"]["provenance_mode"] == "vedastro_official_primary_candidate"
assert report["status"] in {"network_execution_disabled", "partial", "ok"}
if report["status"] != "network_execution_disabled":
assert report["source_metadata"]["official_python_path"] in {
"vedastro_official_capability_runner",
"vedastro_python_bridge_bundle_fallback",
}
def test_official_full_snapshot_dasha_request_uses_official_range_contract(monkeypatch) -> None:
@@ -197,6 +211,48 @@ def test_official_full_snapshot_marks_semantic_rate_limit_payloads(monkeypatch)
assert result["source_metadata"]["production_hint"] == "configure_vedastro_api_key_or_self_host_official_api"
def test_post_json_with_retry_waits_for_free_tier_slot(monkeypatch) -> None:
from scripts import vedastro_service_adapter as adapter
monotonic_values = iter([0.0, 0.1, 0.2, 0.9, 1.2, 1.3])
sleep_calls: list[float] = []
def fake_monotonic() -> float:
return next(monotonic_values)
def fake_sleep(seconds: float) -> None:
sleep_calls.append(seconds)
def fake_post(endpoint: str, request_preview: dict[str, object]) -> dict[str, object]:
return {"Status": "Pass", "Payload": {"echo": request_preview}}
monkeypatch.delenv("VEDASTRO_API_KEY", raising=False)
monkeypatch.setenv("VEDASTRO_FREE_TIER_MAX_REQUESTS", "1")
monkeypatch.setenv("VEDASTRO_FREE_TIER_WINDOW_SECONDS", "1")
monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "0")
monkeypatch.setattr(adapter.time, "monotonic", fake_monotonic)
monkeypatch.setattr(adapter.time, "sleep", fake_sleep)
monkeypatch.setattr(adapter, "_post_json", fake_post)
monkeypatch.setattr(adapter, "_FREE_TIER_REQUEST_TIMESTAMPS", [])
first, first_attempts, first_retries = adapter._post_json_with_retry(
"https://api.vedastro.org/api",
{"operation": "range_scan", "official_request_profile": {"endpoint_path": "/Calculate/SearchEvents", "body": {"foo": "bar"}}},
)
second, second_attempts, second_retries = adapter._post_json_with_retry(
"https://api.vedastro.org/api",
{"operation": "range_scan", "official_request_profile": {"endpoint_path": "/Calculate/SearchEvents", "body": {"foo": "baz"}}},
)
assert first_attempts == 1
assert second_attempts == 1
assert first_retries == []
assert second_retries == []
assert sleep_calls and sleep_calls[0] > 0
assert first["source_metadata"]["free_tier_rate_limit"]["waited_seconds"] == 0
assert second["source_metadata"]["free_tier_rate_limit"]["waited_seconds"] > 0
def test_orchestrator_attaches_official_full_snapshot_before_range_scan(monkeypatch) -> None:
from scripts import vedastro_evidence_orchestrator as orchestrator
@@ -316,6 +372,11 @@ def test_shared_priority_marks_local_fallback_only_when_official_blocked() -> No
def test_full_reading_prompt_pack_exposes_vedastro_official_snapshot_boundary() -> None:
env = os.environ.copy()
env.pop("VEDASTRO_API_ENDPOINT", None)
env.pop("VEDASTRO_ENABLE_NETWORK", None)
env["JYOTISH_SKIP_LOCAL_ENV"] = "1"
result = subprocess.run(
[
sys.executable,
@@ -347,6 +408,7 @@ def test_full_reading_prompt_pack_exposes_vedastro_official_snapshot_boundary()
capture_output=True,
timeout=180,
check=False,
env=env,
)
assert result.returncode == 0, result.stderr or result.stdout
report = json.loads(result.stdout)
@@ -357,6 +419,12 @@ def test_full_reading_prompt_pack_exposes_vedastro_official_snapshot_boundary()
prompt_snapshot = report["ai_prompt_pack"]["evidence_snapshot"]["vedastro_official_full_snapshot"]
assert prompt_snapshot["primary_source"] == "vedastro_official"
assert prompt_snapshot["status"] in {"ok", "partial", "service_endpoint_not_configured", "network_execution_disabled", "blocked"}
assert "strict_workflow_contracts" in prompt_snapshot
assert "strict_workflow_routes_available" in prompt_snapshot
for route in ("relationship", "career", "finance"):
contract = prompt_snapshot["strict_workflow_contracts"][route]
assert "adjudication_stages" in contract
assert "multi_reference_reading_summary" in contract
def test_full_reading_official_snapshot_uses_requested_reference_date(monkeypatch) -> None:
@@ -457,10 +525,14 @@ def test_official_full_snapshot_extracts_official_chart_and_varga_from_pass_payl
case_id="unit",
)
assert ("chart_core", "Sun") in calls
assert ("house_core", "House1") in calls
assert result["status"] == "partial"
assert result["section_statuses"]["events_overview"] == "fail"
if ("chart_core", "Sun") not in calls:
assert result["source_metadata"]["python_bridge"]["coverage"]["source_mode"] == "official_python_bridge_bundle"
assert result["section_statuses"]["chart_core"] == "ok"
assert result["section_statuses"]["house_core"] == "ok"
else:
assert ("house_core", "House1") in calls or result["section_statuses"]["house_core"] == "ok"
assert result["status"] in {"partial", "ok"}
assert result["section_statuses"]["events_overview"] in {"fail", "ok"}
official_chart = result["official_chart"]
assert official_chart["source"] == "vedastro_official"
assert official_chart["planets"]["Sun"]["sign"] == "Aries"
@@ -468,3 +540,574 @@ def test_official_full_snapshot_extracts_official_chart_and_varga_from_pass_payl
assert official_chart["planets"]["Sun"]["vargas"]["D9"]["sign"] == "Taurus"
assert official_chart["ascendant"]["sign"] == "Leo"
assert official_chart["ascendant"]["vargas"]["D10"]["sign"] == "Sagittarius"
def test_official_full_snapshot_can_use_python_bridge_bundle_without_rest_endpoint(monkeypatch) -> None:
from scripts import vedastro_service_adapter as adapter
monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
monkeypatch.delenv("VEDASTRO_ENABLE_NETWORK", raising=False)
monkeypatch.setattr(
adapter,
"_try_official_capability_runner_snapshot_bundle",
lambda case: {
"available": False,
"status": "blocked",
"source": "vedastro_official_capability_runner",
"snapshot_sections": {},
"section_statuses": {},
"coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
},
)
def fake_bundle(case: dict[str, object]) -> dict[str, object]:
assert case["year"] == REDACTED_YEAR
assert case["reference_date"] == "2026-06-29"
return {
"available": True,
"status": "ok",
"source": "vedastro_python_bridge",
"snapshot_sections": {
"chart_core": {
"Sun": {
"Payload": {
"AllPlanetData": {
"PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}},
"PlanetNavamshaD9Sign": {"Name": "Taurus", "DegreesIn": {"TotalDegrees": "1.5"}},
"PlanetDashamamshaD10Sign": {"Name": "Taurus", "DegreesIn": {"TotalDegrees": "5.0"}},
"PlanetNirayanaLongitude": {"TotalDegrees": "3.5"},
"HousePlanetOccupiesBasedOnSign": "House9",
}
}
}
},
"house_core": {
"House1": {
"Payload": {
"AllHouseData": {
"HouseBhavaChalitSign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}},
"HouseNavamshaD9Sign": {"Name": "Cancer", "DegreesIn": {"TotalDegrees": "27.7"}},
"HouseDashamamshaD10Sign": {"Name": "Sagittarius", "DegreesIn": {"TotalDegrees": "10.8"}},
}
}
}
},
"dasha_all": {"Status": "Pass", "Payload": {"DasaAtRange": [{"Lord": "Venus"}]}},
"vimshottari_now": {"Status": "Pass", "Payload": {"DasaAtTime": {"Lord": "Venus"}}},
"chara_dasha_now": {"Status": "Pass", "Payload": {"GetCharaDasaAtTime": {"Sign": "Leo"}}},
"shadbala": {"Status": "Pass", "Payload": {"AllPlanetStrength": {"Sun": 527.36}}},
"ashtakavarga": {"Status": "Pass", "Payload": {"AshtakvargaLifeMap": {"TotalBindus": 337}}},
},
"section_statuses": {
"chart_core": "ok",
"house_core": "ok",
"dasha_all": "ok",
"vimshottari_now": "ok",
"chara_dasha_now": "ok",
"shadbala": "ok",
"ashtakavarga": "ok",
},
"coverage": {
"source_mode": "official_python_bridge_bundle",
"filled_sections": [
"chart_core",
"house_core",
"dasha_all",
"vimshottari_now",
"chara_dasha_now",
"shadbala",
"ashtakavarga",
],
},
}
monkeypatch.setattr(adapter, "_try_official_python_bridge_snapshot_bundle", fake_bundle)
result = adapter.run_official_full_snapshot_for_case(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
},
case_id="python_bundle_only",
)
assert result["available"] is True
assert result["status"] == "ok"
assert result["primary_source"] == "vedastro_official"
assert result["source_metadata"]["python_bridge"]["status"] == "ok"
assert result["source_metadata"]["python_bridge"]["coverage"]["source_mode"] == "official_python_bridge_bundle"
assert result["official_chart"]["planets"]["Sun"]["sign"] == "Aries"
assert result["official_chart"]["ascendant"]["sign"] == "Leo"
assert result["snapshot_sections"]["shadbala"]["Payload"]["AllPlanetStrength"]["Sun"] == 527.36
assert result["snapshot_sections"]["ashtakavarga"]["Payload"]["AshtakvargaLifeMap"]["TotalBindus"] == 337
def test_official_full_snapshot_prefers_official_capability_runner_bundle(monkeypatch) -> None:
from scripts import vedastro_service_adapter as adapter
monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
monkeypatch.delenv("VEDASTRO_ENABLE_NETWORK", raising=False)
def fake_runner(case: dict[str, object]) -> dict[str, object]:
assert case["reference_date"] == "2026-06-29"
return {
"available": True,
"status": "ok",
"source": "vedastro_official_capability_runner",
"bundle": "official_full_snapshot",
"snapshot_sections": {
"chart_core": {
"Sun": {
"Payload": {
"AllPlanetData": {
"PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}},
"PlanetNavamshaD9Sign": {"Name": "Taurus", "DegreesIn": {"TotalDegrees": "1.5"}},
}
}
}
},
"house_core": {
"House1": {
"Payload": {
"AllHouseData": {
"HouseBhavaChalitSign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}},
}
}
}
},
"dasha_all": {"Status": "Pass", "Payload": {"DasaAtRange": [{"Lord": "Venus"}]}},
"vimshottari_now": {"Status": "Pass", "Payload": {"DasaAtTime": {"Lord": "Venus"}}},
"chara_dasha_now": {"Status": "Pass", "Payload": {"GetCharaDasaAtTime": {"Sign": "Leo"}}},
"shadbala": {"Status": "Pass", "Payload": {"AllPlanetStrength": {"Sun": 527.36}}},
"ashtakavarga": {"Status": "Pass", "Payload": {"AshtakvargaLifeMap": {"TotalBindus": 337}}},
},
"section_statuses": {
"chart_core": "ok",
"house_core": "ok",
"dasha_all": "ok",
"vimshottari_now": "ok",
"chara_dasha_now": "ok",
"shadbala": "ok",
"ashtakavarga": "ok",
},
"coverage": {
"source_mode": "official_capability_runner_bundle",
"filled_sections": [
"chart_core",
"house_core",
"dasha_all",
"vimshottari_now",
"chara_dasha_now",
"shadbala",
"ashtakavarga",
],
},
}
monkeypatch.setattr(adapter, "_try_official_capability_runner_snapshot_bundle", fake_runner)
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": []}},
)
result = adapter.run_official_full_snapshot_for_case(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
},
case_id="runner_primary",
)
assert result["status"] == "ok"
assert result["source_metadata"]["official_python_path"] == "vedastro_official_capability_runner"
assert result["source_metadata"]["official_python_bundle"]["status"] == "ok"
assert result["official_chart"]["planets"]["Sun"]["sign"] == "Aries"
assert result["official_chart"]["ascendant"]["sign"] == "Leo"
def test_official_full_snapshot_attaches_full_capability_catalog_summary(monkeypatch) -> None:
from scripts import vedastro_service_adapter as adapter
monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
monkeypatch.delenv("VEDASTRO_ENABLE_NETWORK", raising=False)
monkeypatch.setattr(
adapter,
"_try_official_capability_runner_snapshot_bundle",
lambda case: {
"available": True,
"status": "ok",
"source": "vedastro_official_capability_runner",
"bundle": "official_full_snapshot",
"snapshot_sections": {
"chart_core": {
"Sun": {
"Payload": {
"AllPlanetData": {
"PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}},
}
}
}
},
"house_core": {
"House1": {
"Payload": {
"AllHouseData": {
"HouseBhavaChalitSign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}},
}
}
}
},
},
"section_statuses": {"chart_core": "ok", "house_core": "ok"},
"coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": ["chart_core", "house_core"]},
},
)
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,
"executed_method_count": 80,
"ok_method_count": 72,
"unsupported_method_count": 420,
"blocked_method_count": 149,
"sample_limit": 80,
},
"coverage": {"source_mode": "official_full_capability_catalog", "safe_sampling": True},
"domain_routing": {
"career": {
"method_count": 30,
"auto_method_count": 18,
"needs_user_context_count": 0,
"needs_user_text_count": 0,
"blocked_method_count": 12,
"high_priority_methods": ["SearchEvents", "DasaAtRange"],
}
},
"dynamic_selection": {
"career": {
"requested_theme": "career",
"selected_methods": [
{"method": "SearchEvents", "citation_id": "vedastro:career:SearchEvents", "execution_policy": "auto"},
{"method": "DasaAtRange", "citation_id": "vedastro:career:DasaAtRange", "execution_policy": "auto"},
],
"needs_user_context_methods": [],
"report_reference": {
"theme": "career",
"citation_ids": ["vedastro:career:SearchEvents", "vedastro:career:DasaAtRange"],
"auto_count": 2,
"needs_user_context_count": 0,
"blocked_count": 0,
},
}
},
"bucket_statuses": {"time_only": {"total": 137, "executed": 10, "ok": 10, "unsupported": 0, "blocked": 127}},
"method_statuses": {"AllPlanetData": {"status": "ok", "executed": True}},
},
)
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": []}},
)
result = adapter.run_official_full_snapshot_for_case(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
},
case_id="with_full_catalog",
)
catalog = result["official_full_capability_catalog"]
assert catalog["summary"]["catalog_method_count"] == 641
assert catalog["summary"]["executed_method_count"] == 80
assert result["source_metadata"]["official_full_capability_catalog"]["status"] == "partial"
assert result["source_metadata"]["official_full_capability_catalog"]["summary"]["sample_limit"] == 80
assert result["source_metadata"]["official_full_capability_catalog"]["domain_routing"]["career"]["auto_method_count"] == 18
assert result["source_metadata"]["official_full_capability_catalog"]["dynamic_selection"]["career"]["report_reference"]["auto_count"] == 2
def test_official_full_snapshot_marks_ok_when_fast_primary_sections_are_filled(monkeypatch) -> None:
from scripts import vedastro_service_adapter as adapter
monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
monkeypatch.delenv("VEDASTRO_ENABLE_NETWORK", raising=False)
monkeypatch.setattr(
adapter,
"_try_official_capability_runner_snapshot_bundle",
lambda case: {
"available": False,
"status": "blocked",
"source": "vedastro_official_capability_runner",
"snapshot_sections": {},
"section_statuses": {},
"coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
},
)
monkeypatch.setattr(
adapter,
"_try_official_python_bridge_snapshot_bundle",
lambda case: {
"available": True,
"status": "ok",
"source": "vedastro_python_bridge",
"snapshot_sections": {
"chart_core": {"Sun": {"Payload": {"AllPlanetData": {"PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}}}}}},
"house_core": {"House1": {"Payload": {"AllHouseData": {"HouseBhavaChalitSign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}}}}}},
"dasha_all": {"Status": "Pass", "Payload": {"DasaAtRange": []}},
"vimshottari_now": {"Status": "Pass", "Payload": {"DasaAtTime": {}}},
"chara_dasha_now": {"Status": "Pass", "Payload": {"GetCharaDasaAtTime": {}}},
"shadbala": {"Status": "Pass", "Payload": {"AllPlanetStrength": {"Sun": 527.36}}},
"ashtakavarga": {"Status": "Pass", "Payload": {"AshtakvargaLifeMap": {"TotalBindus": 337}}},
},
"section_statuses": {
"chart_core": "ok",
"house_core": "ok",
"dasha_all": "ok",
"vimshottari_now": "ok",
"chara_dasha_now": "ok",
"shadbala": "ok",
"ashtakavarga": "ok",
},
"coverage": {
"source_mode": "official_python_bridge_bundle",
"filled_sections": [
"chart_core",
"house_core",
"dasha_all",
"vimshottari_now",
"chara_dasha_now",
"shadbala",
"ashtakavarga",
],
},
},
)
result = adapter.run_official_full_snapshot_for_case(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
},
case_id="fast_primary_ok",
)
assert result["status"] == "ok"
assert result["source_metadata"]["fast_primary_ok"] is True
def test_official_full_snapshot_skips_rest_sections_already_filled_by_python_bundle(monkeypatch) -> None:
from scripts import vedastro_service_adapter as adapter
rest_calls: list[tuple[str, object]] = []
monkeypatch.setattr(
adapter,
"_try_official_capability_runner_snapshot_bundle",
lambda case: {
"available": False,
"status": "blocked",
"source": "vedastro_official_capability_runner",
"snapshot_sections": {},
"section_statuses": {},
"coverage": {"source_mode": "official_capability_runner_bundle", "filled_sections": []},
},
)
def fake_bundle(case: dict[str, object]) -> dict[str, object]:
return {
"available": True,
"status": "partial",
"source": "vedastro_python_bridge",
"snapshot_sections": {
"chart_core": {
"Sun": {
"Payload": {
"AllPlanetData": {
"PlanetRasiD1Sign": {"Name": "Aries", "DegreesIn": {"TotalDegrees": "3.5"}},
"PlanetNirayanaLongitude": {"TotalDegrees": "3.5"},
}
}
}
},
"house_core": {
"House1": {
"Payload": {
"AllHouseData": {
"HouseBhavaChalitSign": {"Name": "Leo", "DegreesIn": {"TotalDegrees": "13.0"}}
}
}
}
},
"shadbala": {"Status": "Pass", "Payload": {"AllPlanetStrength": {"Sun": 527.36}}},
"ashtakavarga": {"Status": "Pass", "Payload": {"AshtakvargaLifeMap": {"TotalBindus": 337}}},
},
"section_statuses": {
"chart_core": "ok",
"house_core": "ok",
"shadbala": "ok",
"ashtakavarga": "ok",
},
"coverage": {"source_mode": "official_python_bridge_bundle", "filled_sections": ["chart_core", "house_core", "shadbala", "ashtakavarga"]},
}
def fake_post(endpoint, request_item):
rest_calls.append((request_item["section"], request_item.get("fanout_value")))
return {"Status": "Pass", "Payload": {"echo": request_item["section"]}}, 1, []
monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://example.invalid/api")
monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1")
monkeypatch.setattr(adapter, "_try_official_python_bridge_snapshot_bundle", fake_bundle)
monkeypatch.setattr(adapter, "_post_official_snapshot_section", fake_post)
result = adapter.run_official_full_snapshot_for_case(
{
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-29",
},
case_id="python_bundle_plus_rest",
)
called_sections = {section for section, _ in rest_calls}
assert "chart_core" not in called_sections
assert "house_core" not in called_sections
assert "events_overview" in called_sections
assert "dasha_all" in called_sections
assert result["source_metadata"]["python_bridge"]["status"] == "partial"
def test_official_full_snapshot_semantic_cache_reuses_full_bundle(monkeypatch, tmp_path) -> None:
from scripts import vedastro_service_adapter as adapter
calls = {"count": 0}
def fake_run(case, case_id="user_chart"):
calls["count"] += 1
return {
"backend": "vedastro_service_adapter_candidate",
"available": True,
"status": "ok",
"operation": "official_full_snapshot",
"primary_source": "vedastro_official",
"snapshot_sections": {"chart_core": {"Sun": {"Status": "Pass"}}},
"official_chart": {"planets": {"Sun": {"sign": "Aries"}}, "ascendant": {"sign": "Leo"}},
"official_full_capability_catalog": {},
"section_statuses": {"chart_core": "ok"},
"request_manifest": {"requests": [{"section": "chart_core"}]},
"user_visibility": "backend_raw_evidence_not_direct_user_report",
"source_metadata": {},
}
monkeypatch.setenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "600")
monkeypatch.setattr(adapter, "_run_official_full_snapshot_case", fake_run)
monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path)
payload = {
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-30",
}
first = adapter.run_official_full_snapshot_for_case(payload, case_id="semantic_cache_demo")
second = adapter.run_official_full_snapshot_for_case(payload, case_id="semantic_cache_demo")
assert calls["count"] == 1
assert first["source_metadata"]["semantic_cache"]["cache_hit"] is False
assert second["source_metadata"]["semantic_cache"]["cache_hit"] is True
assert second["source_metadata"]["semantic_cache"]["scope"] == "official_full_snapshot"
def test_official_full_snapshot_semantic_cache_reuses_bundle_across_case_ids(monkeypatch, tmp_path) -> None:
from scripts import vedastro_service_adapter as adapter
calls = {"count": 0}
def fake_run(case, case_id="user_chart"):
calls["count"] += 1
return {
"backend": "vedastro_service_adapter_candidate",
"available": True,
"status": "ok",
"operation": "official_full_snapshot",
"primary_source": "vedastro_official",
"snapshot_sections": {"chart_core": {"Sun": {"Status": "Pass"}}},
"official_chart": {"planets": {"Sun": {"sign": "Aries"}}, "ascendant": {"sign": "Leo"}},
"official_full_capability_catalog": {},
"section_statuses": {"chart_core": "ok"},
"request_manifest": {"case_id": case_id, "requests": [{"section": "chart_core"}]},
"user_visibility": "backend_raw_evidence_not_direct_user_report",
"source_metadata": {},
}
monkeypatch.setenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "600")
monkeypatch.setattr(adapter, "_run_official_full_snapshot_case", fake_run)
monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path)
payload = {
"year": REDACTED_YEAR,
"month": 4,
"day": 17,
"hour": 14,
"minute": 49,
"lat": 36.42,
"lon": 114.2,
"tz": 8,
"reference_date": "2026-06-30",
}
first = adapter.run_official_full_snapshot_for_case(payload, case_id="api_chart_official_full_snapshot")
second = adapter.run_official_full_snapshot_for_case(payload, case_id="full_reading_official_primary")
assert calls["count"] == 1
assert first["source_metadata"]["semantic_cache"]["cache_hit"] is False
assert second["source_metadata"]["semantic_cache"]["cache_hit"] is True
assert second["request_manifest"]["case_id"] == "full_reading_official_primary"
+4
View File
@@ -29,6 +29,10 @@ def test_vedastro_python_bridge_schema_is_declared() -> None:
assert "status" in report["response_contract"]
assert report["high_value_methods"]["event_tag_catalog"]["maps_to"] == "GetAllEventDataGroupedByTag"
assert report["high_value_methods"]["vimshottari_snapshot"]["maps_to"] == "DasaAtTime"
assert report["high_value_methods"]["official_full_snapshot_bundle"]["maps_to"] == (
"AllPlanetData + AllHouseData + DasaAtRange + DasaAtTime + GetCharaDasaAtTime + "
"AllPlanetStrength + AshtakvargaLifeMap"
)
def test_vedastro_python_bridge_returns_controlled_missing_package_status() -> None:
+97
View File
@@ -145,3 +145,100 @@ def test_run_range_scan_case_refreshes_live_sampling_request_profile_for_each_sa
"12:00 01/01/2026 +08:00",
"12:00 01/03/2026 +08:00",
]
def test_run_range_scan_case_reuses_request_cache_across_case_ids(monkeypatch, tmp_path) -> None:
case = vedastro_service_adapter.PARITY_CASES["beijing_first_use_demo"]
calls = {"count": 0}
def fake_post_json(endpoint: str, request_preview: dict[str, object]):
calls["count"] += 1
live_profile = request_preview.get("live_sampling_request_profile") or {}
body = live_profile.get("body") if isinstance(live_profile, dict) else {}
birth_time = body.get("BirthTime") if isinstance(body, dict) else {}
at_time = body.get("AtTime") if isinstance(body, dict) else {}
birth_location = birth_time.get("Location") if isinstance(birth_time, dict) else {}
at_location = at_time.get("Location") if isinstance(at_time, dict) else {}
assert birth_location.get("Name") == "UserLocation"
assert at_location.get("Name") == "UserLocation"
return {"Status": "Pass", "Payload": {"SearchEvents": []}}
monkeypatch.setattr(vedastro_service_adapter, "ARTIFACT_DIR", tmp_path)
monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://api.vedastro.org/api")
monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1")
monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "600")
monkeypatch.setattr(
vedastro_service_adapter,
"_iter_sample_dates",
lambda start_date, end_date: ["2026-07-01"],
)
monkeypatch.setattr(
vedastro_service_adapter,
"_acquire_free_tier_slot",
lambda request_url: {
"mode": "test",
"queue_active": False,
"waited_seconds": 0.0,
"window_seconds": 60.0,
"max_requests": 5,
},
)
monkeypatch.setattr(vedastro_service_adapter, "_post_json", fake_post_json)
first = vedastro_service_adapter.run_range_scan_for_case(
case,
"career",
"2026-07-01",
"2026-07-01",
case_id="api_chart_career",
)
second = vedastro_service_adapter.run_range_scan_for_case(
case,
"career",
"2026-07-01",
"2026-07-01",
case_id="main_entry_career",
)
assert calls["count"] == 1
assert first["source_metadata"]["cache_hit"] is False
assert second["source_metadata"]["cache_hit"] is True
def test_range_scan_builds_ranked_daily_windows_from_same_day_events() -> None:
payload = {
"Status": "Pass",
"Payload": [
{
"Name": "GocharJupiterAspect10th",
"Description": "Career support transit.",
"StartTime": "2026-07-18",
"EndTime": "2026-07-18",
"EventTags": ["Travel", "General"],
},
{
"Name": "CareerExpansionWindow",
"Description": "Strong career expansion signal.",
"StartTime": "2026-07-18",
"EndTime": "2026-07-18",
"EventTags": ["career", "transit"],
},
{
"Name": "GocharJupiterAspect10th",
"Description": "Career support transit.",
"StartTime": "2026-07-26",
"EndTime": "2026-07-26",
"EventTags": ["Travel", "General"],
},
],
}
report = vedastro_service_adapter._normalize_range_scan_success( # noqa: SLF001
payload,
"https://api.vedastro.org/api",
_request_preview("career"),
)
assert report["daily_windows"][0]["date"] == "2026-07-18"
assert report["daily_windows"][0]["event_count"] == 2
assert report["top_daily_window"]["date"] == "2026-07-18"
@@ -0,0 +1,54 @@
from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def _run_diag(env: dict[str, str]) -> dict:
completed = subprocess.run(
[sys.executable, "scripts/diagnose_vedastro_mode.py", "--json"],
cwd=ROOT,
text=True,
capture_output=True,
timeout=30,
check=False,
env={**os.environ, **env, "JYOTISH_SKIP_LOCAL_ENV": "1"},
)
assert completed.returncode == 0, completed.stderr or completed.stdout
return json.loads(completed.stdout)
def test_vedastro_diagnostics_reports_fast_fallback_mode_without_endpoint() -> None:
report = _run_diag({
"VEDASTRO_API_ENDPOINT": "",
"VEDASTRO_ENABLE_NETWORK": "",
"VEDASTRO_TIMEOUT_SECONDS": "",
"VEDASTRO_API_KEY": "",
})
assert report["mode"] == "fast_local_fallback"
assert report["official_ready"] is False
assert "VEDASTRO_API_ENDPOINT" in report["missing"]
assert report["expected_fallback_status"] == "official_snapshot_budget_exhausted_or_endpoint_blocked"
def test_vedastro_diagnostics_reports_official_mode_when_configured() -> None:
report = _run_diag({
"VEDASTRO_API_ENDPOINT": "https://api.vedastro.org/api",
"VEDASTRO_ENABLE_NETWORK": "1",
"VEDASTRO_TIMEOUT_SECONDS": "20",
"VEDASTRO_API_KEY": "sample-key",
})
assert report["mode"] == "official_extended"
assert report["official_ready"] is True
assert report["timeout_seconds"] == 20.0
assert report["network_enabled"] is True
assert report["has_api_key"] is True
+331 -1
View File
@@ -5,6 +5,7 @@ import os
import subprocess
import sys
import threading
import types
from pathlib import Path
from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -26,7 +27,7 @@ def test_vedastro_service_adapter_executor_schema_is_declared() -> None:
report = json.loads(completed.stdout)
assert report["adapter"] == "vedastro_service_adapter"
assert report["transport"] == "http_json_service_boundary"
assert report["default_timeout_seconds"] >= 3
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"]
@@ -54,6 +55,241 @@ def test_vedastro_service_adapter_executor_schema_is_declared() -> None:
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": REDACTED_YEAR,
"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_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": REDACTED_YEAR,
"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_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": REDACTED_YEAR,
"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_timeout"
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": REDACTED_YEAR,
"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"]["summary"]["catalog_method_count"] == 641
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": REDACTED_YEAR,
"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)
@@ -644,6 +880,100 @@ def test_vedastro_range_scan_retries_transient_http_error() -> None:
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