harden consultation reliability and rectification flow

tests/test_api_async_job_contract.py

@@ -1,6 +1,9 @@
 import json
+import time
 from pathlib import Path
 
+import pytest
+
 from scripts import jyotish_api_server as api
 
 
@@ -57,3 +60,34 @@ def test_rectification_page_uses_choice_questionnaire_contract():
     assert "/api/rectification/questionnaire" in source
     assert "/api/rectification/answers" in source
     assert "候选簇排序" in source
+
+
+def test_home_page_keeps_location_confirmation_local():
+    page = Path(api.REPO_ROOT) / "web" / "index.html"
+    source = page.read_text(encoding="utf-8")
+
+    assert "/api/location/resolve" in source
+    assert "第三方地理服务" in source
+
+
+def test_startup_cleanup_removes_only_expired_job_records(monkeypatch, tmp_path):
+    expired = tmp_path / "expired.json"
+    active = tmp_path / "active.json"
+    expired.write_text(json.dumps({"expires_at_unix": time.time() - 1}), encoding="utf-8")
+    active.write_text(json.dumps({"expires_at_unix": time.time() + 60}), encoding="utf-8")
+    monkeypatch.setattr(api, "_async_job_dir", lambda scope: tmp_path)
+
+    result = api.prune_expired_async_jobs()
+
+    assert result["removed"] == 1
+    assert not expired.exists()
+    assert active.exists()
+
+
+def test_rate_limit_is_configurable_and_rejects_over_budget(monkeypatch):
+    api._RATE_LIMIT_BUCKETS.clear()
+    monkeypatch.setenv("JYOTISH_API_RATE_LIMIT_PER_MINUTE", "1")
+
+    api.enforce_rate_limit("test-client", now=0)
+    with pytest.raises(api.RateLimited):
+        api.enforce_rate_limit("test-client", now=1)

tests/test_candidate_time_sensitivity_scan.py

@@ -0,0 +1,21 @@
+from scripts import candidate_time_sensitivity_scan as scanner
+
+
+def test_scanner_reports_real_divisional_transitions(monkeypatch):
+    def fake_engine(command, payload, timeout=20):
+        minute = payload["minute"]
+        if command == "chart":
+            return {"ascendant": {"sign": "Leo", "degree_in_sign": 10 + minute / 100}}
+        ascendant = "Aries" if minute % 2 else "Taurus"
+        return {"divisional_charts": {"D": {"ascendant": ascendant}}}
+
+    monkeypatch.setattr(scanner, "_engine_json", fake_engine)
+    report = scanner.scan_candidate_times(
+        {"year": 2000, "month": 1, "day": 1, "hour": 12, "minute": 1, "lat": 1, "lon": 1, "tz": 0},
+        uncertainty_minutes=1,
+    )
+
+    assert report["candidate_count"] == 3
+    assert report["transitions"]
+    assert report["pending_layers"] == ["UL", "A7", "A10", "KP_cusp"]
+    assert report["rows"][0]["divisional_ascendants"]["D9"] in {"Aries", "Taurus"}

tests/test_external_oracle_raw_import.py

@@ -0,0 +1,35 @@
+from pathlib import Path
+
+import pytest
+
+from scripts.external_oracle_raw_import import build_raw_oracle_import
+
+
+def _metadata(**overrides):
+    value = {
+        "case_id": "public_case",
+        "license_boundary": "external benchmark only",
+        "collection_method": "manual export",
+        "birth_data_policy": "public_case_only",
+    }
+    value.update(overrides)
+    return value
+
+
+def test_raw_import_requires_reviewable_public_case_artifact(tmp_path: Path):
+    artifact = tmp_path / "oracle.json"
+    artifact.write_text('{"raw": true}', encoding="utf-8")
+
+    result = build_raw_oracle_import("VedAstro", artifact, _metadata())
+
+    assert result["status"] == "raw_imported_uncompared"
+    assert result["source_artifact_sha256"]
+    assert result["comparison_ready"] is False
+
+
+def test_raw_import_rejects_non_public_birth_policy(tmp_path: Path):
+    artifact = tmp_path / "oracle.json"
+    artifact.write_text("{}", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="public_case_only"):
+        build_raw_oracle_import("VedAstro", artifact, _metadata(birth_data_policy="private"))