chore(release): harden premium skill packaging

This commit is contained in:
732642856
2026-07-10 22:31:28 +08:00
parent 52266a3b64
commit 3cbc9ca585
30 changed files with 1567 additions and 40 deletions
+60
View File
@@ -2327,6 +2327,66 @@ def test_consultation_workflow_uses_unified_orchestrator_contract(monkeypatch) -
assert result['chart']['special_lagnas']['precision'] == 'sunrise_correct'
def test_consultation_workflow_accepts_western_oracle_payload(monkeypatch) -> None:
handler = _handler()
fake_chart = {
'success': True,
'birth_info': {'date': '1955-02-24', 'time': '19:15', 'tz': 8},
'ascendant': {'lon': 92.0, 'sign': 'Cancer'},
'planets': _sample_planets(),
'chart': {
'ascendant': {'lon': 92.0, 'sign': 'Cancer'},
'planets': _sample_planets(),
},
'modules': {},
'cross_system_signals': [
{
'theme': 'career_relocation',
'claim': 'career_triggered_relocation',
'timing': '2026-07',
'source': 'jyotish_runtime_signal',
}
],
}
monkeypatch.setattr(handler, '_compute_chart', lambda body: fake_chart)
monkeypatch.setattr(handler, '_compute_rectification_gate', lambda body: {'endpoint': 'rectification_gate'})
monkeypatch.setattr(handler, '_compute_thematic_report', lambda body: {'endpoint': 'thematic_report'})
result = handler._compute_consultation_workflow({
'entry_mode': 'direct_chart',
'question': 'career relocation timing',
'year': 1955,
'month': 2,
'day': 24,
'hour': 19,
'minute': 15,
'lat': 37.7749,
'lon': -122.4194,
'tz': 8,
'theme': ['career'],
'western_oracle_payload': {
'source_engine': 'kerykeion_external_json',
'natal': {'ascendant': 'Virgo', 'mc': 'Gemini'},
'timing_techniques': {'transits': [{'date': '2026-07'}]},
'aspects': [
{
'date': '2026-07',
'planet': 'Uranus',
'aspect': 'conjunction',
'target': 'MC',
}
],
},
})
assert result['western_evidence_packet']['source_engine'] == 'kerykeion_external_json'
assert result['runtime_evidence_log']['cross_system_arbitration']['status'] == 'used'
assert result['runtime_evidence_log']['cross_system_arbitration']['shared_signals'][0]['claim'] == 'career_triggered_relocation'
audit = result['runtime_evidence_log']['quality_gate']['technique_audit_table']
assert any(row['technique'] == 'Western Cross-Validation' and row['used'] is True for row in audit)
def test_consultation_workflow_reuses_chart_data_for_thematic_report_without_recursive_full_reading(monkeypatch) -> None:
handler = _handler()
fake_chart = {
+72
View File
@@ -0,0 +1,72 @@
#!/usr/bin/env python3
"""Cross-system Jyotish/Western arbitration contract tests."""
from __future__ import annotations
from scripts.cross_system_arbitrator import build_cross_system_arbitration
def test_cross_system_arbitrator_promotes_shared_timed_signal() -> None:
packet = build_cross_system_arbitration(
route_packet={"question_type": "career", "primary_theme": "career"},
jyotish_evidence={
"status": "complete",
"signals": [
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-08-24..2026-09-28",
"source": "Saturn/Ketu + Rahu 4H/10H axis",
}
],
},
western_evidence={
"status": "complete",
"signals": [
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-08-24..2026-09-28",
"source": "Uranus conjunct MC opposite IC",
}
],
},
)
assert packet["status"] == "used"
assert packet["primary_theme"] == "career"
assert packet["shared_signals"][0]["theme"] == "career_relocation"
assert packet["shared_signals"][0]["claim"] == "career_triggered_relocation"
assert packet["shared_signals"][0]["timing"] == "2026-08-24..2026-09-28"
assert packet["shared_signals"][0]["confidence_effect"] == "raises_confidence"
assert packet["conflicts"] == []
def test_cross_system_arbitrator_blocks_without_western_packet() -> None:
packet = build_cross_system_arbitration(
route_packet={"question_type": "career", "primary_theme": "career"},
jyotish_evidence={"status": "partial", "signals": []},
western_evidence=None,
)
assert packet["status"] == "blocked"
assert packet["western_cross_validation"]["status"] == "blocked"
assert "western_evidence_packet_missing" in packet["blocked_items"]
def test_cross_system_arbitrator_marks_conflicting_claims() -> None:
packet = build_cross_system_arbitration(
route_packet={"question_type": "career", "primary_theme": "career"},
jyotish_evidence={
"status": "complete",
"signals": [{"theme": "career", "claim": "stable_internal_role", "timing": "2026-07"}],
},
western_evidence={
"status": "complete",
"signals": [{"theme": "career", "claim": "external_project_pivot", "timing": "2026-07"}],
},
)
assert packet["status"] == "conflict"
assert packet["conflicts"][0]["theme"] == "career"
assert packet["conflicts"][0]["confidence_effect"] == "lowers_confidence"
@@ -37,4 +37,6 @@ def test_external_engine_adapter_diagnostics_aggregates_three_engines() -> None:
assert "raw_output_path" in contract["expected_oracle_fields"]["PyJHora/JHora"]
assert contract["engine_states"]["jyotishganit"]["available"] is True
assert contract["engine_states"]["PyJHora/JHora"]["tested"] is False
assert contract["replay_manifest"]["tested"] is False
assert contract["replay_manifest"]["blocked_reason"] == "no_same_chart_oracle_rows_imported"
assert report["status"] in {"complete", "partial"}
+68
View File
@@ -186,6 +186,74 @@ def test_mcp_strict_workflow_returns_runtime_evidence_log(monkeypatch) -> None:
assert result["runtime_evidence_log"]["quality_gate"]["technique_audit_table"][1]["technique"] == "VedAstro Raw Archive Manifest"
def test_mcp_strict_workflow_preserves_western_evidence_packet(monkeypatch) -> None:
seen = {}
def fake_execute(**kwargs):
seen.update(kwargs)
return {
"chart": {
"modules": {},
"cross_system_signals": [
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-07",
"source": "jyotish_runtime_signal",
}
],
"ai_prompt_pack": {
"evidence_snapshot": {
"vedastro_official_snapshot": {
"status": "ok",
"official_primary_evidence": {"chart_core": {"status": "ok"}},
}
}
},
},
"routing": {"question_type": "career", "primary_theme": "career"},
"entry_mode": "direct_chart",
"runtime_planner": {"executed_steps": ["compute_chart"], "skipped_steps": []},
"western_evidence_packet": kwargs["western_evidence_packet"],
}
western_packet = {
"system": "western_astrology",
"status": "complete",
"signals": [
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-07",
"source": "western_oracle_signal",
}
],
}
monkeypatch.setattr(mcp_server, "_execute_mcp_consultation_workflow", fake_execute)
monkeypatch.setattr(mcp_server, "_maybe_attach_vedastro_evidence", lambda route, chart, **kwargs: chart)
monkeypatch.setattr(mcp_server, "_collect_strict_evidence", lambda route, chart: {"question_type": route})
result = mcp_server.strict_workflow(
question="career relocation timing",
year=1955,
month=2,
day=24,
hour=19,
minute=15,
lat=37.7749,
lon=-122.4194,
tz=8,
age=33,
transit_date="2026-07-05",
western_evidence_packet=western_packet,
)
assert seen["western_evidence_packet"] == western_packet
assert result["runtime_evidence_log"]["cross_system_arbitration"]["status"] == "used"
assert result["runtime_evidence_log"]["cross_system_arbitration"]["shared_signals"][0]["claim"] == "career_triggered_relocation"
def test_career_blocks_label_when_d10_is_missing_but_preserves_jaimini_context() -> None:
result = _base_career_result()
del result["modules"]["varga_full"]["D10_Dasamsa"]
+16 -6
View File
@@ -2,20 +2,30 @@ from __future__ import annotations
from pathlib import Path
from scripts.public_release_privacy_scan import build_report, scan_text
from scripts.public_release_privacy_scan import build_report, deny_patterns, iter_release_files, scan_text
def test_public_release_privacy_scan_finds_private_birth_tuple() -> None:
def test_public_release_privacy_scan_finds_private_configured_pattern(monkeypatch) -> None:
monkeypatch.setenv("PUBLIC_RELEASE_DENY_PATTERNS", "SECRET_BIRTH_TOKEN")
findings = scan_text(
Path("sample.py"),
'{"year": REDACTED_YEAR, "month": 4, "day": 17, "hour": 14, "minute": 49}',
"safe text SECRET_BIRTH_TOKEN",
deny_patterns(),
)
assert {item["rule_id"] for item in findings} >= {"private_birth_dict_tuple"}
assert {item["rule_id"] for item in findings} == {"private_env_pattern_01"}
def test_public_release_privacy_scan_has_no_release_findings() -> None:
report = build_report()
assert report["status"] == "pass", report["findings"][:20]
assert report["finding_count"] == 0
def test_public_release_privacy_scan_supports_unpacked_zip_without_git(tmp_path: Path) -> None:
(tmp_path / "INSTALL.md").write_text("safe install text\n", encoding="utf-8")
(tmp_path / "scratch").mkdir()
(tmp_path / "scratch" / "ignored.md").write_text("SECRET_BIRTH_TOKEN\n", encoding="utf-8")
assert [path.name for path in iter_release_files(tmp_path)] == ["INSTALL.md"]
report = build_report(tmp_path)
assert report["status"] == "pass", report["findings"]
+73
View File
@@ -0,0 +1,73 @@
from __future__ import annotations
import json
from pathlib import Path
from scripts.real_case_replay_validator import validate_manifest
from scripts.unified_consultation_orchestrator import UnifiedConsultationOrchestrator
ROOT = Path(__file__).resolve().parents[1]
def test_real_case_replay_manifest_blocks_when_no_cases_are_imported() -> None:
result = validate_manifest(ROOT / "references/real_case_calibration/replay_manifest.json")
assert result["status"] == "blocked"
assert result["case_count"] == 0
assert result["replay_ready_count"] == 0
assert result["blocked_reason"] == "no_structured_outcome_replay_cases_imported"
def test_real_case_replay_validator_accepts_one_structured_case(tmp_path: Path) -> None:
manifest = {
"schema_version": "1.0",
"status": "ready",
"case_schema": "references/real_case_calibration/catalog.schema.json",
"cases": [
{
"case_id": "public_case_001",
"source": {
"url": "https://example.com/public-case",
"source_grade": "verified_secondary",
"license_or_quote_boundary": "summary_only",
},
"chart_signature": {"lagna": "Leo", "notable_yogas": ["career_yoga"]},
"event_outcomes": [
{
"event_type": "career_breakthrough",
"event_date": "2000-01",
"outcome": "public_success",
}
],
"similarity": {
"score": 0.72,
"matching_factors": ["D10", "dasha"],
"dissimilar_factors": [],
},
"replay": {
"outcome_replay_status": "replayed",
"do_not_use_for_prediction": False,
},
}
],
}
path = tmp_path / "replay_manifest.json"
path.write_text(json.dumps(manifest), encoding="utf-8")
result = validate_manifest(path)
assert result["status"] == "pass"
assert result["case_count"] == 1
assert result["replay_ready_count"] == 1
def test_orchestrator_exposes_real_case_replay_manifest_status() -> None:
orchestrator = UnifiedConsultationOrchestrator()
route = {"question_type": "career", "primary_theme": "career"}
packet = orchestrator.real_case_calibration_catalog(route_packet=route, machine_evidence_packet={})
replay = packet["outcome_replay_manifest"]
assert replay["status"] == "blocked"
assert replay["case_count"] == 0
assert packet["required_replay_schema"] == "references/real_case_calibration/catalog.schema.json"
+70
View File
@@ -0,0 +1,70 @@
from __future__ import annotations
import json
import subprocess
import sys
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def test_premium_skill_zip_runs_from_clean_directory(tmp_path: Path) -> None:
zip_path = tmp_path / "jyotish-premium.zip"
release = subprocess.run(
[
sys.executable,
"scripts/skill_release_package.py",
"--edition",
"premium_cloud_drive",
"--write-zip",
str(zip_path),
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=120,
check=False,
)
assert release.returncode == 0, release.stderr or release.stdout
extract_dir = tmp_path / "clean"
with zipfile.ZipFile(zip_path) as archive:
archive.extractall(extract_dir)
required = [
"INSTALL.md",
"USER_PROMPTS.md",
"PACKAGE_ACCEPTANCE.json",
"references/real_case_calibration/replay_manifest.json",
"references/oracle/three_engine_parity_replay_manifest.json",
"references/oracle/western_oracle_adapter_contract.md",
]
assert [path for path in required if not (extract_dir / path).exists()] == []
acceptance = json.loads((extract_dir / "PACKAGE_ACCEPTANCE.json").read_text(encoding="utf-8"))
assert acceptance["status"] == "pass"
privacy = subprocess.run(
[sys.executable, "scripts/public_release_privacy_scan.py"],
cwd=extract_dir,
text=True,
capture_output=True,
timeout=120,
check=False,
)
assert privacy.returncode == 0, privacy.stderr or privacy.stdout
user_acceptance = subprocess.run(
[sys.executable, "scripts/user_invocation_acceptance_check.py"],
cwd=extract_dir,
text=True,
capture_output=True,
timeout=240,
check=False,
)
assert user_acceptance.returncode == 0, user_acceptance.stderr or user_acceptance.stdout
report = json.loads(user_acceptance.stdout)
assert report["status"] == "pass"
assert report["checks"]["guided_topics_entrypoint"] is True
+18
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import json
import zipfile
from scripts.skill_release_package import build_package_plan, write_zip
@@ -17,6 +18,13 @@ def test_skill_release_package_dry_run_uses_safe_tracked_files() -> None:
assert ".env.local" not in plan["files"]
assert not any(path.startswith("scratch/") for path in plan["files"])
assert not any("private" in path.lower() for path in plan["files"])
assert "RELEASE_MANIFEST.json" in plan["generated_files"]
assert "PACKAGE_ACCEPTANCE.json" in plan["generated_files"]
assert plan["required_contracts"]["references/real_case_calibration/replay_manifest.json"] is True
assert plan["required_contracts"]["references/oracle/three_engine_parity_replay_manifest.json"] is True
assert "references/real_case_calibration/replay_manifest.json" in plan["files"]
assert "references/oracle/three_engine_parity_replay_manifest.json" in plan["files"]
assert "references/oracle/western_oracle_adapter_contract.md" in plan["files"]
def test_skill_release_package_can_write_zip(tmp_path) -> None:
@@ -31,10 +39,20 @@ def test_skill_release_package_can_write_zip(tmp_path) -> None:
assert "scripts/skill_release_manifest.py" in names
assert "INSTALL.md" in names
assert "USER_PROMPTS.md" in names
assert "RELEASE_MANIFEST.json" in names
assert "PACKAGE_ACCEPTANCE.json" in names
assert "references/real_case_calibration/replay_manifest.json" in names
assert "references/oracle/three_engine_parity_replay_manifest.json" in names
assert "references/oracle/western_oracle_adapter_contract.md" in names
assert ".env.local" not in names
with zipfile.ZipFile(target) as archive:
install = archive.read("INSTALL.md").decode("utf-8")
prompts = archive.read("USER_PROMPTS.md").decode("utf-8")
acceptance = json.loads(archive.read("PACKAGE_ACCEPTANCE.json").decode("utf-8"))
assert "python3 scripts/user_invocation_acceptance_check.py" in install
assert "https://github.com/732642856/yinduzhanxing" in install
assert "guided_topics" in prompts
assert "official_blocked" in prompts
assert "western_oracle_payload" in prompts
assert acceptance["status"] in {"pass", "blocked"}
assert acceptance["required_contracts"]["references/oracle/three_engine_parity_replay_manifest.json"] is True
@@ -0,0 +1,48 @@
from __future__ import annotations
import json
from pathlib import Path
from scripts.three_engine_parity_replay_validator import validate_manifest
ROOT = Path(__file__).resolve().parents[1]
def test_three_engine_parity_manifest_blocks_without_oracle_rows() -> None:
result = validate_manifest(ROOT / "references/oracle/three_engine_parity_replay_manifest.json")
assert result["status"] == "blocked"
assert result["comparison_row_count"] == 0
assert result["tested"] is False
assert result["blocked_reason"] == "no_same_chart_oracle_rows_imported"
def test_three_engine_parity_validator_accepts_one_same_chart_row(tmp_path: Path) -> None:
manifest = {
"case_id": "public_same_chart_001",
"birth_data_policy": "public_case_only",
"status": "tested",
"engines": {
"VedAstro": {"status": "official_verified", "official_raw_response_path": "references/oracle/artifacts/vedastro.json"},
"PyJHora_JHora": {"status": "tested", "raw_output_path": "references/oracle/artifacts/pyjhora.txt"},
"jyotishganit": {"status": "tested", "raw_output_path": "references/oracle/artifacts/jyotishganit.json"},
},
"comparison_rows": [
{
"section": "D1",
"field": "Sun.longitude",
"local_value": 27.1,
"oracle_values": {"VedAstro": 27.1, "PyJHora_JHora": 27.1, "jyotishganit": 27.1},
"status": "match",
}
],
}
path = tmp_path / "three_engine_parity_replay_manifest.json"
path.write_text(json.dumps(manifest), encoding="utf-8")
result = validate_manifest(path)
assert result["status"] == "pass"
assert result["tested"] is True
assert result["comparison_row_count"] == 1
assert result["match_count"] == 1
@@ -3,6 +3,7 @@
from __future__ import annotations
from scripts.western_evidence_packet import build_western_evidence_packet
from scripts.unified_consultation_orchestrator import UnifiedConsultationOrchestrator
@@ -193,6 +194,8 @@ def test_runtime_evidence_log_exposes_blind_packet_case_and_quality_gate_contrac
"VedAstro Cloud State",
"VedAstro Raw Archive Manifest",
"External Engine Cross-Validation",
"Western Cross-Validation",
"Cross-System Arbitration",
"Evidence Packet",
"Blind Technical Mode",
"MEVG / Global Web Evidence",
@@ -212,8 +215,54 @@ def test_runtime_evidence_log_exposes_blind_packet_case_and_quality_gate_contrac
assert engines["jyotishganit"]["adapter_status"] == "available"
assert engines["jyotishganit"]["license"] == "MIT"
assert log["external_engine_cross_validation"]["status"] == "partial"
assert log["cross_system_arbitration"]["status"] == "blocked"
assert "western_evidence_packet_missing" in log["cross_system_arbitration"]["blocked_items"]
assert log["quality_gate"]["blocked_items"]
def test_runtime_evidence_log_uses_cross_system_shared_signal_when_packets_align() -> None:
orchestrator = UnifiedConsultationOrchestrator()
western_packet = build_western_evidence_packet(
route_packet={"question_type": "career", "primary_theme": "career"},
natal={"ascendant": "Virgo", "mc": "Gemini"},
timing_techniques={"transits": [{"aspect": "Uranus conjunct MC"}]},
signals=[
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-08-24..2026-09-28",
"source": "Uranus conjunct MC opposite IC",
}
],
)
log = orchestrator.runtime_evidence_log(
surface="api_web",
entry_mode="direct_chart",
route_packet={"question_type": "career", "primary_theme": "career"},
executed_steps=["compute_chart"],
skipped_steps=[],
machine_evidence_packet={
"status": "complete",
"sections": {"vedastro_official_raw_archive_manifest": {"status": "used"}},
"functional_benefic_malefic": {"status": "used"},
"signals": [
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-08-24..2026-09-28",
"source": "Saturn/Ketu + Rahu 4H/10H axis",
}
],
},
western_evidence_packet=western_packet,
)
assert log["cross_system_arbitration"]["status"] == "used"
assert log["cross_system_arbitration"]["shared_signals"][0]["claim"] == "career_triggered_relocation"
audit = {row["technique"]: row for row in log["quality_gate"]["technique_audit_table"]}
assert audit["Western Cross-Validation"]["used"] is True
assert audit["Cross-System Arbitration"]["used"] is True
packet = orchestrator.machine_evidence_packet(
chart={"chart": {"planets": {"Sun": {}}, "ascendant": {"sign": "Leo"}}},
route_packet={"question_type": "career", "primary_theme": "career"},
@@ -149,4 +149,4 @@ def test_one_command_user_invocation_acceptance_check() -> None:
assert report["checks"]["user_invocation_tests"] is True
assert report["checks"]["guided_topics_entrypoint"] is True
assert report["checks"]["external_adapter_diagnostics"] is True
assert report["external_adapter_status"] in {"pass", "partial"}
assert report["external_adapter_status"] in {"pass", "partial", "complete"}
+45
View File
@@ -0,0 +1,45 @@
#!/usr/bin/env python3
"""Western evidence packet contract tests."""
from __future__ import annotations
from scripts.western_evidence_packet import build_western_evidence_packet
def test_western_evidence_packet_materializes_auditable_sections() -> None:
packet = build_western_evidence_packet(
route_packet={"question_type": "career", "primary_theme": "career"},
natal={"ascendant": "Virgo", "mc": "Gemini"},
timing_techniques={
"transits": [{"aspect": "Uranus conjunct MC"}],
"solar_return": {"annual_focus": "career"},
},
signals=[
{
"theme": "career_relocation",
"claim": "career_triggered_relocation",
"timing": "2026-08-24..2026-09-28",
"source": "Uranus conjunct MC opposite IC",
}
],
)
assert packet["system"] == "western_astrology"
assert packet["status"] == "complete"
assert packet["sections"]["natal"]["status"] == "used"
assert packet["sections"]["timing_techniques"]["status"] == "used"
assert packet["sections"]["signals"]["status"] == "used"
assert packet["signals"][0]["claim"] == "career_triggered_relocation"
def test_western_evidence_packet_marks_missing_timing_as_partial() -> None:
packet = build_western_evidence_packet(
route_packet={"question_type": "career", "primary_theme": "career"},
natal={"ascendant": "Virgo"},
timing_techniques=None,
signals=[],
)
assert packet["status"] == "partial"
assert "timing_techniques" in packet["missing_sections"]
assert "signals" in packet["missing_sections"]
+120
View File
@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""Western external-oracle adapter tests."""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
from scripts.western_oracle_adapter import build_packet_from_oracle_payload
ROOT = Path(__file__).resolve().parents[1]
def test_western_oracle_adapter_maps_external_aspects_to_standard_signals() -> None:
packet = build_packet_from_oracle_payload(
{
"source_engine": "kerykeion_external_json",
"natal": {"ascendant": "Virgo", "mc": "Gemini"},
"timing_techniques": {"solar_return": {"annual_focus": "career"}},
"aspects": [
{
"date": "2026-07-07",
"planet": "Uranus",
"aspect": "conjunction",
"target": "MC",
"orb": 0.2,
},
{
"date": "2026-07-20",
"planet": "Jupiter",
"aspect": "trine",
"target": "Venus",
"orb": 0.4,
},
{
"date": "2027-07-14",
"planet": "Saturn",
"aspect": "conjunction",
"target": "Sun",
"orb": 0.1,
},
],
},
route_packet={"question_type": "career", "primary_theme": "career"},
)
assert packet["status"] == "complete"
assert packet["source_engine"] == "kerykeion_external_json"
claims = {signal["claim"] for signal in packet["signals"]}
assert "career_triggered_relocation" in claims
assert "client_cooperation_opportunity" in claims
assert "career_responsibility_test" in claims
assert packet["adapter_boundary"]["bundles_external_code"] is False
def test_western_oracle_adapter_preserves_explicit_signals() -> None:
packet = build_packet_from_oracle_payload(
{
"source_engine": "manual_astro_com_export",
"natal": {"ascendant": "Virgo"},
"timing_techniques": {"transits": []},
"signals": [
{
"theme": "career",
"claim": "external_project_pivot",
"timing": "2026-07",
"source": "manual review",
}
],
},
route_packet={"question_type": "career", "primary_theme": "career"},
)
assert packet["signals"] == [
{
"theme": "career",
"claim": "external_project_pivot",
"timing": "2026-07",
"source": "manual review",
}
]
def test_western_oracle_adapter_cli_reads_json_file(tmp_path: Path) -> None:
input_path = tmp_path / "western_oracle.json"
input_path.write_text(
json.dumps(
{
"source_engine": "gongshenxing_export",
"natal": {"ascendant": "Virgo", "mc": "Gemini"},
"aspects": [{"date": "2026-07-07", "planet": "Uranus", "aspect": "conj", "target": "MC"}],
}
),
encoding="utf-8",
)
completed = subprocess.run(
[
sys.executable,
"scripts/western_oracle_adapter.py",
"--input",
str(input_path),
"--theme",
"career",
"--question-type",
"career",
],
cwd=ROOT,
text=True,
capture_output=True,
timeout=30,
check=False,
)
assert completed.returncode == 0, completed.stderr
packet = json.loads(completed.stdout)
assert packet["source_engine"] == "gongshenxing_export"
assert packet["signals"][0]["claim"] == "career_triggered_relocation"