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