"""Sprint-5C mobile capture and private action-queue regressions.""" from __future__ import annotations import importlib.util import re import stat import sys import threading import urllib.error import urllib.request from datetime import date, datetime, timedelta, timezone from pathlib import Path from urllib.parse import urlencode import pytest ROOT = Path(__file__).resolve().parents[1] SERVER = ROOT / "scripts/health/health_dashboard_server.py" WORKER = ROOT / "scripts/health/health_dashboard_action_worker.py" QUICK_ADD = ROOT / "scripts/health/health_symptom_quick_add.py" DASHBOARD_V5_GENERATOR = ROOT / "scripts/health/health_dashboard_v5.py" 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 def complete_form(module) -> dict[str, list[str]]: form = {field: ["0"] for field in module.SYMPTOM_FIELDS} form.update({ "csrf_token": ["synthetic-csrf"], "date": [module.local_today().isoformat()], "notes": [""], "return_to": ["v5"], }) return form class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def test_v5_post_queues_once_and_redirects_back_to_v5(tmp_path, monkeypatch): dashboard = tmp_path / "v5.html" dashboard.write_text( "

" "" "" "", encoding="utf-8", ) monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_V5_FILE", str(dashboard)) monkeypatch.delenv("HEALTH_DASHBOARD_DB", raising=False) module = load_module("health_dashboard_server_sprint5c_post", SERVER) monkeypatch.setattr(module, "local_today", lambda: date(2026, 1, 2)) saved: list[tuple[str, dict[str, int], str]] = [] 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) monkeypatch.setattr(module, "EXTERNAL_PORT", server.server_port) 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) def fresh_token_and_cookie() -> tuple[str, str]: response = opener.open(base + module.V5_ROUTE, timeout=5) body = response.read().decode("utf-8") assert "value='2026-01-02'" in body assert "" in body match = re.search(r"name='csrf_token' value='([^']+)'", body) assert match session_id = module.issue_browser_session() cookie = response.headers["Set-Cookie"].split(";", 1)[0] return match.group(1), f"{cookie}; health_api_session={session_id}" try: token, cookie = fresh_token_and_cookie() fields = {field: "0" for field in module.SYMPTOM_FIELDS} fields.update({ "csrf_token": token, "date": "2026-01-02", "notes": "synthetic", "return_to": "v5", }) request = urllib.request.Request( base + module.CHECKIN_ROUTE, data=urlencode(fields).encode(), headers={"Origin": base, "Cookie": cookie}, method="POST", ) wrong_origin = urllib.request.Request( base + module.CHECKIN_ROUTE, data=urlencode(fields).encode(), headers={"Origin": "https://evil.invalid", "Cookie": cookie}, method="POST", ) with pytest.raises(urllib.error.HTTPError) as origin_rejected: opener.open(wrong_origin, timeout=5) assert origin_rejected.value.code == 403 assert saved == [] with pytest.raises(urllib.error.HTTPError) as redirect: opener.open(request, timeout=5) assert redirect.value.code == 303 location = redirect.value.headers["Location"] assert re.fullmatch(r"/health-dashboard-v5\?queued=[A-Za-z0-9_-]+#capture-status", location) head = urllib.request.Request(base + location, method="HEAD") assert opener.open(head, timeout=5).status == 200 confirmation = opener.open(base + location, timeout=5).read().decode("utf-8") assert "data-queued='true'" in confirmation replayed_confirmation = opener.open(base + location, timeout=5).read().decode("utf-8") assert "data-queued='false'" in replayed_confirmation forged_confirmation = opener.open(base + module.V5_ROUTE + "?queued=forged", timeout=5).read().decode("utf-8") assert "data-queued='false'" in forged_confirmation assert len(saved) == 1 token, cookie = fresh_token_and_cookie() fields["csrf_token"] = token fields["unexpected"] = "blocked" invalid = urllib.request.Request( base + module.CHECKIN_ROUTE, data=urlencode(fields).encode(), headers={"Origin": base, "Cookie": cookie}, method="POST", ) with pytest.raises(urllib.error.HTTPError) as rejected: opener.open(invalid, timeout=5) assert rejected.value.code == 400 assert len(saved) == 1 monkeypatch.setattr( module, "write_symptom_checkin", lambda *_args: (_ for _ in ()).throw(module.QueueFullError("full")), ) token, cookie = fresh_token_and_cookie() fields.pop("unexpected") fields["csrf_token"] = token full = urllib.request.Request( base + module.CHECKIN_ROUTE, data=urlencode(fields).encode(), headers={"Origin": base, "Cookie": cookie}, method="POST", ) with pytest.raises(urllib.error.HTTPError) as unavailable: opener.open(full, timeout=5) assert unavailable.value.code == 503 finally: server.shutdown() server.server_close() thread.join(timeout=5) def test_capture_submission_is_exact_single_value_and_v5_scoped(tmp_path, monkeypatch): dashboard = tmp_path / "dashboard.html" dashboard.write_text("", encoding="utf-8") monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) module = load_module("health_dashboard_server_sprint5c_contract", SERVER) assert module.origin_matches_request("http://localhost:8014", "localhost:8014") assert not module.origin_matches_request("https://localhost:8014", "localhost:8014") assert not module.origin_matches_request("https://evil.invalid:8014", "evil.invalid:8014") assert not module.origin_matches_request("http://localhost:8015", "localhost:8014") assert not module.origin_matches_request("http://localhost:8014/path", "localhost:8014") day, scores, notes, return_to = module.validate_symptom_submission(complete_form(module)) assert day == module.local_today().isoformat() assert len(scores) == 7 assert notes == "" assert return_to == "v5" unknown = complete_form(module) unknown["unexpected"] = ["value"] with pytest.raises(ValueError, match="shape"): module.validate_symptom_submission(unknown) duplicate = complete_form(module) duplicate["gi"] = ["0", "3"] with pytest.raises(ValueError, match="single value"): module.validate_symptom_submission(duplicate) wrong_return = complete_form(module) wrong_return["return_to"] = ["https://example.invalid"] with pytest.raises(ValueError, match="return target"): module.validate_symptom_submission(wrong_return) def test_private_inbox_permissions_are_repaired_and_queue_is_bounded(tmp_path, monkeypatch): dashboard = tmp_path / "dashboard.html" dashboard.write_text("", encoding="utf-8") monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) module = load_module("health_dashboard_server_sprint5c_queue", SERVER) inbox = tmp_path / "actions" inbox.mkdir(mode=0o755) monkeypatch.setattr(module, "ACTION_INBOX", inbox) monkeypatch.setattr(module, "MAX_PENDING_ACTIONS", 2) scores = {field: 0 for field in module.SYMPTOM_FIELDS} module.write_symptom_checkin(date.today().isoformat(), scores, "first") module.write_symptom_checkin((date.today() + timedelta(days=1)).isoformat(), scores, "second") assert stat.S_IMODE(inbox.stat().st_mode) == 0o700 assert all(stat.S_IMODE(path.stat().st_mode) == 0o600 for path in inbox.glob("*.json")) with pytest.raises(module.QueueFullError): module.write_symptom_checkin((date.today() + timedelta(days=2)).isoformat(), scores, "third") def test_worker_rejects_insecure_action_file_mode(tmp_path, monkeypatch): module = load_module("health_dashboard_action_worker_sprint5c", WORKER) inbox = tmp_path / "actions" inbox.mkdir(mode=0o700) monkeypatch.setattr(module, "ACTION_INBOX", inbox) action = inbox / ("a" * 32 + ".json") action.write_text("{}", encoding="utf-8") action.chmod(0o644) with pytest.raises(ValueError, match="permissions"): module.load_action(action) def test_worker_requires_explicit_database_before_touching_inbox(tmp_path, monkeypatch, capsys): module = load_module("health_dashboard_action_worker_sprint5c_no_db", WORKER) inbox = tmp_path / "must-not-exist" monkeypatch.setattr(module, "DASHBOARD_DB", None) monkeypatch.setattr(module, "ACTION_INBOX", inbox) assert module.main() == 2 assert not inbox.exists() assert "HEALTH_DASHBOARD_DB is required" in capsys.readouterr().err def test_worker_passes_explicit_database_to_mutation_and_v5_refresh(tmp_path, monkeypatch): module = load_module("health_dashboard_action_worker_sprint5c_commands", WORKER) database = tmp_path / "synthetic.db" monkeypatch.setattr(module, "DASHBOARD_DB", database) monkeypatch.setattr(module, "DASHBOARD_V5_FILE", tmp_path / "v5.html") monkeypatch.setattr(module, "local_today", lambda: date(2026, 1, 2)) commands: list[list[str]] = [] def fake_run(command, **_kwargs): commands.append(command) monkeypatch.setattr(module.subprocess, "run", fake_run) module.apply_action(date.today().isoformat(), {field: 0 for field in module.FIELDS}, "") assert [Path(command[1]).name for command in commands] == [ "health_symptom_quick_add.py", "health_dashboard_v5.py", ] assert commands[0][2:4] == ["--db", str(database)] assert commands[-1][2:4] == ["--db", str(database)] assert commands[-1][-2:] == ["--today", "2026-01-02"] def test_v5_generator_defaults_to_zurich_day_not_host_day(tmp_path, monkeypatch): module = load_module("health_dashboard_v5_sprint5c_timezone", DASHBOARD_V5_GENERATOR) database = tmp_path / "synthetic.db" output = tmp_path / "v5.html" captured = {} monkeypatch.setattr(module, "local_today", lambda: date(2026, 1, 2)) def fake_generate(db, destination, *, today, generated_at=None, runtime_commit=None): captured.update( db=db, output=destination, today=today, generated_at=generated_at, runtime_commit=runtime_commit, ) return destination monkeypatch.setattr(module, "generate", fake_generate) assert module.main(["--db", str(database), "--output", str(output)]) == 0 assert captured == { "db": database, "output": output, "today": "2026-01-02", "generated_at": None, "runtime_commit": None, } def test_capture_pipeline_ignores_external_timezone_override(tmp_path, monkeypatch): dashboard = tmp_path / "v5.html" dashboard.write_text("", encoding="utf-8") monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_TIMEZONE", "Pacific/Kiritimati") modules = [ load_module("health_dashboard_server_sprint5c_fixed_tz", SERVER), load_module("health_dashboard_worker_sprint5c_fixed_tz", WORKER), load_module("health_symptom_quick_add_sprint5c_fixed_tz", QUICK_ADD), load_module("health_dashboard_v5_sprint5c_fixed_tz", DASHBOARD_V5_GENERATOR), ] class FrozenDatetime: @classmethod def now(cls, tz): instant = datetime(2026, 7, 13, 10, 30, tzinfo=timezone.utc) return instant.astimezone(tz) for module in modules: monkeypatch.setattr(module, "datetime", FrozenDatetime) assert module.local_today() == date(2026, 7, 13) assert modules[0].CAPTURE_TIMEZONE.key == "Europe/Zurich" assert all(module.LOCAL_TIMEZONE.key == "Europe/Zurich" for module in modules[1:]) def test_quick_add_refresh_targets_the_explicit_database(tmp_path, monkeypatch): module = load_module("health_symptom_quick_add_sprint5c", QUICK_ADD) commands: list[list[str]] = [] monkeypatch.setattr(module.subprocess, "run", lambda command, check: commands.append(command)) database = tmp_path / "synthetic.db" module.refresh(False, database) assert commands == [[ sys.executable, str(module.CORRELATIONS), "--db", str(database), ]] def test_v5_capture_markup_matches_server_contract(): source = (ROOT / "scripts/health/dashboard_v5/render.py").read_text(encoding="utf-8") assert "name='return_to' value='v5'" in source assert "name='date' value='__CAPTURE_DATE__'" in source assert "__CAPTURE_DATE_LABEL__" in source assert "id='capture-status'" in source assert "" in source def test_systemd_units_create_private_queue_state(): for name in ("health-dashboard.service", "health-dashboard-action-worker.service"): unit = (ROOT / "deploy/systemd" / name).read_text(encoding="utf-8") assert "UMask=0077" in unit assert "StateDirectoryMode=0700" in unit worker = (ROOT / "deploy/systemd/health-dashboard-action-worker.service").read_text(encoding="utf-8") assert "ExecStart=%h/.hermes/hermes-agent/venv/bin/python3 " in worker def test_synthetic_server_identity_cannot_point_at_a_non_tmp_queue(tmp_path, monkeypatch): dashboard = tmp_path / "v5.html" dashboard.write_text("", encoding="utf-8") monkeypatch.setenv("HEALTH_DASHBOARD_FILE", str(dashboard)) monkeypatch.setenv("HEALTH_DASHBOARD_TEST_INSTANCE_ID", "sprint5c-safe") monkeypatch.setenv("HEALTH_DASHBOARD_ACTION_INBOX", "/home/agent/not-a-test-queue") with pytest.raises(RuntimeError, match="below /tmp"): load_module("health_dashboard_server_sprint5c_bad_test_target", SERVER)