Files
Jyotisha/tests/test_rectification_v5_vedastro_validation.py
T
jesse-ux b85c4a686a
Independent Staging Quality Gate / validate (push) Successful in 13m27s
Independent Staging Quality Gate / publish (push) Failing after 1h0m1s
fix(rectification): anchor candidate windows to civil dates across midnight
Carry explicit local date intervals instead of inferring the day from clock
order. Cluster width, delivery, adoption, and reports keep the actual civil
date; adopted date is stored separately from the reported birth_date.

Algorithm identity is scoring-9 / spec-v5. Scoring weights, confirmation
thresholds, and Skill version are unchanged. Isolated Linux final-3 gates
passed; four pre-existing Python failures remain. This is not a production
release.
2026-09-21 02:55:00 +08:00

400 lines
16 KiB
Python

from __future__ import annotations
import pytest
from scripts.jyotish_api_server import (
API_COMMAND_MAP,
TECHNIQUE_EXAMPLE_ENDPOINTS,
BadRequest,
JyotishAPIHandler,
_attach_vedastro_minute_sensitive_to_score,
)
def request(candidate_times=None):
events = [
{
"id": "00000000-0000-4000-8000-000000000001",
"domain": "education",
"event_kind": "education_milestone",
"date_start": "2016-09-01",
"date_end": "2016-09-30",
"precision": "month",
"summary": "private education summary that must not be returned",
},
{
"id": "00000000-0000-4000-8000-000000000002",
"domain": "career",
"event_kind": "career_change",
"date_start": "2020-05-01",
"date_end": "2020-05-01",
"precision": "day",
"summary": "private career summary that must not be returned",
},
{
"id": "00000000-0000-4000-8000-000000000003",
"domain": "career",
"event_kind": "career_change",
"date_start": "2021-06-01",
"date_end": "2021-06-01",
"precision": "day",
"summary": "newer career event wins the bounded selection",
},
{
"id": "00000000-0000-4000-8000-000000000004",
"domain": "relationship",
"event_kind": "relationship_change",
"date_start": "2022-01-01",
"date_end": "2022-12-31",
"precision": "year",
"summary": "private relationship summary that must not be returned",
},
{
"id": "00000000-0000-4000-8000-000000000005",
"domain": "finance",
"event_kind": "finance_change",
"date_start": "2023-03-01",
"date_end": "2023-03-31",
"precision": "month",
"summary": "private finance summary that must not be returned",
},
{
"id": "00000000-0000-4000-8000-000000000006",
"domain": "health_pressure",
"event_kind": "self_health_event",
"date_start": "2024-04-01",
"date_end": "2024-04-30",
"precision": "month",
"summary": "private health summary that must not be returned",
},
]
return {
"birth_date": "1997-08-08",
"start_time": "05:00",
"end_time": "05:30",
"lat": 36.419,
"lon": 114.213,
"tz": 8,
"events": events,
"candidate_times": candidate_times if candidate_times is not None else ["05:13", "05:14"],
}
def minute_snapshot(case, *, same=False):
minute = case["minute"]
suffix = "same" if same else str(minute)
return {
"available": True,
"status": "ok",
"source": "vedastro_official",
"layers": {
"ascendant_house_boundaries": {
"status": "ok",
"fingerprint": f"asc-{suffix}",
"ascendant": {"sign": "Leo", "degree_in_sign": minute / 10},
"houses": {"House1": {}},
},
"D9": {"status": "ok", "fingerprint": f"d9-{suffix}", "houses": {"House1": {}}, "planets": {}},
"D10": {"status": "ok", "fingerprint": f"d10-{suffix}", "houses": {"House1": {}}, "planets": {}},
"dasha_boundaries": {"status": "ok", "fingerprint": f"dasha-{suffix}", "boundary_count": 3},
"kp_cusp_sub_lord": {"status": "unsupported", "reason": "not available"},
},
"raw_request": {"api_key": "secret"},
"raw_response": {"private": True},
}
def test_registry_exposes_independent_vedastro_validation_endpoint():
endpoint = "/api/rectification/v5/vedastro-validate"
assert API_COMMAND_MAP["rectification-v5-vedastro-validate"] == endpoint
assert endpoint in TECHNIQUE_EXAMPLE_ENDPOINTS
def test_requires_exactly_two_distinct_candidate_times_before_running_vedastro():
handler = object.__new__(JyotishAPIHandler)
for candidate_times in ([], ["05:13"], ["05:13", "05:14", "05:15"], ["05:13", "05:13"]):
with pytest.raises(BadRequest, match="candidate_times"):
handler._compute_rectification_v5_vedastro_validate(request(candidate_times))
@pytest.mark.parametrize("dated", [False, True])
def test_passes_only_when_official_layers_discriminate_and_primary_is_strictly_better(monkeypatch, dated):
range_calls = []
minute_dates = []
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
minute_dates.append((case["year"], case["month"], case["day"]))
return minute_snapshot(case)
@staticmethod
def run_range_scan_for_case(case, domain, start, end, case_id="user_chart"):
range_calls.append((case["minute"], domain, start, end))
lift = 3 if case["minute"] == 13 else 1
return {
"available": True,
"status": "ok",
"event_count": lift,
"top_event": {"event_id": f"event-{domain}"},
"evidence_ledger": [{"signal_lift": lift}],
"raw_response": {"must_not": "leak"},
}
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
monkeypatch.setenv("VEDASTRO_API_KEY", "must-not-leak")
body = request()
if dated:
body["birth_date"] = "2000-03-01"
body["candidate_intervals"] = [{"start_at": "2000-02-29T05:13", "end_at": "2000-02-29T05:14"}]
result = object.__new__(JyotishAPIHandler)._compute_rectification_v5_vedastro_validate(body)
if dated:
assert minute_dates == [(2000, 2, 29), (2000, 2, 29)]
assert result["status"] == "pass"
assert result["passed"] is True
assert result["can_confirm_exact_minute"] is False
assert result["candidate_times"] == {"primary": "05:13", "runner_up": "05:14"}
assert result["minute_sensitive_validation"]["discriminated"] is True
assert result["event_validation"]["search_events_primary_supports_local_winner"] is True
assert result["event_validation"]["supported_event_count"] == 3
assert len(range_calls) == 6
assert {call[1] for call in range_calls} == {"career", "marriage", "wealth"}
serialized = str(result)
assert "private" not in serialized
assert "raw_request" not in serialized
assert "raw_response" not in serialized
assert "must-not-leak" not in serialized
@pytest.mark.parametrize("mutation", ["missing_source", "missing_layer"])
def test_missing_official_minute_response_cannot_pass(monkeypatch, mutation):
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
report = minute_snapshot(case)
if mutation == "missing_source":
report.pop("source")
else:
report["layers"].pop("D10")
return report
@staticmethod
def run_range_scan_for_case(case, domain, start, end, case_id="user_chart"):
lift = 3 if case["minute"] == 13 else 1
return {
"available": True,
"status": "ok",
"event_count": lift,
"top_event": {"event_id": f"event-{domain}"},
"evidence_ledger": [{"signal_lift": lift}],
}
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = object.__new__(JyotishAPIHandler)._compute_rectification_v5_vedastro_validate(request())
assert result["status"] == "fail"
assert result["passed"] is False
assert result["can_confirm_exact_minute"] is False
assert "vedastro_official_response_missing" in result["blockers"]
def test_identical_minute_layers_cannot_pass(monkeypatch):
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
return minute_snapshot(case, same=True)
@staticmethod
def run_range_scan_for_case(case, domain, start, end, case_id="user_chart"):
return {
"available": True,
"status": "ok",
"event_count": 1,
"top_event": {"event_id": f"event-{domain}"},
"evidence_ledger": [{"signal_lift": 1}],
}
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = object.__new__(JyotishAPIHandler)._compute_rectification_v5_vedastro_validate(request())
assert result["status"] == "fail"
assert result["passed"] is False
assert result["can_confirm_exact_minute"] is False
assert "vedastro_minute_sensitive_layers_not_discriminated" in result["blockers"]
def test_search_events_disagreement_is_diagnostic_and_cannot_reverse_local_winner(monkeypatch):
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
return minute_snapshot(case)
@staticmethod
def run_range_scan_for_case(case, domain, start, end, case_id="user_chart"):
lift = 1 if case["minute"] == 13 else 3
return {
"available": True,
"status": "ok",
"event_count": lift,
"top_event": {"event_id": f"event-{domain}"},
"evidence_ledger": [{"signal_lift": lift}],
}
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = object.__new__(JyotishAPIHandler)._compute_rectification_v5_vedastro_validate(request())
assert result["status"] == "pass"
assert result["passed"] is True
assert result["event_validation"]["search_events_primary_supports_local_winner"] is False
assert result["blockers"] == []
assert result["can_confirm_exact_minute"] is False
def test_timeout_is_returned_as_safe_failure(monkeypatch):
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
raise TimeoutError("secret upstream URL and key")
@staticmethod
def run_range_scan_for_case(case, domain, start, end, case_id="user_chart"):
raise RuntimeError("secret raw response")
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = object.__new__(JyotishAPIHandler)._compute_rectification_v5_vedastro_validate(request())
assert result["status"] == "fail"
assert result["passed"] is False
assert result["can_confirm_exact_minute"] is False
assert "vedastro_timeout" in result["blockers"]
assert "vedastro_exception" in result["blockers"]
assert "secret" not in str(result)
def _score_request():
return {
"birth_date": "1997-08-08",
"lat": 36.419,
"lon": 114.213,
"tz": 8,
}
def _scored_result(*, acceptance=True):
receipt = {
"acceptance_allowed": acceptance,
"confirmation_allowed": False,
"gates": {"exact_confirmation": {"external_validation_status": "not_evaluated"}},
"technique_audit_table": [
{"technique": "VedAstro 分钟级校验", "status": "blocked", "note": "官方分钟级校验尚未评估。"},
{"technique": "唯一分钟确认", "status": "blocked", "note": "本会话以代表性时间收口,不确认唯一分钟。"},
],
}
return {
"candidate_decisions": [
{"time": "05:13", "rank": 1},
{"time": "05:14", "rank": 2},
],
"decision_receipt": receipt,
"candidate_decision_receipt": receipt,
"confirmation_allowed": False,
}
def test_v5_score_attaches_minute_sensitive_pass_without_search_events(monkeypatch):
range_calls = []
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
return minute_snapshot(case)
@staticmethod
def run_range_scan_for_case(case, domain, start, end, case_id="user_chart"):
range_calls.append((domain, start, end))
return {"available": True, "status": "ok"}
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = _attach_vedastro_minute_sensitive_to_score(_scored_result(), _score_request())
exact = result["decision_receipt"]["gates"]["exact_confirmation"]
assert exact["external_validation_status"] == "passed"
assert exact["vedastro_minute_sensitive"]["discriminated"] is True
assert result["confirmation_allowed"] is False
audit = {row["technique"]: row for row in result["decision_receipt"]["technique_audit_table"]}
assert audit["VedAstro 分钟级校验"]["status"] == "executed"
assert "已通过" in audit["VedAstro 分钟级校验"]["note"]
assert audit["唯一分钟确认"]["status"] == "blocked"
assert range_calls == []
serialized = str(result)
assert "36.419" not in serialized
assert "114.213" not in serialized
assert "asc-" not in serialized
def test_v5_score_marks_identical_minute_snapshots_failed_not_not_evaluated(monkeypatch):
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
return minute_snapshot(case, same=True)
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = _attach_vedastro_minute_sensitive_to_score(_scored_result(), _score_request())
exact = result["decision_receipt"]["gates"]["exact_confirmation"]
assert exact["external_validation_status"] == "failed"
assert exact["vedastro_minute_sensitive"]["reason"] == "vedastro_minute_sensitive_layers_not_discriminated"
assert result["confirmation_allowed"] is False
def test_v5_score_timeout_stays_not_evaluated(monkeypatch):
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
raise TimeoutError("secret upstream URL")
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = _attach_vedastro_minute_sensitive_to_score(_scored_result(), _score_request())
exact = result["decision_receipt"]["gates"]["exact_confirmation"]
assert exact["external_validation_status"] == "not_evaluated"
assert exact["vedastro_minute_sensitive"]["reason"] == "vedastro_minute_snapshot_timeout"
assert "secret" not in str(result)
assert result["confirmation_allowed"] is False
def test_v5_score_skips_vedastro_until_local_acceptance(monkeypatch):
calls = []
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
calls.append(True)
return minute_snapshot(case)
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = _attach_vedastro_minute_sensitive_to_score(
_scored_result(acceptance=False),
_score_request(),
)
exact = result["decision_receipt"]["gates"]["exact_confirmation"]
assert exact["external_validation_status"] == "not_evaluated"
assert calls == []
def test_v5_score_join_timeout_stays_not_evaluated(monkeypatch):
import time
class Adapter:
@staticmethod
def run_rectification_minute_snapshot_for_case(case, case_id="user_chart"):
time.sleep(0.4)
return minute_snapshot(case)
monkeypatch.setattr("scripts.jyotish_api_server._foreground_vedastro_budget_seconds", lambda: 0.05)
monkeypatch.setattr("scripts.jyotish_api_server._load_local_module", lambda name: Adapter)
result = _attach_vedastro_minute_sensitive_to_score(_scored_result(), _score_request())
exact = result["decision_receipt"]["gates"]["exact_confirmation"]
assert exact["external_validation_status"] == "not_evaluated"
assert exact["vedastro_minute_sensitive"]["reason"] == "vedastro_minute_snapshot_timeout"
assert result["confirmation_allowed"] is False