from __future__ import annotations

import hashlib
import importlib.util
import json
import re
import sqlite3
import sys
import threading
import urllib.error
import urllib.request
from datetime import date
from pathlib import Path
from urllib.parse import urlencode

import pytest

ROOT = Path(__file__).resolve().parents[1]
SCHEMA = ROOT / "database/schema.sql"
DASHBOARD = ROOT / "scripts/health/health_dashboard_v4.py"
SERVER = ROOT / "scripts/health/health_dashboard_server.py"
ACTION_WORKER = ROOT / "scripts/health/health_dashboard_action_worker.py"
CHART = ROOT / "scripts/health/assets/health-assets/chart.umd.min.js"


def load_module(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


@pytest.fixture
def generated_dashboard(tmp_path, monkeypatch):
    db = tmp_path / "health.db"
    c = sqlite3.connect(db)
    c.executescript(SCHEMA.read_text(encoding="utf-8"))
    c.execute(
        """INSERT INTO apple_health_records
           (metric,start_date,end_date,value,unit,source_name,device,file_name,file_hash,raw_json)
           VALUES(?,?,?,?,?,?,?,?,?,?)""",
        (
            "physical_effort",
            "2025-01-02T12:00:00+01:00",
            "2025-01-02T12:05:00+01:00",
            4.2,
            "MET",
            "Synthetic Watch",
            "Synthetic Device",
            "HealthAutoExport_jahr-2025-01-03.json",
            "synthetic-physical-effort",
            "{}",
        ),
    )
    c.execute(
        """INSERT INTO medication_administrations
           (datum,medication_name,dose,event_type)
           VALUES('2025-01-01','ACTUAL_ONLY','40 mg','administered'),
                 ('2025-01-02','PLANNED_ONLY','40 mg','planned'),
                 ('2025-01-03','MISSED_ONLY','40 mg','missed'),
                 ('2025-01-04','CANCELLED_ONLY','40 mg','cancelled')"""
    )
    c.execute(
        """INSERT INTO nutrition_daily_summary_v2
           (datum,kcal,protein_g,carb_g,fat_g,item_count,histamine_score,histamine_unknown_count,histamine_label)
           VALUES('2025-01-01',NULL,NULL,NULL,NULL,1,NULL,1,'green'),
                 ('2025-01-02',0,0,0,0,1,0,0,'green')"""
    )
    c.execute(
        """INSERT INTO nutrition_items
           (id,source,datum,meal,name,amount,kcal,protein_g,carb_g,fat_g,item_hash)
           VALUES(1,'yazio_api','2025-01-02','lunch','INCOMPLETE_GREEN_ITEM',NULL,NULL,NULL,NULL,NULL,'item-incomplete'),
                 (2,'yazio_api','2025-01-02','lunch','COMPLETE_ZERO_GREEN_ITEM',0,0,0,0,0,'item-complete')"""
    )
    c.execute(
        """INSERT INTO nutrition_histamine_scores
           (item_id,canonical_food,sighi_score,traffic_light,confidence)
           VALUES(1,NULL,NULL,'green','high'),
                 (2,'Gurke',0,'green','high')"""
    )
    c.commit()
    c.close()

    module = load_module("health_dashboard_v4_test", DASHBOARD)
    monkeypatch.setattr(module, "DB", db)
    monkeypatch.setattr(module, "BASE", tmp_path)
    monkeypatch.setattr(module, "REPORTS", tmp_path / "reports")
    monkeypatch.setattr(module, "db_counts", lambda: {})
    monkeypatch.setattr(module, "best_reference_xlsx", lambda: tmp_path / "missing.xlsx")
    monkeypatch.setattr(module, "get_lab_matrix", lambda _path: {})
    monkeypatch.setattr(module, "load_doctor_report_meta", lambda: None)
    monkeypatch.setattr(
        sys.modules["apple_health_analytics"],
        "DB",
        tmp_path / "must-not-be-opened" / "production.db",
    )
    output = tmp_path / "reports/health_dashboard_v4.html"
    module.main(output)
    verification = sqlite3.connect(db)
    try:
        assert verification.execute("SELECT COUNT(*) FROM nutrition_daily_features").fetchone()[0] == 0
    finally:
        verification.close()
    return output


def test_dashboard_v4_retains_real_dashboard_and_accessible_shell(generated_dashboard):
    html = generated_dashboard.read_text(encoding="utf-8")

    assert "JARVIS Health Dashboard v4" in html
    assert "Zum Hauptinhalt springen" in html
    assert "id='main-content'" in html
    assert "aria-label='Mobile Schnellnavigation'" in html
    assert "min-height:44px" in html
    assert "prefers-reduced-motion:reduce" in html
    assert "prefers-contrast:more" in html
    assert ":focus-visible" in html
    assert "src='/health-assets/chart.umd.min.js'" in html
    assert "cdn.jsdelivr" not in html
    assert "Relevante Übersicht" in html
    assert "Heute & Aktionen" in html
    assert "Quellenfrische" in html
    assert "Physical Effort" in html
    assert "MET" in html
    medication_card_start = html.index("<div class='card'><h3>Medikation</h3>")
    medication_card = html[medication_card_start:medication_card_start + 900]
    assert "ACTUAL_ONLY" in medication_card
    assert "PLANNED_ONLY" not in medication_card
    assert "MISSED_ONLY" not in medication_card
    assert "CANCELLED_ONLY" not in medication_card
    missing_marker = "<span class='day-date'>2025-01-01</span>"
    zero_marker = "<span class='day-date'>2025-01-02</span>"
    missing_nutrition = html[html.index(missing_marker):html.index(missing_marker) + 500]
    explicit_zero_nutrition = html[html.index(zero_marker):html.index(zero_marker) + 500]
    assert "unbekannt" in missing_nutrition
    assert "Histamin unknown" in missing_nutrition
    assert "0 kcal" in explicit_zero_nutrition
    assert "Load 0.0" in explicit_zero_nutrition
    incomplete_item_at = html.index("INCOMPLETE_GREEN_ITEM")
    complete_item_at = html.index("COMPLETE_ZERO_GREEN_ITEM")
    assert "food-row unknown" in html[incomplete_item_at - 180:incomplete_item_at]
    assert "food-row green" in html[complete_item_at - 180:complete_item_at]
    assert "unbekannt" in html[incomplete_item_at:incomplete_item_at + 400]
    assert "0 kcal" in html[complete_item_at:complete_item_at + 400]
    assert "Daily Symptom Quick Log" in html
    assert "action='/health-actions/symptom-checkin'" in html
    assert "Persönliche 14-Tage-Baseline (Median)" not in html  # no synthetic observations
    assert "data-period='30'" in html
    assert "Explorative multimodale Zusammenhänge" in html


def test_dashboard_v4_has_ephemeral_privacy_and_doctor_print_modes(generated_dashboard):
    html = generated_dashboard.read_text(encoding="utf-8")

    assert "data-sensitive='1'" in html
    assert "applyPrivacyMode" in html
    assert "localStorage" not in html
    assert "@media print" in html
    assert "Evidenz- und Provenienzhinweise" in html
    assert "keine Diagnose, Therapieempfehlung oder Entwarnung" in html
    assert "Fehlende Daten bedeuten unbekannt" in html
    assert "oninput=" not in html
    assert "Ernährungsplan-Konsequenz Legacy" not in html
    assert "erkannte Risiken" not in html
    assert "🟢 safe" not in html
    assert "bisher persönlich dokumentiert vertragen" not in html  # no synthetic rows
    assert "Sie prüfen keine Kausalität" in html
    assert "Potenziell riskante Expositionen nur nach ärztlicher Abstimmung" in html


def test_dashboard_v4_baseline_and_source_freshness_are_missingness_safe():
    module = load_module("health_dashboard_v4_helpers_test", DASHBOARD)
    days = [f"2026-01-{day:02d}" for day in range(1, 11)]
    observed = {day: index for index, day in enumerate(days, 1) if index not in {2, 5}}
    baseline = module.rolling_median_baseline(days, observed)
    assert baseline[:8] == [None] * 8
    assert baseline[8] == 6.0
    assert baseline[9] == 6.5
    assert module.source_freshness("2026-01-09", today="2026-01-10", expected_within_days=2) == ("aktuell", 1)
    assert module.source_freshness("2026-01-01", today="2026-01-10", expected_within_days=2) == ("prüfen", 9)
    assert module.source_freshness(None, today="2026-01-10", expected_within_days=2) == ("unbekannt", None)
    assert module.source_freshness("2026-01-09 nonsense", today="2026-01-10", expected_within_days=2) == ("unbekannt", None)
    assert module.source_freshness("2026-01-09T12", today="2026-01-10", expected_within_days=2) == ("unbekannt", None)
    assert module.source_freshness("2026-01-09T12:34", today="2026-01-10", expected_within_days=2) == ("unbekannt", None)
    assert module.source_freshness("2026-01-09T12:34:56Z", today="2026-01-10", expected_within_days=2) == ("aktuell", 1)
    assert module.source_freshness("2026-01-11", today="2026-01-10", expected_within_days=2) == ("prüfen", -1)
    assert module.is_administered_event("administered")
    assert module.is_administered_event(" VERABREICHT ")
    for non_administration in (None, "", "scheduled", "planned", "cancelled", "missed"):
        assert not module.is_administered_event(non_administration)


def test_pipeline_exposes_dashboard_v4_command(tmp_path, monkeypatch):
    module = load_module("health_pipeline_v4_test", ROOT / "scripts/health/health_pipeline.py")
    calls = {}

    def fake_run(command, check):
        calls["command"] = command

    monkeypatch.setattr(module.subprocess, "run", fake_run)
    output = tmp_path / "dashboard.html"
    module.generate_dashboard_v4(output=output)

    command = calls["command"]
    assert command[0] == module.sys.executable
    assert Path(command[1]).name == "health_dashboard_v4.py"
    assert command[-2:] == ["--output", str(output)]


def test_local_chart_asset_is_exact_pinned_version():
    text = CHART.read_text(encoding="utf-8", errors="strict")
    assert "Chart.js v4.5.1" in text
    assert CHART.stat().st_size == 208_522
    assert hashlib.sha256(CHART.read_bytes()).hexdigest() == (
        "48444a82d4edcb5bec0f1965faacdde18d9c17db3063d042abada2f705c9f54a"
    )


def test_versioned_server_serves_local_asset_and_nonce_csp(tmp_path, monkeypatch):
    dashboard = tmp_path / "dashboard.html"
    dashboard.write_text(
        "<!doctype html><html><head><style>body{color:black}</style>"
        "<script src='/health-assets/chart.umd.min.js'></script></head>"
        "<body><script>window.ok=true</script></body></html>",
        encoding="utf-8",
    )
    asset_dir = tmp_path / "assets"
    asset_dir.mkdir()
    (asset_dir / "chart.umd.min.js").write_bytes(CHART.read_bytes())
    monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard))
    module = load_module("health_dashboard_server_test", SERVER)
    monkeypatch.setattr(module, "ASSET_DIR", asset_dir)
    assert module.consume_csrf_token(
        module.issue_csrf_token(now=100.0),
        now=100.0 + module.CSRF_TTL_SECONDS,
    )
    replayed = module.issue_csrf_token(now=200.0)
    assert module.consume_csrf_token(replayed, now=201.0)
    assert not module.consume_csrf_token(replayed, now=202.0)
    expired = module.issue_csrf_token(now=300.0)
    assert not module.consume_csrf_token(
        expired, now=301.0 + module.CSRF_TTL_SECONDS
    )

    server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    base = f"http://127.0.0.1:{server.server_port}"
    try:
        response = urllib.request.urlopen(base + "/health-dashboard", timeout=5)
        body = response.read().decode("utf-8")
        csp = response.headers["Content-Security-Policy"]
        assert response.status == 200
        assert "default-src 'none'" in csp
        assert "'unsafe-inline'" not in csp
        assert "nonce-" in csp
        assert "<script nonce='" in body
        assert "<style nonce='" in body
        assert "health_csrf=" in response.headers["Set-Cookie"]
        asset_response = urllib.request.urlopen(
            base + "/health-assets/chart.umd.min.js", timeout=5
        )
        assert asset_response.status == 200
        assert asset_response.headers["X-Content-Type-Options"] == "nosniff"
        assert len(asset_response.read()) == CHART.stat().st_size
        evil_request = urllib.request.Request(base + "/health-dashboard", headers={"Host": "evil.example"})
        with pytest.raises(urllib.error.HTTPError) as rejected_host:
            urllib.request.urlopen(evil_request, timeout=5)
        assert rejected_host.value.code == 421
        for method in ("HEAD", "POST", "OPTIONS", "PUT", "DELETE", "PATCH", "TRACE"):
            method_request = urllib.request.Request(
                base + "/health-dashboard",
                headers={"Host": "evil.example"},
                method=method,
            )
            with pytest.raises(urllib.error.HTTPError) as rejected_method:
                urllib.request.urlopen(method_request, timeout=5)
            assert rejected_method.value.code == 421
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def test_server_path_containment_rejects_prefix_siblings_and_escape_symlinks(
    tmp_path, monkeypatch
):
    dashboard = tmp_path / "dashboard.html"
    dashboard.write_text("<html></html>", encoding="utf-8")
    monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard))
    module = load_module("health_dashboard_server_paths_test", SERVER)
    base = tmp_path / "Gesundheit"
    reports = base / "reports"
    reports.mkdir(parents=True)
    sibling = tmp_path / "Gesundheit-evil"
    sibling.mkdir()
    outside = sibling / "secret.pdf"
    outside.write_text("not health data", encoding="utf-8")
    monkeypatch.setattr(module, "BASE", base)
    monkeypatch.setattr(module, "REPORTS", reports)

    assert module.resolve_doc_path(outside) is None
    (reports / "escape.pdf").symlink_to(outside)
    assert module.safe_report_path("escape.pdf") is None
    valid = reports / "valid.pdf"
    valid.write_text("synthetic", encoding="utf-8")
    assert module.safe_report_path("valid.pdf") == valid.resolve()


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def test_touch_checkin_requires_same_origin_one_time_csrf_and_complete_scores(
    tmp_path, monkeypatch
):
    dashboard = tmp_path / "dashboard.html"
    dashboard.write_text(
        "<html><body><form><input value='__CSRF_TOKEN__'></form></body></html>",
        encoding="utf-8",
    )
    asset_dir = tmp_path / "assets"
    asset_dir.mkdir()
    (asset_dir / "chart.umd.min.js").write_bytes(CHART.read_bytes())
    monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard))
    module = load_module("health_dashboard_server_post_test", SERVER)
    monkeypatch.setattr(module, "ASSET_DIR", asset_dir)
    saved = []
    monkeypatch.setattr(
        module,
        "write_symptom_checkin",
        lambda day, scores, notes: saved.append((day, scores, notes)),
    )
    server = module.ThreadingHTTPServer(("127.0.0.1", 0), module.Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    base = f"http://127.0.0.1:{server.server_port}"
    opener = urllib.request.build_opener(NoRedirect)
    try:
        response = opener.open(base + "/health-dashboard", timeout=5)
        body = response.read().decode("utf-8")
        token_match = re.search(r"value='([^']+)'", body)
        assert token_match
        token = token_match.group(1)
        cookie = response.headers["Set-Cookie"].split(";", 1)[0]
        cookie += f"; health_api_session={module.issue_browser_session()}"
        fields = {
            "csrf_token": token,
            "date": date.today().isoformat(),
            "aphthen": "0",
            "gi": "1",
            "fatigue": "2",
            "skin": "0",
            "eyes": "0",
            "joints": "1",
            "vascular": "0",
            "notes": "synthetic test",
        }
        request = urllib.request.Request(
            base + "/health-actions/symptom-checkin",
            data=urlencode(fields).encode(),
            headers={"Origin": base, "Cookie": cookie},
            method="POST",
        )
        with pytest.raises(urllib.error.HTTPError) as redirect:
            opener.open(request, timeout=5)
        assert redirect.value.code == 303
        assert saved == [
            (
                date.today().isoformat(),
                {
                    "aphthen": 0,
                    "gi": 1,
                    "fatigue": 2,
                    "skin": 0,
                    "eyes": 0,
                    "joints": 1,
                    "vascular": 0,
                },
                "synthetic test",
            )
        ]
        with pytest.raises(urllib.error.HTTPError) as replay:
            opener.open(request, timeout=5)
        assert replay.value.code == 403
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


def test_symptom_form_rejects_missing_future_and_out_of_range_values(
    tmp_path, monkeypatch
):
    dashboard = tmp_path / "dashboard.html"
    dashboard.write_text("<html></html>", encoding="utf-8")
    monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard))
    module = load_module("health_dashboard_server_validation_test", SERVER)
    complete = {field: ["0"] for field in module.SYMPTOM_FIELDS}
    complete["date"] = [date.today().isoformat()]
    day, scores, notes = module.validate_symptom_form(complete)
    assert day == date.today().isoformat()
    assert len(scores) == 7
    assert notes == ""

    missing = dict(complete)
    missing.pop("vascular")
    with pytest.raises(ValueError):
        module.validate_symptom_form(missing)
    invalid = dict(complete)
    invalid["gi"] = ["4"]
    with pytest.raises(ValueError):
        module.validate_symptom_form(invalid)
    future = dict(complete)
    future["date"] = ["2999-01-01"]
    with pytest.raises(ValueError):
        module.validate_symptom_form(future)


