jesse-ux
16f9f86843
The cache key now includes the request date, and files older than two days are removed. The model card keeps values only. Parashari aspects are copied from the engine. Saving a chart warms the packet without blocking the response. Biography severe conflicts rose from 1 to 3, so this is not accepted for merge. Not pushed.
305 lines
11 KiB
Python
305 lines
11 KiB
Python
"""Consult-card facts are copied from the full-data packet."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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"]
|