Files
archery/test/test_charging_shutdown.py
T
2026-08-11 13:38:27 +08:00

145 lines
4.5 KiB
Python

import importlib.util
from pathlib import Path
import sys
import types
import unittest
from unittest import mock
class _StopMonitor(Exception):
pass
class _FakeTime:
now_ms = 0
stop_at_ms = None
@classmethod
def reset(cls, stop_at_ms=None):
cls.now_ms = 0
cls.stop_at_ms = stop_at_ms
@classmethod
def ticks_ms(cls):
return cls.now_ms
@classmethod
def sleep_ms(cls, milliseconds):
cls.now_ms += milliseconds
if cls.stop_at_ms is not None and cls.now_ms >= cls.stop_at_ms:
raise _StopMonitor()
def _load_power_module():
module_path = Path(__file__).resolve().parents[1] / "power.py"
module_name = "power_charging_shutdown_test"
maix_module = types.ModuleType("maix")
maix_module.time = _FakeTime
previous_maix = sys.modules.get("maix")
sys.modules["maix"] = maix_module
try:
spec = importlib.util.spec_from_file_location(module_name, module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
finally:
if previous_maix is None:
sys.modules.pop("maix", None)
else:
sys.modules["maix"] = previous_maix
power = _load_power_module()
class ChargingShutdownTests(unittest.TestCase):
def setUp(self):
self.config_patch = mock.patch.multiple(
power.config,
CHARGING_SHUTDOWN_ENABLED=True,
CHARGING_DIAGNOSTIC_LOG_ENABLED=False,
CHARGING_CHECK_INTERVAL_MS=5000,
CHARGING_CURRENT_THRESHOLD_MA=100.0,
CHARGING_CONFIRM_COUNT=2,
CHARGING_NOTIFY_TIMEOUT_MS=30000,
CHARGING_EXIT_SCRIPT="/tmp/charging_exit.sh",
)
self.config_patch.start()
self.network_manager = mock.Mock()
self.network_manager.safe_enqueue_and_wait.return_value = True
network_module = types.ModuleType("network")
network_module.network_manager = self.network_manager
self.network_module_patch = mock.patch.dict(
sys.modules,
{"network": network_module},
)
self.network_module_patch.start()
_FakeTime.reset()
def tearDown(self):
self.network_module_patch.stop()
self.config_patch.stop()
def test_two_charging_samples_notify_server_and_exit(self):
popen_calls = []
with (
mock.patch.object(power, "get_current", return_value=-200.0),
mock.patch.object(power.os.path, "isfile", return_value=True),
mock.patch.object(
power.subprocess,
"Popen",
side_effect=lambda args: popen_calls.append(args),
),
):
power.charging_shutdown_monitor()
self.assertEqual(_FakeTime.now_ms, 5000)
self.assertEqual(len(popen_calls), 1)
self.network_manager.safe_enqueue_and_wait.assert_called_once_with(
{"poweroff": "充电中"}, 2, high=True, timeout_ms=30000
)
def test_discharging_does_not_notify_or_exit(self):
_FakeTime.reset(stop_at_ms=10000)
popen_calls = []
with (
mock.patch.object(power, "get_current", return_value=200.0),
mock.patch.object(power.os.path, "isfile", return_value=True),
mock.patch.object(
power.subprocess,
"Popen",
side_effect=lambda args: popen_calls.append(args),
),
self.assertRaises(_StopMonitor),
):
power.charging_shutdown_monitor()
self.assertEqual(popen_calls, [])
self.network_manager.safe_enqueue_and_wait.assert_not_called()
def test_failed_sample_resets_confirmation_count(self):
popen_calls = []
currents = iter((-200.0, 0.0, -200.0, -200.0))
with (
mock.patch.object(power, "get_current", side_effect=lambda: next(currents)),
mock.patch.object(power.os.path, "isfile", return_value=True),
mock.patch.object(
power.subprocess,
"Popen",
side_effect=lambda args: popen_calls.append(args),
),
):
power.charging_shutdown_monitor()
self.assertEqual(_FakeTime.now_ms, 15000)
self.assertEqual(len(popen_calls), 1)
self.network_manager.safe_enqueue_and_wait.assert_called_once_with(
{"poweroff": "充电中"}, 2, high=True, timeout_ms=30000
)
if __name__ == "__main__":
unittest.main()