"""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()