"""Isolated integration tests. Requires httpx in the test environment only. Run: python -m unittest -v _fleet_test All devices, credentials, logs and OTA files are disposable simulations. """ import json import asyncio import ast import io import logging import re import os from pathlib import Path import subprocess import sqlite3 import sys import tempfile import time import unittest from unittest.mock import patch from urllib.parse import urlsplit import anyio from fastapi.testclient import TestClient from starlette.websockets import WebSocketDisconnect ROOT = Path(__file__).resolve().parent A = "esp32-001122aabbcc" B = "esp32-001122aabbdd" ORIGIN = "http://localhost:4000" TEL = ("LOG:[2026-10-30 10:34:03] TEL: t=246s, steps=29825(+1000), " "avgC=5469365, avgV=5160269, avgP=679686, LPF=4338597, " "prox=0.00, CmaxN=53643680, X=-9827, Y=-65788") SAMPLE = "LOG:[2026-10-30 10:34:07] X:54810,Y:48890,C:5469365,V:5160269,P:679686,S:17,er:1500" class FleetTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp = tempfile.TemporaryDirectory(prefix="fleet-test-") cls.directory = Path(cls.temp.name) cls.environment = { "STATE_DIR": str(cls.directory / "state"), "LOGS_DIR": str(cls.directory / "logs"), "FIRMWARE_DIR": str(cls.directory / "firmware"), "ALLOWED_ORIGINS": ORIGIN, "PRIMARY_DEVICE_ID": "", "CREDENTIAL_POLL_SECONDS": "0.1", } cls.original_env = {key: os.environ.get(key) for key in cls.environment} os.environ.update(cls.environment) cls.tokens = {} for device, alias in ((A, "Tracker A"), (B, "Tracker B")): token_file = cls.directory / "state" / "provision" / (device + ".v1.token") cls.cli("add", "--device-id", device, "--alias", alias) cls.tokens[device] = token_file.read_text(encoding="utf-8").strip() # Import only after redirecting every persistent path. from app import config for key in ("STATE_DIR", "LOGS_DIR", "FIRMWARE_DIR"): assert Path(getattr(config, key)).resolve() == Path(cls.environment[key]).resolve() from app.main import app cls.client_context = TestClient(app) cls.client = cls.client_context.__enter__() @classmethod def tearDownClass(cls): cls.client_context.__exit__(None, None, None) for key, value in cls.original_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value cls.temp.cleanup() @classmethod def cli(cls, *arguments): result = subprocess.run( [sys.executable, "-m", "app.devices", *arguments], cwd=ROOT, env=os.environ.copy(), capture_output=True, text=True, timeout=15, ) if result.returncode: raise AssertionError("CLI failed: " + result.stderr) return result def api(self, device, suffix): return "/api/devices/" + device + suffix def esp(self, device, token=None): return self.client.websocket_connect( "/ws/esp32?device_id=" + device, headers={"Authorization": "Bearer " + (token or self.tokens[device])}, ) def browser(self, device): return self.client.websocket_connect( "/ws/client?device_id=" + device, headers={"Origin": ORIGIN}, ) def receive_text(self, socket, timeout=8): # Starlette's synchronous receive has no deadline; bound its async stream. async def bounded(): with anyio.fail_after(timeout): return await socket._send_rx.receive() message = socket.portal.call(bounded) socket._raise_on_close(message) return message["text"] def until(self, socket, prefix, containing=None): deadline = time.monotonic() + 8 for _ in range(60): remaining = deadline - time.monotonic() self.assertGreater(remaining, 0, "Deadline waiting for " + prefix) message = self.receive_text(socket, remaining) if message.startswith(prefix) and (containing is None or containing in message): return message[len(prefix):] self.fail("Expected browser prefix " + prefix) def test_01_registry_and_rejections(self): rows = self.client.get("/api/devices").json() self.assertEqual({r["device_id"] for r in rows}, {A, B}) self.assertTrue(all("token" not in key and "hash" not in key for r in rows for key in r)) security_log = io.StringIO() handler = logging.StreamHandler(security_log) logger = logging.getLogger("tracker_hub") logger.addHandler(handler) try: for device, token in ((A, "invalid"), (B, self.tokens[A]), ("esp32-000000000000", self.tokens[A])): with self.assertRaises(WebSocketDisconnect): with self.esp(device, token): self.fail("unauthenticated connection accepted") finally: logger.removeHandler(handler) self.assertIn("esp32_auth_failed", security_log.getvalue()) for token in self.tokens.values(): self.assertNotIn(token, security_log.getvalue()) with self.assertRaises(WebSocketDisconnect): with self.client.websocket_connect("/ws/esp32?device_id=" + A): self.fail("anonymous device accepted") with self.assertRaises(WebSocketDisconnect): with self.client.websocket_connect("/ws/client?device_id=" + A, headers={"Origin": "https://evil.invalid"}): self.fail("foreign browser origin accepted") self.assertEqual(self.client.get(self.api("esp32-000000000000", "/status")).status_code, 404) self.assertGreaterEqual(self.client.get("/api/status").status_code, 400) def test_02_two_devices_and_scoped_commands(self): with self.esp(A) as a, self.esp(B) as b, self.browser(A) as ca, self.browser(B) as cb: self.assertEqual(json.loads(self.until(ca, "CONFIG:"))["device_id"], A) self.assertEqual(json.loads(self.until(cb, "CONFIG:"))["device_id"], B) a.send_text(TEL) point_a = json.loads(self.until(ca, "TEL:")) b.send_text(TEL.replace("X=-9827", "X=-100")) point_b = json.loads(self.until(cb, "TEL:")) self.assertEqual((point_a["device_id"], point_a["x"]), (A, 9827)) self.assertEqual((point_b["device_id"], point_b["x"]), (B, 100)) a.send_text(SAMPLE) self.assertEqual(json.loads(self.until(ca, "DATA:"))["device_id"], A) b.send_text("VER:test-B") self.assertEqual(self.until(cb, "FW:"), "test-B") ca.send_text("CMD:10") self.assertEqual(self.receive_text(a), "CMD:10") for invalid in ("OTA:https://evil.invalid/payload.bin", "SET_COORDS:nan:0", "SET_PARAM:errMax=inf"): ca.send_text(invalid) self.until(ca, "LOG:", containing="ERROR") ca.send_text("CMD:10") self.assertEqual(self.receive_text(a), "CMD:10", "invalid commands must not reach the device") cb.send_text("SET_COORDS:12:34") self.assertEqual(self.receive_text(b), "SET_COORDS:12:34") status_a = self.client.get(self.api(A, "/status")).json() status_b = self.client.get(self.api(B, "/status")).json() self.assertNotEqual(status_a["firmware_device"], status_b["firmware_device"]) self.assertEqual(status_b["firmware_device"], "test-B") for device in (A, B): rows = self.client.get(self.api(device, "/logs")).json() self.assertTrue(rows) response = self.client.get(self.api(device, "/logs/" + rows[0]["filename"])) self.assertEqual(response.status_code, 200) if device == A: self.assertNotIn("test-B", response.text) def test_03_configuration_isolation_and_validation(self): with self.browser(A) as ca, self.browser(B) as cb: self.until(ca, "CONFIG:") self.until(cb, "CONFIG:") before_b = self.client.get(self.api(B, "/calib")).json() with self.esp(A) as a: a.send_text(SAMPLE) before = json.loads(self.until(ca, "DATA:"))["isc_ma"] self.assertAlmostEqual(before, 5469365 / 8388607 * 2048 / 20 / 0.075, delta=0.01) response = self.client.post(self.api(A, "/pcb"), json={"version": "V2"}) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["r_shunt"], 0.130) history = json.loads(self.until(ca, "HISTORY:")) # The first HISTORY is the initial replay; consume until recalculated history. if not history or history[-1]["isc_ma"] == before: history = json.loads(self.until(ca, "HISTORY:")) self.assertAlmostEqual(history[-1]["isc_ma"] / before, 0.075 / 0.130, delta=0.001) self.assertAlmostEqual(history[-1]["isc_ma"], 5469365 / 8388607 * 2048 / 20 / 0.130, delta=0.01) pcb_b = self.client.get(self.api(B, "/pcb")).json() self.assertEqual((pcb_b["version"], pcb_b["r_shunt"]), ("V1", 0.075)) self.assertEqual(self.client.post(self.api(A, "/calib"), json={"mV_per_WM2": 0.3, "az_dev": 18}).status_code, 200) self.assertEqual(self.client.get(self.api(B, "/calib")).json(), before_b) self.assertEqual(self.client.get(self.api(A, "/pcb")).json()["version"], "V2") self.assertEqual(self.client.post(self.api(A, "/pcb"), json={"version": "V9"}).status_code, 400) for value in (0, -1, "nan", "inf"): self.assertEqual(self.client.post(self.api(A, "/calib"), json={"mV_per_WM2": value}).status_code, 400) def test_04_reconnect_old_socket_cannot_disconnect_new(self): with self.esp(A) as old, self.browser(A) as browser: self.until(browser, "CONFIG:") with self.esp(A) as new: old.close() new.send_text("VER:new-session") self.assertEqual(self.until(browser, "FW:"), "new-session") self.assertTrue(self.client.get(self.api(A, "/status")).json()["esp32_connected"]) browser.send_text("CMD:10") self.assertEqual(self.receive_text(new), "CMD:10") def test_05_ota_is_manual_and_scoped_to_confirmed_image(self): with self.esp(A) as a, self.esp(B) as b: # A simulated ESP32 is simply a WebSocket inbox; no hardware is touched. response = self.client.post("/ota/upload", files={"file": ("test.bin", b"test-firmware-1")}, data={"version": "test-2"}) self.assertEqual(response.status_code, 200) manifest = response.json()["manifest"] self.assertEqual(self.client.post(self.api(A, "/ota/push"), json={"sha256": "wrong"}).status_code, 409) sent = self.client.post(self.api(B, "/ota/push"), json={"sha256": manifest["sha256"]}) self.assertEqual(sent.status_code, 200) self.assertEqual(sent.json()["device_id"], B) ota_command = self.receive_text(b) self.assertTrue(ota_command.startswith("OTA:")) # A receives only its explicitly requested status command, not B's OTA. with self.browser(A) as browser: self.until(browser, "CONFIG:") browser.send_text("CMD:10") self.assertEqual(self.receive_text(a), "CMD:10") self.assertFalse(self.client.get(self.api(A, "/status")).json()["ota"]["auto"]) self.assertFalse(self.client.get(self.api(B, "/status")).json()["ota"]["auto"]) self.assertEqual(self.client.get("/ota/firmware").content, b"test-firmware-1") def test_06_log_paths_and_disconnected_command(self): self.assertEqual(self.client.get(self.api(A, "/logs/log_1990-01-01.txt")).status_code, 404) self.assertEqual(self.client.get(self.api(A, "/logs/missing.txt")).status_code, 400) self.assertEqual(self.client.get(self.api(A, "/logs/..%2F..%2Fstate%2Fdevices.sqlite3")).status_code, 404) with self.browser(A) as browser: self.until(browser, "CONFIG:") # Drain initial replay before checking the explicit command failure. self.until(browser, "OTA_STATE:") browser.send_text("CMD:10") self.assertIn("ERROR", self.until(browser, "LOG:", containing="ERROR")) def test_07_new_process_persists_settings_but_not_connections(self): script = ( "import json\nfrom fastapi.testclient import TestClient\nfrom app.main import app\n" "with TestClient(app) as client:\n" f" base='/api/devices/{A}'\n" " print('__RESULT__'+json.dumps({'pcb':client.get(base+'/pcb').json()," "'calib':client.get(base+'/calib').json(),'status':client.get(base+'/status').json()}))" ) result = subprocess.run([sys.executable, "-c", script], cwd=ROOT, env=os.environ.copy(), capture_output=True, text=True, timeout=15) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(next(line[len("__RESULT__"):] for line in result.stdout.splitlines() if line.startswith("__RESULT__"))) self.assertEqual(payload["pcb"]["version"], "V2") self.assertEqual(payload["calib"]["mV_per_WM2"], 0.3) self.assertEqual(payload["calib"]["az_dev"], 18) self.assertFalse(payload["status"]["esp32_connected"]) self.assertEqual(payload["status"]["samples_in_buffer"], 0) for token in self.tokens.values(): self.assertNotIn(token, result.stdout + result.stderr) listing = self.cli("list") for token in self.tokens.values(): self.assertNotIn(token, listing.stdout + listing.stderr) def test_08_upload_never_triggers_ota_even_when_device_opted_in(self): original = self.client.get("/ota/manifest").json() with self.esp(A) as a, self.browser(A) as browser: self.until(browser, "CONFIG:") a.send_text("VER:" + original["version"]) self.until(browser, "FW:") self.assertEqual(self.client.post(self.api(A, "/ota/auto"), json={"enabled": True}).status_code, 200) response = self.client.post("/ota/upload", files={"file": ("test.bin", b"test-firmware-2")}, data={"version": "test-3"}) self.assertEqual(response.status_code, 200) try: # Outwait the legacy 3-second auto-update delay to catch accidental push-on-upload. time.sleep(4) browser.send_text("CMD:10") self.assertEqual(self.receive_text(a), "CMD:10", "upload unexpectedly pushed firmware") self.assertEqual(self.client.post(self.api(A, "/ota/push"), json={"sha256": original["sha256"]}).status_code, 409) self.assertFalse(self.client.get(self.api(B, "/status")).json()["ota"]["auto"]) old_url = urlsplit(original["url"]) self.assertEqual(self.client.get(old_url.path).content, b"test-firmware-1", "an already confirmed image URL must remain immutable") finally: self.client.post(self.api(A, "/ota/auto"), json={"enabled": False}) def test_09_rotation_revocation_and_database_secrecy(self): previous = self.tokens[B] self.cli("rotate", "--device-id", B, "--grace-hours", "1") token_file = self.directory / "state" / "provision" / (B + ".v2.token") current = token_file.read_text(encoding="utf-8").strip() self.assertNotEqual(previous, current) with self.esp(B, previous): pass with self.esp(B, current): pass self.cli("retire-previous", "--device-id", B) with self.assertRaises(WebSocketDisconnect): with self.esp(B, previous): self.fail("retired token accepted") with self.esp(B, current) as active: self.cli("revoke", "--device-id", B) with self.assertRaises(WebSocketDisconnect): self.receive_text(active) with self.assertRaises(WebSocketDisconnect): with self.esp(B, current): self.fail("revoked token accepted") self.tokens[B] = current for database in (self.directory / "state").glob("*.sqlite3*"): content = database.read_bytes() for token in (previous, current, self.tokens[A]): self.assertFalse(token.encode() in content, "plaintext token leaked into SQLite") with self.esp(A): pass # Revoking B must leave A's credential valid. def test_10_explicit_idempotent_primary_migration(self): c = "esp32-001122aabbee" self.cli("add", "--device-id", c, "--alias", "Migration target") legacy = self.directory / "firmware" / "pcb_state.json" legacy.write_text(json.dumps({"version": "V2"}), encoding="utf-8") (legacy.parent / "ota_state.json").write_text(json.dumps({"auto": True}), encoding="utf-8") self.assertEqual(self.client.get(self.api(c, "/pcb")).json()["version"], "V1") self.cli("migrate-primary", "--device-id", c) self.assertEqual(self.client.get(self.api(c, "/pcb")).json()["version"], "V2") self.assertEqual(self.client.get(self.api(B, "/pcb")).json()["version"], "V1") self.assertFalse(self.client.get(self.api(c, "/status")).json()["ota"]["auto"]) # A repeated migration must not overwrite a later explicit setting. self.client.post(self.api(c, "/pcb"), json={"version": "V1"}) self.cli("migrate-primary", "--device-id", c) self.assertEqual(self.client.get(self.api(c, "/pcb")).json()["version"], "V1") self.assertEqual(json.loads(legacy.read_text(encoding="utf-8"))["version"], "V2") def test_11_oversized_session_isolated_and_corrupt_database_fails_closed(self): with self.esp(A) as a: a.send_text("x" * 16385) with self.assertRaises(WebSocketDisconnect): self.receive_text(a) self.assertEqual(self.client.get(self.api(B, "/status")).status_code, 200) broken = self.directory / "broken-state" broken.mkdir() (broken / "devices.sqlite3").write_bytes(b"not a sqlite database") env = dict(os.environ, STATE_DIR=str(broken)) result = subprocess.run([sys.executable, "-c", "from app.main import app"], cwd=ROOT, env=env, capture_output=True, text=True, timeout=15) self.assertNotEqual(result.returncode, 0, "corrupt registry must not fall back to anonymous access") def test_12_ended_sessions_and_failed_accept_are_inert(self): from app.hub import HubManager class Socket: def __init__(self, fail=False): self.fail = fail self.closed = False async def accept(self): if self.fail: raise RuntimeError("simulated accept failure") async def close(self, *args, **kwargs): self.closed = True async def send_text(self, message): pass async def check(): manager = HubManager(A) original = Socket() connection = await manager.connect_esp32(original) await manager.handle_esp32_message("POS:1:2", connection) with self.assertRaises(RuntimeError): await manager.connect_esp32(Socket(fail=True)) self.assertIs(manager.esp32_socket, original) self.assertEqual(manager.connection_id, connection) manager.disconnect_esp32(connection) await manager.handle_esp32_message("POS:999:999", connection) self.assertEqual((manager.last_pos_x, manager.last_pos_y), (1, 2)) self.assertIsNone(manager.connection_id) connection = await manager.connect_esp32(Socket()) await manager.force_close() await manager.handle_esp32_message("POS:888:888", connection) self.assertEqual((manager.last_pos_x, manager.last_pos_y), (1, 2)) self.assertFalse(manager.esp32_connected) asyncio.run(check()) def test_13_failed_commit_rolls_back_before_next_write(self): from app.device_store import DeviceStore store = DeviceStore(str(self.directory / "rollback.sqlite3")) store.add_device(A, "rollback fixture") before = store.get_settings(A) connection = store._conn class FailedCommit: attempted = False def __getattr__(self, name): return getattr(connection, name) def commit(self): self.attempted = True raise sqlite3.OperationalError("simulated full disk") def __enter__(self): return self def __exit__(self, kind, error, traceback): if kind is not None: connection.rollback() return False try: self.commit() except Exception: connection.rollback() raise try: failure = FailedCommit() store._conn = failure with self.assertRaises(Exception): store.save_settings(A, az_dev=77) self.assertTrue(failure.attempted, "fixture did not reach the failing commit") store._conn = connection self.assertFalse(connection.in_transaction, "failed write left a pending transaction") self.assertEqual(store.get_settings(A), before) store.save_settings(A, pyrano=0.4) self.assertEqual(store.get_settings(A)["az_dev"], before["az_dev"]) finally: store._conn = connection connection.rollback() store.close() def test_14_unknown_id_cannot_allocate_rate_limit_state(self): from app.main import fleet from app.config import is_valid_device_id self.assertFalse(is_valid_device_id(A + "\n")) before = len(fleet._buckets) self.assertFalse(fleet.check_rate("not-a-device")) self.assertFalse(fleet.check_rate("esp32-ffffffffffff")) self.assertEqual(len(fleet._buckets), before) def test_15_source_remains_compatible_with_python_311_image(self): for source in (ROOT / "app").glob("*.py"): with self.subTest(source=source.name): ast.parse(source.read_text(encoding="utf-8"), filename=str(source), feature_version=(3, 11)) def test_16_retiring_previous_token_closes_that_active_session_only(self): previous = self.tokens[A] with self.esp(A, previous) as active: self.cli("rotate", "--device-id", A, "--grace-hours", "1") current = (self.directory / "state" / "provision" / (A + ".v2.token")).read_text(encoding="utf-8").strip() # The old credential is still valid during the grace period. self.assertTrue(self.client.get(self.api(A, "/status")).json()["esp32_connected"]) self.cli("retire-previous", "--device-id", A) with self.assertRaises(WebSocketDisconnect): self.receive_text(active) with self.assertRaises(WebSocketDisconnect): with self.esp(A, previous): self.fail("retired credential accepted") self.tokens[A] = current with self.esp(A, current): pass def test_17_log_quota_is_visible_without_stopping_telemetry(self): from app.hub import HubManager from app.config import LOGS_DIR, get_local_now device = "esp32-001122aabbee" path = Path(LOGS_DIR) / device / f"log_{get_local_now():%Y-%m-%d}.txt" path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"sentinel\n") manager = HubManager(device) with patch("app.hub.LOG_DAILY_MAX_BYTES", 9): asyncio.run(manager.handle_esp32_message(SAMPLE)) self.assertEqual(path.read_bytes(), b"sentinel\n") self.assertEqual(len(manager.samples), 1) self.assertTrue(manager.status_payload()["log_degraded"]) self.assertEqual(manager.status_payload()["log_error_count"], 1) def test_18_rate_retention_and_slow_subscriber_are_bounded(self): from app.device_store import DeviceStore from app.fleet import FleetManager from app.hub import HubManager from app.config import LOGS_DIR store = DeviceStore(str(self.directory / "limits.sqlite3")) try: store.add_device(A) fleet = FleetManager(store, rate=0, burst=2) self.assertTrue(fleet.check_rate(A)) self.assertTrue(fleet.check_rate(A)) self.assertFalse(fleet.check_rate(A)) self.assertFalse(fleet.check_rate("esp32-ffffffffffff")) self.assertEqual(len(fleet._buckets), 1) old = Path(LOGS_DIR) / A / "log_2000-01-01.txt" legacy = Path(LOGS_DIR) / "log_2000-01-01.txt" for path in (old, legacy): path.parent.mkdir(parents=True, exist_ok=True) path.write_text("fixture", encoding="utf-8") os.utime(path, (1, 1)) self.assertEqual(fleet.prune_logs(30), 1) self.assertFalse(old.exists()) self.assertTrue(legacy.exists()) finally: store.close() class Subscriber: def __init__(self, slow=False): self.slow, self.closed, self.received = slow, False, [] async def send_text(self, message): if self.slow: await asyncio.sleep(10) self.received.append(message) async def close(self, **kwargs): self.closed = True async def check(): manager = HubManager() slow, fast = Subscriber(True), Subscriber() manager.client_sockets.extend([slow, fast]) with patch("app.hub.BROADCAST_TIMEOUT", 0.01): await manager._broadcast("test") self.assertEqual(fast.received, ["test"]) self.assertTrue(slow.closed) self.assertEqual(manager.client_sockets, [fast]) asyncio.run(check()) def test_19_cross_origin_http_writes_are_rejected_before_mutation(self): before = self.client.get(self.api(A, "/calib")).json() rejected = self.client.post(self.api(A, "/calib"), json={"mV_per_WM2": 99}, headers={"Origin": "https://evil.invalid"}) self.assertEqual(rejected.status_code, 403) self.assertEqual(self.client.get(self.api(A, "/calib")).json(), before) manifest = self.client.get("/ota/manifest").json() upload = self.client.post("/ota/upload", files={"file": ("bad.bin", b"not-approved")}, headers={"Origin": "null"}) self.assertEqual(upload.status_code, 403) self.assertEqual(self.client.get("/ota/manifest").json(), manifest) accepted = self.client.post(self.api(A, "/calib"), json={"mV_per_WM2": before["mV_per_WM2"]}, headers={"Origin": ORIGIN}) self.assertEqual(accepted.status_code, 200) def test_20_auto_ota_pending_task_cancels_when_disabled_or_already_installed(self): from app.hub import HubManager class Socket: async def accept(self): pass async def send_text(self, text): raise AssertionError("Cancelled OTA must not send any command") async def check(): manager = HubManager(A) cid = await manager.connect_esp32(Socket()) manager.ota_manifest = {"version": "target", "sha256": "a" * 64, "url": "https://test.invalid/image"} manager.esp32_firmware_version = "old" manager._version_session_id = cid manager.ota_auto_enabled = True manager._maybe_auto_update() pending = manager._auto_task self.assertIsNotNone(pending) await manager.set_ota_auto(False) await asyncio.gather(pending, return_exceptions=True) self.assertTrue(pending.cancelled()) self.assertIsNone(manager._auto_task) manager.ota_auto_enabled = True manager._maybe_auto_update() pending = manager._auto_task manager.esp32_firmware_version = "target" manager._maybe_auto_update() await asyncio.gather(pending, return_exceptions=True) self.assertTrue(pending.cancelled()) self.assertIsNone(manager._auto_task) manager.disconnect_esp32(cid) asyncio.run(check()) def test_21_docker_healthcheck_targets_a_live_endpoint(self): dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8") path = re.search(r"http://127\.0\.0\.1:8000([^'\" ]+)", dockerfile).group(1) response = self.client.get(path) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["ok"]) def test_22_tracker_settings_reach_selected_device(self): with self.esp(A) as device, self.browser(A) as browser: self.until(browser, "CONFIG:") for command in ( "SET_PARAM:minErrMax=193", "SET_PARAM:lpfBeta=0.70", "SET_PARAM:smallThreshold=4500", "SET_PARAM:peakProximity=0.30", "SET_PARAM:cicleMax=2", "SET_PARAM:axisXFactor=3", "SET_PARAM:axisYFactor=1", ): with self.subTest(command=command): browser.send_text(command) browser.send_text("CMD:10") # The status sentinel also bounds a rejected setting without waiting for it. self.assertEqual(self.receive_text(device), command) self.assertEqual(self.receive_text(device), "CMD:10") def test_23_command_validation_covers_dashboard_settings_and_keeps_guards(self): from app.main import COMMAND_MAX_LEN, _valid_command html = (ROOT / "static" / "index.html").read_text(encoding="utf-8") parameters = set(re.findall(r"sendParam\('([^']+)'\s*,\s*'[^']+'\)", html)) self.assertTrue(parameters, "dashboard parameter controls were not found") for name in parameters | {"errMin", "coarseStep", "fineStep", "angleThreshold"}: with self.subTest(parameter=name): self.assertTrue(_valid_command(f"SET_PARAM:{name}=1")) for value in ("1", "3", "8"): self.assertTrue(_valid_command("SET_PARAM:axisXFactor=" + value)) for value in ("0", "9", "-1", "1.5", "nan", "inf"): self.assertFalse(_valid_command("SET_PARAM:axisXFactor=" + value)) for value in ("1", "4", "64"): self.assertTrue(_valid_command("SET_PARAM:cicleMax=" + value)) for value in ("0", "-1", "1.5", "65", "999999999999"): self.assertFalse(_valid_command("SET_PARAM:cicleMax=" + value)) html_controls = ("axisXFactor", "axisYFactor", "axisRequested", "device_id", "x_budget") for text in html_controls: self.assertIn(text, html) prefix = "SET_PARAM:minErrMax=" at_limit = prefix + "0" * (COMMAND_MAX_LEN - len(prefix)) self.assertTrue(_valid_command(at_limit)) for command in ( at_limit + "0", "SET_PARAM:unknown=1", "SET_PARAM:minerrmax=193", "SET_PARAM:minErrMax=", "SET_PARAM:minErrMax=abc", "SET_PARAM:lpfBeta=nan", "SET_PARAM:lpfBeta=inf", "SET_PARAM:lpfBeta=1.2.3", "SET_PARAM:lpfBeta=0,70", "SET_PARAM:lpfBeta=0.70\nCMD:4", "SET_PARAM:lpfBeta=0.70\x00", "SET_PARAM:lpfBeta=1e309", "OTA:https://evil.invalid/payload.bin", ): with self.subTest(command=command): self.assertFalse(_valid_command(command)) def test_24_axis_confirmation_is_device_scoped_and_reports_effective_budgets(self): from app.hub import HubManager async def check(): sent_by_device = {} managers = {} for device in (A, B): manager = HubManager(device_id=device) managers[device] = manager sent = sent_by_device[device] = [] async def capture(message, sent=sent): sent.append(message) manager._broadcast = capture await manager._process_line("LOG:[2026-11-04 12:00:00] AXISCFG:3:1:4") ack = json.loads(sent[0][len("AXISCFG:"):]) self.assertEqual(ack, {"x": 3, "y": 1, "base": 4, "device_id": device, "x_budget": 12, "y_budget": 4}) for invalid in ( "junk AXISCFG:3:1:4", "AXISCFG:3:1:4 junk", "LOG:[bad timestamp] AXISCFG:3:1:4", "LOG:[2026-11-04 12:00:00] prefix AXISCFG:3:1:4", "AXISCFG:8:1:9", "AXISCFG:3:1:2147483647", "AXISCFG:3:-1:4", ): before = len(sent_by_device[A]) await managers[A]._process_line(invalid) self.assertFalse(any(m.startswith("AXISCFG:") for m in sent_by_device[A][before:])) asyncio.run(check()) if __name__ == "__main__": unittest.main(verbosity=2)