"""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)