248 lines
9.7 KiB
Python
248 lines
9.7 KiB
Python
"""Flashing: image validation on the real image in firmware/, the folder
|
|
catalogue (compile-date order), plan building, and the runner against a fake
|
|
esptool (no hardware)."""
|
|
|
|
import hashlib
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from kota_tech.flash.image import ImageError, app_image_length, folder_images, inspect_image
|
|
from kota_tech.flash.plan import Mode, build_plan, check_partition_table
|
|
from kota_tech.flash.runner import FlashRunner
|
|
|
|
FW_DIR = Path(__file__).resolve().parents[1] / "firmware"
|
|
FW = sorted(FW_DIR.glob("*.bin"))[0]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def image():
|
|
imgs, problems = folder_images()
|
|
assert imgs and problems == []
|
|
return imgs[0]
|
|
|
|
|
|
# ------------------------------------------------------------------ image
|
|
|
|
def test_folder_image_is_a_kota_build(image):
|
|
assert image.app.project == "kota-sensorboard"
|
|
assert image.kind == "merged"
|
|
assert image.app.built_at is not None # compile date is readable
|
|
assert image.flash_size == "8MB"
|
|
assert image.partition("ota_0").offset == 0x40000
|
|
assert image.partition("otadata").offset == 0x31000
|
|
assert image.partition("nvs").offset == 0x10000
|
|
|
|
|
|
def test_label_starts_with_compile_date(image):
|
|
label = image.label()
|
|
assert label.startswith(f"Compiled {image.app.built_at:%Y-%m-%d %H:%M}")
|
|
assert image.app.version in label and image.name in label
|
|
|
|
|
|
def test_app_bytes_are_a_complete_app_image(image):
|
|
app = image.app_bytes()
|
|
assert app[0] == 0xE9
|
|
assert len(app) == len(image.data) - 0x40000 # ends where the file ends
|
|
assert len(image.blank_otadata()) == 0x2000 and set(image.blank_otadata()) == {0xFF}
|
|
|
|
|
|
@pytest.mark.parametrize("patch,msg", [
|
|
((0, 0x00), "not an ESP32 firmware"),
|
|
((12, 0x05), "not ESP32-S3"),
|
|
])
|
|
def test_wrong_images_are_rejected(tmp_path, patch, msg):
|
|
data = bytearray(FW.read_bytes())
|
|
data[patch[0]] = patch[1]
|
|
p = tmp_path / "x.bin"
|
|
p.write_bytes(data)
|
|
with pytest.raises(ImageError, match=msg):
|
|
inspect_image(p)
|
|
|
|
|
|
def test_integrity_check_on_folder_image(image):
|
|
d = image.data
|
|
assert app_image_length(d, 0) > 0 # bootloader
|
|
assert app_image_length(d, 0x40000) == len(d) - 0x40000 # app ends exactly at file end
|
|
|
|
|
|
@pytest.mark.parametrize("damage,msg", [
|
|
(lambda b: b[:len(b) // 2], "truncated"),
|
|
(lambda b: b[:0x40000 + 5000] + bytes([b[0x40000 + 5000] ^ 1]) + b[0x40000 + 5001:], "corrupted"),
|
|
])
|
|
def test_corrupted_or_truncated_images_refused(tmp_path, damage, msg):
|
|
p = tmp_path / "broken.bin"
|
|
p.write_bytes(damage(FW.read_bytes()))
|
|
with pytest.raises(ImageError, match=msg):
|
|
inspect_image(p)
|
|
|
|
|
|
# ------------------------------------------------------- firmware folder
|
|
|
|
def _with_compile_date(data: bytes, date: str, time: str, off: int = 0x40000) -> bytes:
|
|
"""Copy of a merged image whose app claims another compile date, with the
|
|
segment checksum and appended SHA-256 recomputed (a valid, different build)."""
|
|
d = bytearray(data)
|
|
desc = off + 0x20
|
|
old = bytes(d[desc + 80:desc + 112])
|
|
new = time.encode().ljust(16, b"\0") + date.encode().ljust(16, b"\0")
|
|
d[desc + 80:desc + 112] = new
|
|
end = off + app_image_length(data, off) - 32 # SHA-256 follows the checksum byte
|
|
xor = 0
|
|
for a, b in zip(old, new):
|
|
xor ^= a ^ b
|
|
d[end - 1] ^= xor
|
|
d[end:end + 32] = hashlib.sha256(bytes(d[off:end])).digest()
|
|
return bytes(d)
|
|
|
|
|
|
def test_folder_lists_builds_newest_compile_date_first(tmp_path):
|
|
data = FW.read_bytes()
|
|
(tmp_path / "a_old.bin").write_bytes(_with_compile_date(data, "Jan 5 2026", "08:00:00"))
|
|
(tmp_path / "z_new.bin").write_bytes(_with_compile_date(data, "Dec 24 2026", "18:30:00"))
|
|
(tmp_path / "m_mid.bin").write_bytes(_with_compile_date(data, "Jun 15 2026", "12:00:00"))
|
|
imgs, problems = folder_images(tmp_path)
|
|
assert problems == []
|
|
assert [i.name for i in imgs] == ["z_new.bin", "m_mid.bin", "a_old.bin"]
|
|
assert imgs[0].label().startswith("Compiled 2026-12-24 18:30")
|
|
assert imgs[2].label().startswith("Compiled 2026-01-05 08:00")
|
|
|
|
|
|
def test_folder_reports_bad_files_and_duplicates(tmp_path):
|
|
data = FW.read_bytes()
|
|
(tmp_path / "good.bin").write_bytes(data)
|
|
(tmp_path / "copy.bin").write_bytes(data)
|
|
(tmp_path / "notes.bin").write_bytes(b"hello firmware team")
|
|
(tmp_path / "readme.txt").write_text("ignored: not a .bin")
|
|
imgs, problems = folder_images(tmp_path)
|
|
assert len(imgs) == 1
|
|
assert any("notes.bin" in p and "not an ESP32 firmware" in p for p in problems)
|
|
assert any("same file as" in p for p in problems)
|
|
assert not any("readme" in p for p in problems)
|
|
|
|
|
|
def test_app_only_file_uses_the_merged_layout(tmp_path, image):
|
|
(tmp_path / "merged.bin").write_bytes(image.data)
|
|
(tmp_path / "kota-sensorboard.bin").write_bytes(
|
|
_with_compile_date(image.data, "Dec 1 2026", "10:00:00")[0x40000:])
|
|
imgs, problems = folder_images(tmp_path)
|
|
assert problems == []
|
|
app = next(i for i in imgs if i.kind == "app")
|
|
assert imgs[0] is app # newer compile date first
|
|
assert app.partition("ota_0").offset == 0x40000 and "app only" in app.label()
|
|
assert [s.kind for s in build_plan(app, Mode.APP_ONLY, "COM5", tmp_path / "w").steps] == ["read_pt", "write"]
|
|
with pytest.raises(ValueError, match="app only"):
|
|
build_plan(app, Mode.FULL, "COM5", tmp_path / "w")
|
|
|
|
|
|
def test_app_only_file_without_merged_image_is_not_offered(tmp_path, image):
|
|
(tmp_path / "kota-sensorboard.bin").write_bytes(image.app_bytes())
|
|
imgs, problems = folder_images(tmp_path)
|
|
assert imgs == [] and "no complete (merged) image" in problems[0]
|
|
|
|
|
|
def test_empty_or_missing_folder(tmp_path):
|
|
assert folder_images(tmp_path) == ([], [])
|
|
imgs, problems = folder_images(tmp_path / "nope")
|
|
assert imgs == [] and "not found" in problems[0]
|
|
|
|
|
|
# ------------------------------------------------------------------- plan
|
|
|
|
def test_app_only_plan_keeps_nvs(image, tmp_path):
|
|
plan = build_plan(image, Mode.APP_ONLY, "COM5", tmp_path)
|
|
kinds = [s.kind for s in plan.steps]
|
|
assert kinds == ["read_pt", "write"]
|
|
write = plan.steps[-1].args
|
|
offsets = [a for a in write if a.startswith("0x")]
|
|
assert offsets == ["0x31000", "0x40000"] # otadata + ota_0 only
|
|
assert "0x10000" not in write and "0x0" not in write # NVS / bootloader untouched
|
|
assert write[write.index("--after") + 1] == "hard_reset"
|
|
assert (tmp_path / "app_ota0.bin").read_bytes() == image.app_bytes()
|
|
|
|
|
|
def test_full_plan_with_backup(image, tmp_path):
|
|
backup = tmp_path / "bk.bin"
|
|
plan = build_plan(image, Mode.FULL, "COM5", tmp_path, backup_to=backup)
|
|
assert [s.kind for s in plan.steps] == ["backup", "write"]
|
|
assert plan.steps[0].args[-3:] == ["0", "0x800000", str(backup)]
|
|
assert plan.steps[0].args[plan.steps[0].args.index("--after") + 1] == "no_reset"
|
|
assert plan.steps[1].args[-2:] == ["0x0", str(image.path)]
|
|
assert "Back up" in plan.summary()[0]
|
|
|
|
|
|
def test_partition_table_check(image):
|
|
assert check_partition_table(image.partition_table, image) is None
|
|
assert "blank" in check_partition_table(b"\xff" * 0xC00, image)
|
|
other = bytearray(image.partition_table)
|
|
other[40] ^= 1
|
|
assert "different partition layout" in check_partition_table(bytes(other), image)
|
|
|
|
|
|
# ----------------------------------------------------------------- runner
|
|
|
|
FAKE = """
|
|
import sys, pathlib
|
|
args = sys.argv[1:]
|
|
if "read_flash" in args:
|
|
out = pathlib.Path(args[-1])
|
|
src = pathlib.Path(r"{fw}").read_bytes()
|
|
off, size = int(args[-3], 0), int(args[-2], 0)
|
|
{pt_line}
|
|
print("Read ... (50 %)", end="\\r"); print("Read ... (100 %)")
|
|
sys.exit(0)
|
|
if "write_flash" in args:
|
|
for p in (10, 60, 100):
|
|
print(f"Writing at 0x00040000... ({{p}} %)", end="\\r", flush=True)
|
|
print(); print("Hash of data verified.")
|
|
sys.exit({write_code})
|
|
"""
|
|
|
|
|
|
def _fake(tmp_path, write_code=0, pt="same", fail_text=""):
|
|
pt_line = {"same": "out.write_bytes(src[off:off+size])",
|
|
"blank": "out.write_bytes(b'\\xff' * size)"}[pt]
|
|
script = FAKE.format(fw=FW, pt_line=pt_line, write_code=write_code)
|
|
if fail_text:
|
|
script = script.replace('print("Hash of data verified.")', f'print("{fail_text}")')
|
|
f = tmp_path / "fake_esptool.py"
|
|
f.write_text(script)
|
|
return [sys.executable, str(f)]
|
|
|
|
|
|
def _run(plan, cmd):
|
|
steps, progress, output, done = [], [], [], threading.Event()
|
|
result = {}
|
|
r = FlashRunner(plan, lambda i, t: steps.append(i), progress.append, output.append,
|
|
lambda res: (result.setdefault("r", res), done.set()), esptool_cmd=cmd)
|
|
r.start()
|
|
assert done.wait(30)
|
|
return result["r"], steps, progress, output
|
|
|
|
|
|
def test_runner_app_only_success(image, tmp_path):
|
|
plan = build_plan(image, Mode.APP_ONLY, "COM5", tmp_path)
|
|
res, steps, progress, output = _run(plan, _fake(tmp_path))
|
|
assert res.ok, res
|
|
assert steps == [0, 1]
|
|
assert 60 in progress and progress[-1] == 100
|
|
assert any("Hash of data verified" in o for o in output)
|
|
|
|
|
|
def test_runner_refuses_app_only_on_blank_chip(image, tmp_path):
|
|
plan = build_plan(image, Mode.APP_ONLY, "COM5", tmp_path)
|
|
res, steps, *_ = _run(plan, _fake(tmp_path, pt="blank"))
|
|
assert not res.ok and "Full mode" in res.message
|
|
assert steps == [0] # never reached the write step
|
|
|
|
|
|
def test_runner_reports_connect_failure_with_hint(image, tmp_path):
|
|
plan = build_plan(image, Mode.FULL, "COM5", tmp_path)
|
|
cmd = _fake(tmp_path, write_code=2,
|
|
fail_text="A fatal error occurred: Failed to connect to ESP32-S3: No serial data received.")
|
|
res, *_ = _run(plan, cmd)
|
|
assert not res.ok and res.failed_step == 0
|
|
assert "BOOT" in res.hint
|