def test_network_server_only_queues_atomic_private_action(tmp_path, monkeypatch):
    dashboard = tmp_path / "dashboard.html"
    dashboard.write_text("<html></html>", encoding="utf-8")
    monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard))
    module = load_module("health_dashboard_server_queue_test", SERVER)
    inbox = tmp_path / "actions"
    monkeypatch.setattr(module, "ACTION_INBOX", inbox)
    scores = {field: index % 4 for index, field in enumerate(module.SYMPTOM_FIELDS)}
    module.write_symptom_checkin(date.today().isoformat(), scores, "synthetic")

    queued = list(inbox.glob("*.json"))
    assert len(queued) == 1
    assert queued[0].stat().st_mode & 0o777 == 0o600
    assert not list(inbox.glob("*.tmp"))
    payload = queued[0].read_text(encoding="utf-8")
    assert "synthetic" in payload


def test_action_worker_revalidates_and_consumes_queue(tmp_path, monkeypatch):
    module = load_module("health_dashboard_action_worker_test", ACTION_WORKER)
    inbox = tmp_path / "actions"
    inbox.mkdir()
    monkeypatch.setattr(module, "ACTION_INBOX", inbox)
    monkeypatch.setattr(module, "DASHBOARD_DB", tmp_path / "synthetic.db")
    scores = {field: index % 4 for index, field in enumerate(module.FIELDS)}
    payload = {
        "version": 1,
        "action": "symptom_checkin",
        "date": date.today().isoformat(),
        "scores": scores,
        "notes": "synthetic",
    }
    action = inbox / ("a" * 32 + ".json")
    action.write_text(json.dumps(payload), encoding="utf-8")
    action.chmod(0o600)
    applied = []
    monkeypatch.setattr(
        module,
        "apply_action",
        lambda day, values, notes: applied.append((day, values, notes)),
    )
    assert module.main() == 0
    assert not action.exists()
    assert applied == [(date.today().isoformat(), scores, "synthetic")]

    invalid = inbox / ("b" * 32 + ".json")
    invalid.write_text("{}", encoding="utf-8")
    invalid.chmod(0o600)
    assert module.main() == 1
    assert invalid.with_suffix(".failed").exists()

    score_pairs = ",".join(f'"{field}":0' for field in module.FIELDS)
    duplicate_top = inbox / ("c" * 32 + ".json")
    duplicate_top.write_text(
        '{"version":1,"version":1,"action":"symptom_checkin","date":"'
        + date.today().isoformat()
        + '","scores":{'
        + score_pairs
        + '},"notes":""}',
        encoding="utf-8",
    )
    duplicate_top.chmod(0o600)
    with pytest.raises(ValueError, match="duplicate JSON key"):
        module.load_action(duplicate_top)

    duplicate_score = inbox / ("d" * 32 + ".json")
    duplicate_score.write_text(
        '{"version":1,"action":"symptom_checkin","date":"'
        + date.today().isoformat()
        + '","scores":{'
        + score_pairs
        + ',"gi":1},"notes":""}',
        encoding="utf-8",
    )
    duplicate_score.chmod(0o600)
    with pytest.raises(ValueError, match="duplicate JSON key"):
        module.load_action(duplicate_score)