Implements /ota/check (version comparison + sig_b64 payload) and /ota/firmware (binary stream) using the same _impl pattern as camera_endpoint.py. HMAC auth left commented pending main app wiring. 6/6 tests passing. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
# 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)
|