Danil Iarosh
add8aff0af
- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth. - Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку. - В терминал команд добавлена ручная кнопка авторизации. - Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском. - Проверка обновлений учитывает только более новые версии по схеме major, minor и patch. - Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from collections import deque
|
|
from unittest.mock import patch
|
|
|
|
from app.services.auth_service import AuthStatus
|
|
from app.services.firmware_service import perform_authenticated_ota
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self, responses: list[str | None]) -> None:
|
|
self.is_connected = True
|
|
self._responses = deque(responses)
|
|
self.events: list[str] = []
|
|
self.release_count = 0
|
|
self.firmware_version: str | None = None
|
|
self.auth_required: bool | None = None
|
|
|
|
def read_response(self, timeout: float | None = None) -> str | None:
|
|
response = self._responses.popleft() if self._responses else None
|
|
self.events.append(f"read:{response}")
|
|
return response
|
|
|
|
def send_command(self, command: str) -> None:
|
|
self.events.append(f"send:{command}")
|
|
|
|
def write_raw(self, command: str) -> None:
|
|
self.events.append(f"write:{command}")
|
|
|
|
def release(self) -> None:
|
|
self.release_count += 1
|
|
|
|
def set_firmware_version(self, version: str, *, auth_required: bool) -> None:
|
|
self.firmware_version = version
|
|
self.auth_required = auth_required
|
|
|
|
|
|
class AuthenticatedOtaTests(unittest.TestCase):
|
|
_HOST_CHALLENGE = bytes.fromhex("00112233445566778899AABBCCDDEEFF")
|
|
_CHAL = "CHAL00112233445566778899AABBCCDDEEFF"
|
|
_RAES = "RAES40E9A18C22C2CFCD764666C90B358F9D"
|
|
|
|
def _run(self, conn: FakeConnection):
|
|
with patch(
|
|
"app.services.auth_service.os.urandom",
|
|
return_value=self._HOST_CHALLENGE,
|
|
):
|
|
return perform_authenticated_ota(
|
|
conn,
|
|
ota_timeout=0.1,
|
|
banner_timeout=0.1,
|
|
startup_delay=0.0,
|
|
handshake_timeout=0.1,
|
|
ready_timeout=0.1,
|
|
sleep_fn=lambda _seconds: None,
|
|
)
|
|
|
|
def test_sends_04_before_authentication_after_successful_update(self) -> None:
|
|
conn = FakeConnection([
|
|
"Writing 100%",
|
|
"Successful update",
|
|
"periphery controller ver_2.4.0 release",
|
|
self._CHAL,
|
|
self._RAES,
|
|
"READY!",
|
|
])
|
|
|
|
result = self._run(conn)
|
|
|
|
self.assertTrue(result.success)
|
|
self.assertEqual(AuthStatus.OK, result.post_auth.status)
|
|
send_events = [event for event in conn.events if event.startswith("send:")]
|
|
self.assertEqual("send:04", send_events[0])
|
|
self.assertTrue(send_events[1].startswith("send:07"))
|
|
|
|
def test_sends_04_without_pre_authentication(self) -> None:
|
|
conn = FakeConnection([None])
|
|
|
|
result = self._run(conn)
|
|
|
|
self.assertFalse(result.success)
|
|
self.assertEqual("send:04", conn.events[0])
|
|
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
|
|
|
|
def test_no_auth_firmware_skips_post_ota_handshake(self) -> None:
|
|
conn = FakeConnection([
|
|
"Successful update",
|
|
"periphery controller ver_2.0.0_no_auth",
|
|
])
|
|
|
|
result = self._run(conn)
|
|
|
|
self.assertTrue(result.success)
|
|
self.assertEqual(AuthStatus.OK, result.post_auth.status)
|
|
self.assertTrue(result.post_auth.auth_skipped)
|
|
self.assertFalse(conn.auth_required)
|
|
self.assertFalse(any(event.startswith("send:07") for event in conn.events))
|
|
self.assertIn("без авторизации", result.detail)
|
|
|
|
def test_reports_post_authentication_failure_after_update(self) -> None:
|
|
conn = FakeConnection([
|
|
"Successful update",
|
|
None,
|
|
])
|
|
|
|
result = self._run(conn)
|
|
|
|
self.assertFalse(result.success)
|
|
self.assertEqual(AuthStatus.TIMEOUT, result.post_auth.status)
|
|
self.assertIn("после OTA", result.detail)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|