Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
211 lines
11 KiB
Python
211 lines
11 KiB
Python
"""Offline tests for the GUI bridge; serial/device I/O is always mocked."""
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
TOOLS = Path(__file__).resolve().parents[1] / "tools"
|
|
sys.path.insert(0, str(TOOLS))
|
|
import flash_app_bridge as bridge
|
|
from test_device_flash import MAC, SimulatedEsptool, make_build
|
|
|
|
|
|
class BridgeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix=".bridge-test-", dir=Path(__file__).parent)
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name).resolve()
|
|
self.build = self.root / "build"
|
|
make_build(self.build)
|
|
self.key = "ab" * 32
|
|
self.password = "private-setup-password"
|
|
self.credential = self.root / "credential.json"
|
|
self.info = {"factory_id": "KP01-123456789ABC", "model": "ECHOEAR-S3", "key_version": 1,
|
|
"origin": "https://example.com", "device_key_hex": self.key}
|
|
self.credential.write_text(json.dumps(self.info), encoding="utf-8")
|
|
self.request = {"credential_path": str(self.credential), "build_dir": str(self.build),
|
|
"mode": "identity", "expected_mac": MAC, "port": "COM6",
|
|
"output_root": str(self.root / "output")}
|
|
self.events = []
|
|
for owner, name, replacement in [
|
|
(bridge.device_prepare, "private_directory", lambda path: path.mkdir(parents=True, mode=0o700)),
|
|
(bridge.device_flash, "private_directory", lambda path: path.mkdir(parents=True, mode=0o700)),
|
|
(bridge.device_prepare, "prepare", self.fake_prepare),
|
|
]:
|
|
item = patch.object(owner, name, side_effect=replacement)
|
|
item.start()
|
|
self.addCleanup(item.stop)
|
|
|
|
def fake_prepare(self, credential, output, expected_mac):
|
|
info = bridge.device_prepare.read_credential(credential)
|
|
output.mkdir()
|
|
data = b"\xab" * bridge.device_flash.FACTORY_SIZE
|
|
(output / "ble_factory.bin").write_bytes(data)
|
|
(output / "factory.csv").write_text("device_key," + self.key)
|
|
(output / "setup-label.json").write_text(json.dumps({"setup_password": self.password}))
|
|
manifest = {"schema_version": 1, "chip": "esp32s3", "flash_size": bridge.device_flash.FLASH_SIZE,
|
|
"device_id": info["device_id"], "model": info["model"], "expected_mac": expected_mac,
|
|
"image": {"file": "ble_factory.bin", "offset": bridge.device_flash.FACTORY_OFFSET,
|
|
"size": len(data), "sha256": bridge.device_flash.sha256(data)}}
|
|
(output / "manifest.json").write_text(json.dumps(manifest))
|
|
return manifest
|
|
|
|
def preview(self, **overrides):
|
|
bridge.preview({**self.request, **overrides}, self.events.append)
|
|
return self.events[-1]
|
|
|
|
def assert_blocked(self, request, code):
|
|
with patch.object(bridge.device_flash, "execute_plan") as execute:
|
|
with self.assertRaises(bridge.BridgeError) as raised:
|
|
bridge.flash(request, self.events.append)
|
|
self.assertEqual(raised.exception.code, code)
|
|
execute.assert_not_called()
|
|
|
|
def test_preview_is_offline_and_secret_free(self):
|
|
with patch.object(bridge.device_flash, "execute_plan") as execute, patch.object(bridge, "hidden_run") as run:
|
|
result = self.preview()
|
|
execute.assert_not_called()
|
|
run.assert_not_called()
|
|
self.assertEqual(result["event"], "preview")
|
|
summary = result["summary"]
|
|
self.assertEqual(summary["mode"], "identity")
|
|
self.assertIsNone(summary["firmware_version"])
|
|
self.assertEqual(summary["reference_firmware_version"], "1.7.0-controls.28")
|
|
self.assertEqual({image["name"] for image in summary["regions"]}, {"ble_factory", "reset-runtime"})
|
|
text = json.dumps(self.events)
|
|
self.assertNotIn(self.key, text)
|
|
self.assertNotIn(self.password, text)
|
|
session = bridge.load_session(Path(result["session_file"]))
|
|
self.assertEqual(session["port"], "COM6")
|
|
self.assertEqual(session["expected_mac"], MAC)
|
|
self.assertNotIn(self.key, json.dumps(session))
|
|
|
|
def test_full_preview_has_complete_regions_and_version(self):
|
|
result = self.preview(mode="full")
|
|
self.assertEqual(result["summary"]["firmware_version"], "1.7.0-controls.28")
|
|
self.assertEqual(len(result["summary"]["regions"]), 8)
|
|
|
|
def test_bad_build_refuses_to_create_factory(self):
|
|
(self.build / "app.bin").write_bytes(b"not firmware")
|
|
with self.assertRaises(bridge.BridgeError) as raised:
|
|
self.preview()
|
|
self.assertEqual(raised.exception.code, "invalid_build")
|
|
self.assertFalse((self.root / "output").exists())
|
|
|
|
def test_bad_credential_refuses_to_create_factory(self):
|
|
self.info["device_key_hex"] = "invalid key"
|
|
self.credential.write_text(json.dumps(self.info))
|
|
with self.assertRaises(bridge.BridgeError) as raised:
|
|
self.preview()
|
|
self.assertEqual(raised.exception.code, "invalid_credential")
|
|
self.assertFalse((self.root / "output").exists())
|
|
|
|
def test_flash_requires_literal_true_confirmation(self):
|
|
session = self.preview()["session_file"]
|
|
for value in (None, False, 1, "true"):
|
|
with self.subTest(value=value):
|
|
self.assert_blocked({"session_file": session, "execute": value}, "confirmation_required")
|
|
|
|
def test_port_and_mac_bound_to_preview(self):
|
|
session = self.preview()["session_file"]
|
|
for override in ({"port": "COM7"}, {"expected_mac": "11:22:33:44:55:66"}):
|
|
with self.subTest(override=override):
|
|
self.assert_blocked({"session_file": session, "execute": True, **override}, "port_mismatch")
|
|
|
|
def test_frozen_inputs_changes_block_before_hardware(self):
|
|
for filename in ("credential.json", "build/app.bin", "build/flasher_args.json", "factory/ble_factory.bin", "factory/setup-label.json"):
|
|
with self.subTest(filename=filename):
|
|
session_path = Path(self.preview()["session_file"])
|
|
path = session_path.parent / filename if filename.startswith("factory/") else self.root / filename
|
|
original = path.read_bytes()
|
|
try:
|
|
path.write_bytes(original + b"changed")
|
|
self.assert_blocked({"session_file": str(session_path), "execute": True}, "stale_preview")
|
|
finally:
|
|
path.write_bytes(original)
|
|
|
|
def test_expiry_blocks_before_hardware(self):
|
|
session = self.preview()["session_file"]
|
|
with patch.object(bridge.time, "time", return_value=bridge.time.time() + bridge.SESSION_TTL + 1):
|
|
self.assert_blocked({"session_file": session, "execute": True}, "stale_preview")
|
|
|
|
def test_modified_session_cannot_change_target(self):
|
|
session = Path(self.preview()["session_file"])
|
|
value = json.loads(session.read_text())
|
|
value["port"] = "COM42"
|
|
session.write_text(json.dumps(value))
|
|
self.assert_blocked({"session_file": str(session), "execute": True}, "stale_preview")
|
|
|
|
def test_cannot_supply_arbitrary_command_or_runner(self):
|
|
session = self.preview()["session_file"]
|
|
for field in ("command", "python", "baud", "factory_dir", "backup_dir"):
|
|
self.assert_blocked({"session_file": session, "execute": True, field: "evil"}, "invalid_request")
|
|
|
|
def test_success_runs_existing_backup_write_verify_and_consumes_session(self):
|
|
session = self.preview()["session_file"]
|
|
runner = SimulatedEsptool()
|
|
with patch.object(bridge, "hidden_run", side_effect=runner):
|
|
bridge.flash({"session_file": session, "execute": True}, self.events.append)
|
|
self.assertEqual(runner.commands, ["version", "read_mac", "flash_id", "get_security_info", "read_flash", "write_flash", "verify_flash"])
|
|
self.assertEqual(self.events[-1]["event"], "complete")
|
|
self.assertTrue((Path(session).parent / "backup" / "before-16mb.bin").is_file())
|
|
self.assertIn("backup", [item.get("stage") for item in self.events])
|
|
self.assertIn("writing", [item.get("stage") for item in self.events])
|
|
self.assertIn("verifying", [item.get("stage") for item in self.events])
|
|
self.assert_blocked({"session_file": session, "execute": True}, "stale_preview")
|
|
|
|
def test_device_mac_mismatch_never_writes(self):
|
|
session = self.preview()["session_file"]
|
|
runner = SimulatedEsptool(mac="11:22:33:44:55:66")
|
|
with patch.object(bridge, "hidden_run", side_effect=runner):
|
|
with self.assertRaises(bridge.BridgeError) as raised:
|
|
bridge.flash({"session_file": session, "execute": True}, self.events.append)
|
|
self.assertEqual(raised.exception.code, "flash_failed")
|
|
self.assertNotIn("write_flash", runner.commands)
|
|
|
|
def test_backup_failure_never_writes(self):
|
|
session = self.preview()["session_file"]
|
|
runner = SimulatedEsptool(short_backup=True)
|
|
with patch.object(bridge, "hidden_run", side_effect=runner):
|
|
with self.assertRaises(bridge.BridgeError):
|
|
bridge.flash({"session_file": session, "execute": True}, self.events.append)
|
|
self.assertNotIn("write_flash", runner.commands)
|
|
|
|
def test_main_outputs_json_lines_and_redacts_exception(self):
|
|
session = self.preview()["session_file"]
|
|
request_file = self.root / "request.json"
|
|
request_file.write_text(json.dumps({"session_file": session, "execute": True}))
|
|
output = io.StringIO()
|
|
with patch.object(bridge.device_flash, "execute_plan", side_effect=RuntimeError(self.key + self.password)):
|
|
code = bridge.main(["flash", "--request", str(request_file)], output=output)
|
|
self.assertEqual(code, 2)
|
|
rows = [json.loads(line) for line in output.getvalue().splitlines()]
|
|
self.assertEqual(rows[-1]["code"], "flash_failed")
|
|
self.assertNotIn(self.key, output.getvalue())
|
|
self.assertNotIn(self.password, output.getvalue())
|
|
|
|
def test_ports_enumeration_does_not_open_serial(self):
|
|
import serial
|
|
from serial.tools import list_ports
|
|
port = types.SimpleNamespace(device="COM6", description="USB", hwid="USB VID:PID", serial_number="SERIAL")
|
|
with patch.object(serial, "Serial") as open_serial, patch.object(list_ports, "comports", return_value=[port]):
|
|
bridge.enumerate_ports({}, self.events.append)
|
|
open_serial.assert_not_called()
|
|
self.assertEqual(self.events[-1]["ports"][0]["serial_number"], "SERIAL")
|
|
|
|
def test_windows_child_processes_are_hidden(self):
|
|
if bridge.os.name != "nt":
|
|
self.skipTest("Windows-only subprocess flag")
|
|
with patch.object(bridge.subprocess, "run") as run:
|
|
bridge.hidden_run(["example"], capture_output=True)
|
|
self.assertTrue(run.call_args.kwargs["creationflags"] & subprocess.CREATE_NO_WINDOW)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|