from __future__ import annotations import importlib.util import sqlite3 import sys from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from statistics import median from urllib.parse import urlencode ROOT = Path(__file__).resolve().parents[1] HEALTH = ROOT / "scripts" / "health" if str(HEALTH) not in sys.path: sys.path.insert(0, str(HEALTH)) from dashboard_v5 import read_api # noqa: E402 from dashboard_v5.data_provider import build_bundle, connect # noqa: E402 from dashboard_v5.read_api import dispatch_api # noqa: E402 from dashboard_v5.render import render # noqa: E402 _FIXTURE_PATH = ROOT / "tests" / "fixtures" / "dashboard_v5_fixture.py" _loader = importlib.util.spec_from_file_location("dashboard_v5_fixture_for_sprint6c", _FIXTURE_PATH) if _loader is None or _loader.loader is None: raise RuntimeError("Unable to load dashboard_v5_fixture helper") _fixture_module = importlib.util.module_from_spec(_loader) _loader.loader.exec_module(_fixture_module) # type: ignore[attr-defined] build_dashboard_v5_fixture = _fixture_module.build_dashboard_v5_fixture # type: ignore[attr-defined] SYNTHETIC_TODAY = date(2026, 6, 30) ANCHOR = date(2026, 6, 15) def iso_noon(day: date) -> str: return datetime.combine(day, time(12), tzinfo=timezone.utc).isoformat() def api_database(tmp_path: Path) -> Path: database = tmp_path / "sprint6c.db" build_dashboard_v5_fixture(database, anchor_date=ANCHOR) return database def _insert_step_rows(database: Path) -> None: connection = sqlite3.connect(database) try: connection.execute("DELETE FROM apple_health_records WHERE metric='step_count'") for index, value in enumerate([1, 3, 5, 100, 200], start=1): observed_day = date(2026, 6, index) key = f"s6c-step-{index}" connection.execute( """INSERT INTO apple_health_records (record_type,metric,start_date,end_date,value,value_text,unit,source_name, source_version,device,file_name,file_hash,record_hash,raw_json,imported_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( "HKQuantityType", "step_count", iso_noon(observed_day), iso_noon(observed_day), value, None, "count", "Synthetic Baseline Source", "1", "Synthetic Device", f"{key}.json", key, f"{key}-record", "{}", iso_noon(observed_day), ), ) connection.commit() finally: connection.close() def _insert_laboratory_observations(database: Path) -> None: connection = sqlite3.connect(database) try: connection.execute("DELETE FROM laborwerte") first = date(2026, 6, 4) second = date(2026, 6, 5) connection.executemany( """INSERT INTO laborwerte( parameter_name, wert, einheit, reference_min, reference_max, bemerking, abnahme_datum, befund_datum, wert_original, validierungsstatus, source_type, reference_range_source, verified_against_original, provenance_note ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", [ ( "C-Reaktives Protein (CRP)", "0.8", "mg/L", "0", "5", None, first.isoformat(), None, "0.8", "validiert", "scanned_original", "scanned_original", 1, "with observation-specific reference", ), ( "C-Reaktives Protein (CRP)", "0.4", "mg/L", None, None, None, second.isoformat(), None, "0.4", "validiert", "scanned_original", "scanned_original", 1, "missing reference values", ), ], ) connection.commit() finally: connection.close() def _week_bounds(iso_week: str) -> tuple[str, str]: year, week = iso_week.split("-W") start = date.fromisocalendar(int(year), int(week), 1) end = start + timedelta(days=6) return start.isoformat(), end.isoformat() def test_renderer_owns_the_sprint6c_feature_flag_and_assets(tmp_path): database = api_database(tmp_path) connection = connect(database) try: bundle = build_bundle( connection, today=ANCHOR.isoformat(), generated_at="2026-06-15T12:00:00+00:00", ) finally: connection.close() standard = render(bundle) explorer = render(bundle, explorer_6c=True) assert "data-api-explorer-enabled='0'" in standard assert "dashboard-v5-api-explorer.js" not in standard assert "echarts-6.1.0.min.js" in standard assert "data-api-explorer-enabled='1'" in explorer assert "dashboard-v5-api-explorer.js" in explorer assert "echarts-6.1.0.min.js" in explorer assert "sprint6c=1" not in explorer def test_series_responses_use_authoritative_server_today_anchor_for_open_ranges(tmp_path, monkeypatch): """Sprint-6C: ensure series responses expose server authoritative today anchor.""" database = api_database(tmp_path) monkeypatch.setattr(read_api, "local_today", lambda: SYNTHETIC_TODAY) response = dispatch_api(database, "/api/v1/series", urlencode({"metric": "apple.steps", "resolution": "day"})) assert response["today"] == SYNTHETIC_TODAY.isoformat() assert response["coverage"]["to"] == response["today"] assert response["coverage"]["from"] is not None assert response["coverage"]["observed_days"] <= response["coverage"]["expected_days"] def test_series_points_expose_past_only_baseline_on_each_day(tmp_path, monkeypatch): """Sprint-6C: baseline per day should use only past observations.""" database = api_database(tmp_path) _insert_step_rows(database) monkeypatch.setattr(read_api, "local_today", lambda: date(2026, 6, 10)) response = dispatch_api( database, "/api/v1/series", urlencode( { "metric": "apple.steps", "from": "2026-06-01", "to": "2026-06-05", "resolution": "day", } ), ) by_day = {item["date"]: item for item in response["points"]} assert by_day["2026-06-01"]["baseline"] is None assert by_day["2026-06-02"]["baseline"] is None assert by_day["2026-06-03"]["baseline"] is None assert by_day["2026-06-04"]["baseline"] == 3 assert by_day["2026-06-05"]["baseline"] == 4 # sanity check: baseline of 2026-06-04 should equal median of prior values only. assert by_day["2026-06-04"]["baseline"] == median((1, 3, 5)) def test_lab_series_points_emit_reference_or_explicit_missing_reference_state(tmp_path, monkeypatch): """Sprint-6C: lab series points need reference metadata or explicit missing status.""" database = api_database(tmp_path) _insert_laboratory_observations(database) response = dispatch_api( database, "/api/v1/series", urlencode( { "metric": "lab.crp", "from": "2026-06-01", "to": "2026-06-10", "resolution": "day", } ), ) by_day = {item["date"]: item for item in response["points"]} present = by_day["2026-06-04"] assert present["reference"] == { "min": "0", "max": "5", "source": "scanned_original", } missing = by_day["2026-06-05"] assert "reference" not in missing or missing["reference"] is None assert missing["reference_status"] in {"missing", "missing_reference", "not_available"} def test_weekly_series_points_expose_stable_week_anchors(tmp_path, monkeypatch): """Sprint-6C: weekly series must include stable week anchors.""" database = api_database(tmp_path) monkeypatch.setattr(read_api, "local_today", lambda: SYNTHETIC_TODAY) response = dispatch_api( database, "/api/v1/series", urlencode( { "metric": "apple.steps", "from": "2026-05-25", "to": "2026-06-14", "resolution": "week", } ), ) points = response["points"] assert points assert all({"week_start", "week_end", "drilldown_date"} <= set(point) for point in points) for point in points: expected_start, expected_end = _week_bounds(point["week"]) assert point["week_start"] == expected_start assert point["week_end"] == expected_end assert point["drilldown_date"] == expected_start assert point["week_start"] <= point["date"] <= point["week_end"] assert len(point["week"]) == 8