675 lines
33 KiB
Python
675 lines
33 KiB
Python
"""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)
|