# server/test_ota_endpoint.py import base64 import hashlib import json from pathlib import Path import pytest from server.ota_endpoint import ota_check_impl, ota_firmware_impl def write_firmware(firmware_dir: Path, version: str, data: bytes = b"fake_fw") -> None: sig = bytes(64) # zero sig (not validated server-side) manifest = { "version": version, "size": len(data), "sha256": hashlib.sha256(data).hexdigest(), } (firmware_dir / "current.bin").write_bytes(data) (firmware_dir / "current.sig").write_bytes(sig) (firmware_dir / "manifest.json").write_text(json.dumps(manifest)) @pytest.fixture(autouse=True) def patch_firmware_dir(tmp_path, monkeypatch): import server.ota_endpoint as mod monkeypatch.setattr(mod, "FIRMWARE_DIR", tmp_path) yield tmp_path def test_check_no_update_same_version(tmp_path): write_firmware(tmp_path, "1.0.0") result = ota_check_impl(current_version="1.0.0", firmware_dir=tmp_path) assert result["update"] is False def test_check_no_update_newer_local(tmp_path): write_firmware(tmp_path, "1.0.0") result = ota_check_impl(current_version="1.1.0", firmware_dir=tmp_path) assert result["update"] is False def test_check_update_available(tmp_path): write_firmware(tmp_path, "1.1.0", data=b"new firmware") result = ota_check_impl(current_version="1.0.0", firmware_dir=tmp_path) assert result["update"] is True assert result["version"] == "1.1.0" assert result["size"] == len(b"new firmware") assert "sha256" in result assert "sig_b64" in result sig_bytes = base64.b64decode(result["sig_b64"]) assert len(sig_bytes) == 64 def test_check_no_manifest(tmp_path): result = ota_check_impl(current_version="1.0.0", firmware_dir=tmp_path) assert result["update"] is False def test_firmware_endpoint_returns_binary(tmp_path): fw_data = b"firmware binary content" write_firmware(tmp_path, "1.1.0", data=fw_data) content = ota_firmware_impl(firmware_dir=tmp_path) assert content == fw_data def test_firmware_endpoint_missing_raises(tmp_path): import server.ota_endpoint as mod with pytest.raises(mod.FirmwareNotFoundError): ota_firmware_impl(firmware_dir=tmp_path) def test_check_malformed_manifest(tmp_path): (tmp_path / "manifest.json").write_text("not valid json{{{") result = ota_check_impl(current_version="1.0.0", firmware_dir=tmp_path) assert result["update"] is False def test_check_wrong_arity_version_no_update(tmp_path): write_firmware(tmp_path, "1.2") # wrong arity server version result = ota_check_impl(current_version="1.0.0", firmware_dir=tmp_path) # server "1.2" → (0,0,0) ≤ client (1,0,0) → no update assert result["update"] is False