140 lines
3.8 KiB
Python
140 lines
3.8 KiB
Python
import json
|
|
import sys
|
|
import types
|
|
import unittest
|
|
|
|
|
|
class _FakeTime:
|
|
@staticmethod
|
|
def sleep(_seconds):
|
|
pass
|
|
|
|
@staticmethod
|
|
def sleep_ms(_milliseconds):
|
|
pass
|
|
|
|
@staticmethod
|
|
def ticks_ms():
|
|
return 0
|
|
|
|
@staticmethod
|
|
def ticks_diff(left, right):
|
|
return left - right
|
|
|
|
|
|
class _FakeLogger:
|
|
def debug(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def info(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def warning(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
def error(self, *_args, **_kwargs):
|
|
pass
|
|
|
|
|
|
class _FakeSocket:
|
|
def __init__(self, recv_data=b""):
|
|
self.recv_data = recv_data
|
|
self.closed = False
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
def recv(self, _size, *_flags):
|
|
return self.recv_data
|
|
|
|
|
|
class _StopAfterCallback:
|
|
def __init__(self):
|
|
self.stopped = False
|
|
|
|
def is_set(self):
|
|
return self.stopped
|
|
|
|
|
|
maix_module = types.ModuleType("maix")
|
|
maix_module.time = _FakeTime
|
|
maix_module.network = types.SimpleNamespace()
|
|
maix_module.err = types.SimpleNamespace()
|
|
sys.modules.setdefault("maix", maix_module)
|
|
sys.modules.setdefault("ujson", json)
|
|
|
|
netcore_module = types.ModuleType("archery_netcore")
|
|
netcore_module.get_config = lambda: {"SERVER_IP": "127.0.0.1", "SERVER_PORT": 1234}
|
|
netcore_module.parse_packet = lambda _packet: (0, {})
|
|
netcore_module.make_packet = lambda *_args, **_kwargs: b""
|
|
netcore_module.actions_for_inner_cmd = lambda *_args, **_kwargs: []
|
|
sys.modules["archery_netcore"] = netcore_module
|
|
|
|
hardware_module = types.ModuleType("hardware")
|
|
hardware_module.hardware_manager = types.SimpleNamespace()
|
|
sys.modules["hardware"] = hardware_module
|
|
|
|
power_module = types.ModuleType("power")
|
|
power_module.get_bus_voltage = lambda: 0
|
|
power_module.voltage_to_percent = lambda _voltage: 0
|
|
sys.modules["power"] = power_module
|
|
|
|
import logger_manager
|
|
import wifi
|
|
import network
|
|
|
|
|
|
class WiFiFailoverTests(unittest.TestCase):
|
|
def setUp(self):
|
|
logger_manager.logger_manager._logger = _FakeLogger()
|
|
|
|
def test_monitor_switches_when_sta_association_is_lost(self):
|
|
manager = wifi.wifi_manager
|
|
stop_event = _StopAfterCallback()
|
|
callbacks = []
|
|
|
|
manager._wifi_socket = _FakeSocket()
|
|
manager._wifi_quality_stop_event = stop_event
|
|
manager._network_type_callback = lambda: "wifi"
|
|
manager.is_sta_associated = lambda: False
|
|
manager._get_wifi_rssi_dbm = lambda: None
|
|
|
|
def on_poor_quality():
|
|
callbacks.append(True)
|
|
stop_event.stopped = True
|
|
|
|
manager._on_poor_quality_callback = on_poor_quality
|
|
manager._quality_monitor_loop()
|
|
|
|
self.assertEqual(callbacks, [True])
|
|
self.assertIsNone(manager.last_wifi_rtt_ms)
|
|
|
|
def test_tls_connection_check_rejects_lost_sta_association(self):
|
|
manager = network.network_manager
|
|
sock = _FakeSocket()
|
|
wifi.wifi_manager._wifi_socket = sock
|
|
wifi.wifi_manager._wifi_connected = True
|
|
wifi.wifi_manager._wifi_ip = "192.168.1.2"
|
|
wifi.wifi_manager.is_sta_associated = lambda: False
|
|
manager._tcp_connected = True
|
|
|
|
self.assertFalse(manager._check_wifi_connection())
|
|
self.assertTrue(sock.closed)
|
|
self.assertIsNone(wifi.wifi_manager.wifi_socket)
|
|
self.assertFalse(manager.tcp_connected)
|
|
|
|
def test_receive_eof_marks_wifi_tcp_disconnected(self):
|
|
manager = network.network_manager
|
|
sock = _FakeSocket(recv_data=b"")
|
|
wifi.wifi_manager._wifi_socket = sock
|
|
manager._tcp_connected = True
|
|
|
|
self.assertEqual(manager.receive_tcp_data_via_wifi(), b"")
|
|
self.assertTrue(sock.closed)
|
|
self.assertIsNone(wifi.wifi_manager.wifi_socket)
|
|
self.assertFalse(manager.tcp_connected)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|