jesse-ux
85c3030e96
The career checklist now says the direction section only covers strength, resistance, timing, and three events. Warm builds one packet per cache key and skips when its own cap is full. A three-repeat biography run still has one career answer that names a direction from a planet nature, so this is not accepted. Not pushed.
428 lines
15 KiB
Python
428 lines
15 KiB
Python
"""Consult-card facts are copied from the full-data packet."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from consult_card_domain_facts import (
|
|
CONSULTATION_DOMAINS,
|
|
PRECISION_SENTENCE,
|
|
RAHU_NOTE,
|
|
assert_fact_leaves,
|
|
build_domain_facts,
|
|
domain_source_catalog,
|
|
minute_confirmed_from_body,
|
|
)
|
|
from consult_full_data_cache import (
|
|
attach_consult_full_data,
|
|
build_full_data_packet,
|
|
cache_dir,
|
|
cache_key,
|
|
get_or_build,
|
|
identity_from_body,
|
|
prune_cache,
|
|
warm_consult_packet,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DENSITY = ROOT / "frontend" / "tests" / "fixtures" / "report-density-fictional-engine.json"
|
|
PUBLIC = ROOT / "references" / "public_oracle_cases.json"
|
|
TAYLOR = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v4.json"
|
|
|
|
|
|
def _packet() -> dict:
|
|
return json.loads(DENSITY.read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_every_domain_row_is_sourced_or_blocked():
|
|
catalog = domain_source_catalog()
|
|
assert set(catalog) == set(CONSULTATION_DOMAINS)
|
|
for domain, rows in catalog.items():
|
|
assert rows, domain
|
|
for row in rows:
|
|
assert row["status"] in {"sourced", "blocked"}, row["id"]
|
|
citation = row["citation"]
|
|
assert citation
|
|
if row["status"] == "sourced" and "references/" in citation:
|
|
relative = citation.split()[0]
|
|
assert relative.startswith("references/")
|
|
assert (ROOT / relative).is_file(), relative
|
|
if row["status"] == "blocked":
|
|
assert "references/" in citation or citation.startswith("决策")
|
|
|
|
|
|
def test_density_fixture_leaves_match_the_packet():
|
|
packet = _packet()
|
|
for domain in CONSULTATION_DOMAINS:
|
|
facts = build_domain_facts(packet, domain, minute_confirmed=True)
|
|
assert_fact_leaves(packet, facts)
|
|
encoded = json.dumps(facts, ensure_ascii=False)
|
|
assert "tajika_named_yogas" not in encoded
|
|
assert facts["header_note"] == RAHU_NOTE
|
|
assert facts["planets"]["Rahu"]["functional_role"]["value"] is None
|
|
assert facts["planets"]["Ketu"]["functional_role"]["value"] is None
|
|
|
|
|
|
def test_precision_gate_withholds_d16_and_ghati_until_a_minute_is_confirmed():
|
|
packet = _packet()
|
|
education = build_domain_facts(packet, "education", minute_confirmed=False)
|
|
d24 = next(item for item in education["participation"] if item["id"] == "D24")
|
|
assert d24["status"] == "withheld"
|
|
assert d24["value"] == PRECISION_SENTENCE
|
|
career = build_domain_facts(packet, "career", minute_confirmed=False)
|
|
ghati = next(item for item in career["participation"] if item["id"] == "Ghati")
|
|
assert ghati["value"] == PRECISION_SENTENCE
|
|
confirmed = build_domain_facts(packet, "education", minute_confirmed=True)
|
|
d24_confirmed = next(item for item in confirmed["participation"] if item["id"] == "D24")
|
|
assert d24_confirmed["status"] != "withheld"
|
|
assert minute_confirmed_from_body({"hour": 1, "minute": 1, "declared_accuracy": "rectified"}) is True
|
|
assert minute_confirmed_from_body({"hour": 1, "minute": 1, "birth_time_status": "confirmed"}) is True
|
|
assert minute_confirmed_from_body({"hour": 1, "minute": 1, "birth_time_status": "accepted"}) is True
|
|
assert minute_confirmed_from_body({"hour": 1, "minute": 1, "declared_accuracy": "minute"}) is False
|
|
assert minute_confirmed_from_body({"hour": 1, "minute": 1, "birth_time_accuracy": "provisional"}) is False
|
|
assert minute_confirmed_from_body({"hour": 1, "minute": 1}) is False
|
|
assert minute_confirmed_from_body({"declared_accuracy": "rectified"}) is False
|
|
|
|
|
|
def test_cache_miss_then_hit_and_birth_change_misses():
|
|
token = uuid.uuid4().hex
|
|
calls = {"n": 0}
|
|
|
|
def builder(identity):
|
|
calls["n"] += 1
|
|
return {"marker": identity["engine_version"], "minute": identity["minute"]}
|
|
|
|
base = {
|
|
"schema": "consult-full-data-cache-v1",
|
|
"year": 1901,
|
|
"month": 1,
|
|
"day": 1,
|
|
"hour": 0,
|
|
"minute": 1,
|
|
"second": 0,
|
|
"lat": 0,
|
|
"lon": 0,
|
|
"tz": 0,
|
|
"ayanamsa": "raman",
|
|
"node_mode": "mean",
|
|
"engine_version": token,
|
|
}
|
|
first = get_or_build(base, builder)
|
|
second = get_or_build(base, builder)
|
|
changed = dict(base)
|
|
changed["minute"] = 2
|
|
third = get_or_build(changed, builder)
|
|
assert first["hit"] is False
|
|
assert second["hit"] is True
|
|
assert third["hit"] is False
|
|
assert calls["n"] == 2
|
|
assert second["elapsed_s"] < 1.66
|
|
assert second["packet"]["minute"] == 1
|
|
assert third["packet"]["minute"] == 2
|
|
|
|
|
|
def test_attach_skips_unless_asked_and_gaps_without_a_birth_dump():
|
|
original = {"success": True}
|
|
assert attach_consult_full_data(original, {"theme": "career"}) is original
|
|
failed = attach_consult_full_data({"success": True}, {"include_consult_card_facts": True})
|
|
facts = failed["consult_card_facts"]
|
|
assert facts["gaps"] == ["consult_full_data_unavailable"]
|
|
assert facts["error_type"]
|
|
encoded = json.dumps(facts, ensure_ascii=False)
|
|
assert "Traceback" not in encoded
|
|
assert "year" not in facts
|
|
|
|
|
|
def _public_body(case_id: str) -> dict:
|
|
data = json.loads(PUBLIC.read_text(encoding="utf-8"))
|
|
birth = next(case["birth"] for case in data["cases"] if case.get("id") == case_id)
|
|
return {
|
|
"year": birth["year"],
|
|
"month": birth["month"],
|
|
"day": birth["day"],
|
|
"hour": birth["hour"],
|
|
"minute": birth["minute"],
|
|
"second": birth.get("second", 0),
|
|
"lat": birth["lat"],
|
|
"lon": birth["lon"],
|
|
"tz": birth["tz"],
|
|
"ayanamsa": "raman",
|
|
"node_mode": "mean",
|
|
}
|
|
|
|
|
|
def _taylor_body() -> dict:
|
|
data = json.loads(TAYLOR.read_text(encoding="utf-8"))
|
|
birth = next(case["birth"] for case in data["cases"] if case.get("case_id") == "elizabeth_taylor_1932_aa_v4_holdout")
|
|
year, month, day = (int(part) for part in str(birth["date"]).split("-"))
|
|
clock = str(birth["time"]).split(":")
|
|
return {
|
|
"year": year,
|
|
"month": month,
|
|
"day": day,
|
|
"hour": int(clock[0]),
|
|
"minute": int(clock[1]),
|
|
"second": int(clock[2]) if len(clock) > 2 else 0,
|
|
"lat": birth["latitude"],
|
|
"lon": birth["longitude"],
|
|
"tz": birth["timezone_offset"],
|
|
"ayanamsa": "raman",
|
|
"node_mode": "mean",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("label,body", [
|
|
("jobs", None),
|
|
("obama", None),
|
|
("taylor", None),
|
|
])
|
|
def test_public_charts_copy_every_domain(label, body):
|
|
del body
|
|
selected = {
|
|
"jobs": _public_body("steve_jobs_1955_aa"),
|
|
"obama": _public_body("barack_obama_1961_aa"),
|
|
"taylor": _taylor_body(),
|
|
}[label]
|
|
cached = get_or_build(identity_from_body(selected))
|
|
if cached["slow"]:
|
|
pytest.fail(f"{label} miss {cached['elapsed_s']:.3f}s exceeds 20s")
|
|
packet = cached["packet"]
|
|
slowest = 0.0
|
|
for domain in CONSULTATION_DOMAINS:
|
|
started = time.perf_counter()
|
|
facts = build_domain_facts(packet, domain, minute_confirmed=True)
|
|
elapsed = time.perf_counter() - started
|
|
slowest = max(slowest, elapsed)
|
|
assert_fact_leaves(packet, facts)
|
|
assert "tajika_named_yogas" not in json.dumps(facts, ensure_ascii=False)
|
|
assert any(row["id"].startswith("house:") or row["kind"] == "blocked" for row in facts["factors"])
|
|
aspects = next(item for item in facts["participation"] if item["id"] == "parashari_aspects")
|
|
assert aspects["status"] == "value", domain
|
|
assert aspects["citation"].startswith("references/signs-and-houses.md")
|
|
filled = [
|
|
name for name, row in facts["planets"].items()
|
|
if isinstance(row, dict) and (
|
|
(isinstance(row.get("aspected_by"), dict) and row["aspected_by"].get("value"))
|
|
or (isinstance(row.get("aspects_houses"), dict) and row["aspects_houses"].get("value"))
|
|
)
|
|
]
|
|
assert filled, f"{label} {domain}"
|
|
md = facts["timing"]["md"]
|
|
assert md["status"] == "value", domain
|
|
assert md["value"] == packet["worksheets"]["timing_and_predictive_systems"]["dasha"]["current_dasha"]["lord"]
|
|
assert slowest < 1.66, f"{label} domain read {slowest:.3f}s"
|
|
# Timing only. Birth fields stay out of the assertion text.
|
|
assert cached["elapsed_s"] >= 0
|
|
print(f"{label} hit={cached['hit']} packet_s={cached['elapsed_s']:.3f} domain_s={slowest:.3f}")
|
|
|
|
|
|
def _birth():
|
|
return {
|
|
"year": 1901,
|
|
"month": 1,
|
|
"day": 1,
|
|
"hour": 0,
|
|
"minute": 1,
|
|
"second": 0,
|
|
"lat": 0,
|
|
"lon": 0,
|
|
"tz": 0,
|
|
"ayanamsa": "raman",
|
|
"node_mode": "mean",
|
|
}
|
|
|
|
|
|
def test_reference_date_changes_the_cache_key_and_old_files_are_removed():
|
|
early = identity_from_body({**_birth(), "today": "2001-01-01"})
|
|
later = identity_from_body({**_birth(), "transit_date": "2001-06-01"})
|
|
assert early["reference_date"] == "2001-01-01"
|
|
assert later["reference_date"] == "2001-06-01"
|
|
assert cache_key(early) != cache_key(later)
|
|
seen = []
|
|
|
|
def builder(identity):
|
|
seen.append(identity["reference_date"])
|
|
return {"reference_date": identity["reference_date"], "lord": identity["reference_date"]}
|
|
|
|
for identity in (early, later):
|
|
(cache_dir() / f"{cache_key(identity)}.json").unlink(missing_ok=True)
|
|
first = get_or_build(early, builder)
|
|
second = get_or_build(later, builder)
|
|
assert first["hit"] is False and second["hit"] is False
|
|
assert first["packet"]["lord"] != second["packet"]["lord"]
|
|
assert seen == ["2001-01-01", "2001-06-01"]
|
|
stale = cache_dir() / f"stale-{uuid.uuid4().hex}.json"
|
|
stale.write_text("{}", encoding="utf-8")
|
|
old = time.time() - (3 * 24 * 60 * 60)
|
|
os_utime = __import__("os").utime
|
|
os_utime(stale, (old, old))
|
|
assert prune_cache() >= 1
|
|
assert not stale.exists()
|
|
assert (cache_dir() / f"{cache_key(early)}.json").is_file()
|
|
|
|
|
|
def test_packet_builder_uses_one_reference_date_and_copies_house_aspects(monkeypatch):
|
|
seen = {}
|
|
|
|
def fake_reading(args):
|
|
seen["today"] = args.today
|
|
seen["transit_date"] = args.transit_date
|
|
seen["target_year"] = args.target_year
|
|
return {
|
|
"modules": {
|
|
"aspects": {
|
|
"house_aspects": [
|
|
{"planet": "Saturn", "house": 1, "aspects_to": {7: {"aspect_type": "7宫相位"}}},
|
|
],
|
|
},
|
|
},
|
|
}
|
|
|
|
def fake_packet(_reading):
|
|
return {"worksheets": {}, "core_chart": {}}
|
|
|
|
import calculation_profile_contract
|
|
import jyotish_engine
|
|
|
|
monkeypatch.setattr(jyotish_engine, "cmd_full_reading", fake_reading)
|
|
monkeypatch.setattr(jyotish_engine, "build_pl9_style_export_packet", fake_packet)
|
|
monkeypatch.setattr(calculation_profile_contract, "attach_calculation_profile", lambda packet, _args: packet)
|
|
packet = build_full_data_packet({**_birth(), "reference_date": "1999-05-06"})
|
|
assert seen == {"today": "1999-05-06", "transit_date": "1999-05-06", "target_year": 1999}
|
|
assert packet["worksheets"]["graha_drishti"]["house_aspects"][0]["planet"] == "Saturn"
|
|
|
|
|
|
def test_warm_response_does_not_include_the_packet():
|
|
warmed = warm_consult_packet({})
|
|
assert warmed["success"] is False
|
|
assert "packet" not in warmed
|
|
assert warmed["error_type"]
|
|
|
|
|
|
def _fictional_body(today: str, minute: int) -> dict:
|
|
return {
|
|
"year": 1902,
|
|
"month": 3,
|
|
"day": 3,
|
|
"hour": 1,
|
|
"minute": minute,
|
|
"second": 0,
|
|
"lat": 1,
|
|
"lon": 2,
|
|
"tz": 0,
|
|
"ayanamsa": "raman",
|
|
"node_mode": "mean",
|
|
"today": today,
|
|
}
|
|
|
|
|
|
def _drop_cache(identity: dict) -> None:
|
|
path = cache_dir() / f"{cache_key(identity)}.json"
|
|
path.unlink(missing_ok=True)
|
|
path.with_suffix(".json.tmp").unlink(missing_ok=True)
|
|
|
|
|
|
def test_one_warm_at_a_time_and_same_key_user_waits_for_that_build(monkeypatch):
|
|
"""H2: warms share a cap of 1; a question for the key being warmed builds once."""
|
|
calls = {"n": 0}
|
|
calls_lock = threading.Lock()
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def blocking(identity):
|
|
with calls_lock:
|
|
calls["n"] += 1
|
|
entered.set()
|
|
assert release.wait(timeout=5)
|
|
return {"marker": identity["reference_date"]}
|
|
|
|
monkeypatch.setattr("consult_full_data_cache.build_full_data_packet", blocking)
|
|
|
|
bodies = [_fictional_body(f"1904-01-0{index}", 11) for index in (1, 2, 3)]
|
|
identities = [identity_from_body(body) for body in bodies]
|
|
for identity in identities:
|
|
_drop_cache(identity)
|
|
results: list[dict | None] = [None, None, None]
|
|
|
|
def run_warm(index: int) -> None:
|
|
results[index] = warm_consult_packet(bodies[index])
|
|
|
|
threads = [threading.Thread(target=run_warm, args=(index,)) for index in range(3)]
|
|
try:
|
|
for thread in threads:
|
|
thread.start()
|
|
assert entered.wait(timeout=5)
|
|
deadline = time.time() + 3
|
|
while time.time() < deadline and sum(item is not None for item in results) < 2:
|
|
time.sleep(0.02)
|
|
assert calls["n"] == 1
|
|
assert sum(item is not None and item.get("skipped") is True for item in results) == 2
|
|
for item in results:
|
|
if item is not None:
|
|
assert "packet" not in item
|
|
|
|
same = _fictional_body("1904-02-02", 12)
|
|
same_identity = identity_from_body(same)
|
|
_drop_cache(same_identity)
|
|
# The cap is still held by the first warm. A same-key question must join
|
|
# that pattern on its own key: start a warm, then the question.
|
|
release.set()
|
|
for thread in threads:
|
|
thread.join(timeout=5)
|
|
assert calls["n"] == 1
|
|
built = [item for item in results if item and item.get("skipped") is not True]
|
|
assert len(built) == 1
|
|
assert built[0]["success"] is True
|
|
assert built[0]["hit"] is False
|
|
finally:
|
|
release.set()
|
|
for thread in threads:
|
|
thread.join(timeout=5)
|
|
for identity in identities:
|
|
_drop_cache(identity)
|
|
|
|
entered.clear()
|
|
release.clear()
|
|
calls["n"] = 0
|
|
warm_box: dict = {}
|
|
user_box: dict = {}
|
|
same = _fictional_body("1904-02-02", 12)
|
|
same_identity = identity_from_body(same)
|
|
_drop_cache(same_identity)
|
|
|
|
def run_same_warm() -> None:
|
|
warm_box["result"] = warm_consult_packet(same)
|
|
|
|
def run_user() -> None:
|
|
user_box["result"] = get_or_build(same_identity)
|
|
|
|
warm_thread = threading.Thread(target=run_same_warm)
|
|
user_thread = threading.Thread(target=run_user)
|
|
try:
|
|
warm_thread.start()
|
|
assert entered.wait(timeout=5)
|
|
user_thread.start()
|
|
time.sleep(0.3)
|
|
assert calls["n"] == 1
|
|
assert user_thread.is_alive()
|
|
release.set()
|
|
warm_thread.join(timeout=5)
|
|
user_thread.join(timeout=5)
|
|
assert calls["n"] == 1
|
|
assert warm_box["result"]["success"] is True
|
|
assert warm_box["result"].get("skipped") is not True
|
|
assert "packet" not in warm_box["result"]
|
|
assert user_box["result"]["hit"] is True
|
|
assert user_box["result"]["packet"]["marker"] == "1904-02-02"
|
|
finally:
|
|
release.set()
|
|
warm_thread.join(timeout=5)
|
|
user_thread.join(timeout=5)
|
|
_drop_cache(same_identity)
|