"""Sprint-6B.1 integration contracts: browser sessions, weekly rules and inventory reports.""" from __future__ import annotations import base64 import http.cookiejar import json import re import sys import threading import urllib.error import urllib.request from datetime import timedelta from pathlib import Path from urllib.parse import urlencode import pytest 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.apple_identifier_report import build_apple_identifier_report # noqa: E402 from dashboard_v5.read_api import dispatch_api # noqa: E402 from tests.test_dashboard_v5_sprint6b_api import ANCHOR, api_database, load_server # noqa: E402 from tests.test_dashboard_v5_sprint6b_catalog import inventory_db # noqa: E402 def _start_server(module): server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server, f"http://127.0.0.1:{server.server_port}" def test_v5_browser_session_is_httponly_same_site_and_calls_api_without_bearer(tmp_path, monkeypatch): database = api_database(tmp_path) dashboard = tmp_path / "dashboard.html" dashboard.write_text( "synthetic v5", encoding="utf-8", ) api_token_file = tmp_path / "api-token" api_token_file.write_text("synthetic_api_token_0123456789abcdef", encoding="ascii") api_token_file.chmod(0o600) action_inbox = tmp_path / "inbox" action_inbox.mkdir(mode=0o700) monkeypatch.setenv("HEALTH_DASHBOARD_ACTION_INBOX", str(action_inbox)) module = load_server("health_dashboard_server_sprint6b1_session", dashboard, database, api_token_file, monkeypatch) server, base = _start_server(module) try: jar = http.cookiejar.CookieJar() browser = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) v4 = browser.open(base + "/health-dashboard", timeout=5) assert "connect-src 'none'" in v4.headers["Content-Security-Policy"] with pytest.raises(urllib.error.HTTPError) as unauthorized_page: browser.open(base + "/health-dashboard-v5", timeout=5) assert unauthorized_page.value.code == 401 assert unauthorized_page.value.headers["WWW-Authenticate"] == 'Basic realm="Health Dashboard V5"' basic = base64.b64encode(b"health:synthetic_api_token_0123456789abcdef").decode("ascii") browser.addheaders = [("Authorization", f"Basic {basic}")] v5 = browser.open(base + "/health-dashboard-v5", timeout=5) assert "connect-src 'self'" in v5.headers["Content-Security-Policy"] html = v5.read().decode("utf-8") assert "Bearer" not in html and "Basic" not in html match = re.search(r"name='health-browser-session-csrf' content='([^']+)'", html) assert match is not None csrf_token = match.group(1) assert csrf_token not in {"", "__BROWSER_SESSION_CSRF__"} action_match = re.search(r"name='csrf_token' value='([^']+)'", html) assert action_match is not None action_csrf_token = action_match.group(1) assert action_csrf_token not in {"", "__CSRF_TOKEN__"} for action_route in ( "/health-actions/symptom-checkin", "/health-actions/nutrition-mapping", "/health-actions/symptom-event", "/health-actions/medication-event", "/health-actions/general-event", "/health-actions/supplement-event", "/health-actions/observation", "/health-actions/document-review", "/health-actions/capture", ): unauthenticated_action = urllib.request.Request( base + action_route, data=urlencode({"csrf_token": action_csrf_token}).encode("ascii"), method="POST", headers={ "Content-Type": "application/x-www-form-urlencoded", "Origin": base, "Sec-Fetch-Site": "same-origin", "Accept": "application/json", }, ) with pytest.raises(urllib.error.HTTPError) as rejected_action: browser.open(unauthenticated_action, timeout=5) assert rejected_action.value.code == 401, action_route assert list(action_inbox.iterdir()) == [] with pytest.raises(urllib.error.HTTPError) as no_session: browser.open(base + "/api/v1/metric-catalog?q=HRV", timeout=5) assert no_session.value.code == 401 missing_csrf = urllib.request.Request( base + "/api/v1/browser-session", data=b"", method="POST", headers={"Origin": base, "Sec-Fetch-Site": "same-origin"}, ) with pytest.raises(urllib.error.HTTPError) as missing: browser.open(missing_csrf, timeout=5) assert missing.value.code == 403 bootstrap = urllib.request.Request( base + "/api/v1/browser-session", data=b"", method="POST", headers={ "Origin": base, "Sec-Fetch-Site": "same-origin", "X-Health-Browser-CSRF": csrf_token, }, ) session_response = browser.open(bootstrap, timeout=5) assert session_response.status == 204 cookie = session_response.headers["Set-Cookie"] # The browser session also authorizes same-origin V5 action routes outside /api/v1. assert "HttpOnly" in cookie and "SameSite=Strict" in cookie and "Path=/;" in cookie assert "Path=/api/v1" not in cookie assert "synthetic_api_token" not in cookie and "Bearer" not in cookie and "Basic" not in cookie api = urllib.request.Request( base + "/api/v1/metric-catalog?q=HRV", headers={"Origin": base, "Sec-Fetch-Site": "same-origin"}, ) payload = json.load(browser.open(api, timeout=5)) assert payload["groups"]["metrics"][0]["id"] == "apple.hrv" foreign = urllib.request.Request( base + "/api/v1/metric-catalog?q=HRV", headers={"Origin": "http://attacker.invalid", "Sec-Fetch-Site": "cross-site"}, ) with pytest.raises(urllib.error.HTTPError) as blocked: browser.open(foreign, timeout=5) assert blocked.value.code == 403 finally: server.shutdown() server.server_close() def test_weekly_aggregation_is_metric_specific_and_reports_observation_coverage(tmp_path): database = api_database(tmp_path) monday = ANCHOR - timedelta(days=7) sunday = ANCHOR - timedelta(days=1) daily = dispatch_api( database, "/api/v1/series", urlencode({"metric": "apple.steps", "from": monday.isoformat(), "to": sunday.isoformat(), "resolution": "week"}), ) assert daily["aggregation_rule"] == { "id": "complete_calendar_week_daily_metric", "minimum_observations": 7, "requires_complete_calendar_week": True, } assert daily["coverage"]["observation_count"] >= 7 assert daily["points"] and daily["points"][0]["observation_count"] == 7 intermittent = dispatch_api( database, "/api/v1/series", urlencode({"metric": "apple.weight", "from": monday.isoformat(), "to": sunday.isoformat(), "resolution": "week"}), ) assert intermittent["aggregation_rule"] == { "id": "observed_measurements_within_iso_week", "minimum_observations": 1, "requires_complete_calendar_week": False, } assert intermittent["points"] == [{ "week": "2026-W24", "date": "2026-06-10", "week_start": "2026-06-08", "week_end": "2026-06-14", "drilldown_date": "2026-06-08", "value": 70.5, "observation_count": 2, "observed_days": 2, "quality": "observed_measurements", }] assert intermittent["coverage"]["observation_count"] == 2 unsupported = dispatch_api( database, "/api/v1/series", urlencode({"metric": "lab.crp", "from": monday.isoformat(), "to": sunday.isoformat(), "resolution": "week"}), ) assert unsupported["resolution"] == "week" assert unsupported["points"] == [] assert unsupported["aggregation_rule"] == { "id": "unsupported_for_metric", "minimum_observations": None, "requires_complete_calendar_week": False, } def test_machine_readable_apple_identifier_report_has_64_rows_and_never_auto_releases_unknowns(): connection = inventory_db() try: connection.executemany( "INSERT INTO apple_health_records(metric,unit,value,raw_json,start_date) VALUES(?,?,?,?,?)", [ (f"unreleased_metric_{index:02d}", "count", float(index), None, "2026-06-15T12:00:00+02:00") for index in range(1, 52) ], ) connection.commit() report = build_apple_identifier_report(connection, expected_identifier_count=64) finally: connection.close() assert report["schema_version"] == 1 assert report["summary"] == {"identifiers": 64, "released": 12, "not_released": 52} assert len(report["identifiers"]) == 64 unknown = next(item for item in report["identifiers"] if item["identifier"] == "unreleased_metric_01") assert unknown == { "identifier": "unreleased_metric_01", "released": False, "reason": "no_explicit_source_spec_and_contract_test", "observed_units": ["count"], "observations": 1, "priority": "P3_evidence_review", } released = next(item for item in report["identifiers"] if item["identifier"] == "step_count") assert released["released"] is True assert released["reason"] == "explicit_source_spec_and_contract_test" assert released["priority"] == "P1_released"