import asyncio from types import SimpleNamespace from app.auto.co2_policy import Co2SpeedPolicy from app.auto.controller import AutoController # ============================================================ # Fake ScheduleService # ============================================================ class FakeScheduleService: def __init__( self, *, enabled: bool = True, auto_active: bool = False, fallback_speed: int | None = None, ): self.enabled = enabled self.auto_active = auto_active self.fallback_speed = fallback_speed self.override_active = False def resolve(self, now): return SimpleNamespace( enabled=self.enabled, auto_active=self.auto_active, auto_fallback_speed=self.fallback_speed, ) # ============================================================ # Fake QingpingService # ============================================================ class FakeQingpingService: def __init__( self, *, online: bool = False, co2: int | None = None, ): self.online = online self.state = SimpleNamespace( co2=co2, ) # ============================================================ # Fake Tion # ============================================================ class FakeTionController: def __init__(self): self.calls = [] async def set_speed( self, speed: int, ): self.calls.append( ("set_speed", speed) ) class FakeTionService: def __init__(self): self.controller = FakeTionController() async def execute( self, operation, ): return await operation( self.controller ) # ============================================================ # Helper # ============================================================ def print_state( title: str, auto: AutoController, tion: FakeTionService, ): print() print("=" * 70) print(title) print("=" * 70) print( "Auto status:", auto.status(), ) print( "Tion calls:", tion.controller.calls, ) # ============================================================ # Main test # ============================================================ async def main(): schedule = FakeScheduleService() qingping = FakeQingpingService() tion = FakeTionService() policy = Co2SpeedPolicy( base_speed=1, thresholds=( (800, 2), (1000, 3), (1300, 4), (1600, 5), (2000, 6), ), hysteresis=100, ) auto = AutoController( schedule_service=schedule, qingping_service=qingping, tion_service=tion, policy=policy, ) # ======================================================== # TEST 1 # # AUTO не активен. # # AutoController не должен ничего делать. # ======================================================== schedule.enabled = True schedule.auto_active = False schedule.fallback_speed = None schedule.override_active = False qingping.online = True qingping.state.co2 = 1200 await auto._process() print_state( "TEST 1 — AUTO INACTIVE", auto, tion, ) assert ( auto.status()["state"] == "inactive" ) assert ( auto.status()["reason"] is None ) assert ( auto.status()["target_speed"] is None ) assert ( auto.status()["auto_speed"] is None ) assert tion.controller.calls == [] # ======================================================== # TEST 2 # # AUTO активен. # Qingping работает. # CO2 = 700. # # Ожидаем speed 1. # ======================================================== schedule.auto_active = True schedule.fallback_speed = 2 qingping.online = True qingping.state.co2 = 700 await auto._process() print_state( "TEST 2 — CO2 700", auto, tion, ) assert ( auto.status()["state"] == "active" ) assert ( auto.status()["reason"] is None ) assert ( auto.status()["auto_speed"] == 1 ) assert ( auto.status()["target_speed"] == 1 ) assert tion.controller.calls == [ ("set_speed", 1), ] # ======================================================== # TEST 3 # # CO2 вырос до 850. # # Ожидаем переход: # # speed 1 -> speed 2 # ======================================================== qingping.state.co2 = 850 await auto._process() print_state( "TEST 3 — CO2 850", auto, tion, ) assert ( auto.status()["auto_speed"] == 2 ) assert ( auto.status()["target_speed"] == 2 ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ] # ======================================================== # TEST 4 # # CO2 вырос до 1050. # # Ожидаем: # # speed 2 -> speed 3 # ======================================================== qingping.state.co2 = 1050 await auto._process() print_state( "TEST 4 — CO2 1050", auto, tion, ) assert ( auto.status()["auto_speed"] == 3 ) assert ( auto.status()["target_speed"] == 3 ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ] # ======================================================== # TEST 5 # # CO2 опустился до 950. # # Порог speed 3: # # вверх = 1000 # вниз = 900 # # Поэтому остаёмся на speed 3. # ======================================================== qingping.state.co2 = 950 await auto._process() print_state( "TEST 5 — HYSTERESIS CO2 950", auto, tion, ) assert ( auto.status()["auto_speed"] == 3 ) assert ( auto.status()["target_speed"] == 3 ) # Новой команды быть не должно. assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ] # ======================================================== # TEST 6 # # CO2 дошёл до 900. # # Теперь переходим: # # speed 3 -> speed 2 # ======================================================== qingping.state.co2 = 900 await auto._process() print_state( "TEST 6 — CO2 900", auto, tion, ) assert ( auto.status()["auto_speed"] == 2 ) assert ( auto.status()["target_speed"] == 2 ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ("set_speed", 2), ] # ======================================================== # TEST 7 # # Qingping offline. # # AUTO должен перейти в fallback. # # fallback speed = 2 # # Но Tion уже находится на speed 2, # поэтому повторная команда не нужна. # ======================================================== qingping.online = False await auto._process() print_state( "TEST 7 — QINGPING OFFLINE", auto, tion, ) assert ( auto.status()["state"] == "fallback" ) assert ( auto.status()["reason"] == "qingping_offline" ) assert ( auto.status()["auto_speed"] is None ) assert ( auto.status()["target_speed"] == 2 ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ("set_speed", 2), ] # ======================================================== # TEST 8 # # Qingping всё ещё offline. # # Одинаковую fallback-команду повторять нельзя. # ======================================================== await auto._process() print_state( "TEST 8 — FALLBACK NOT REPEATED", auto, tion, ) assert ( auto.status()["state"] == "fallback" ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ("set_speed", 2), ] # ======================================================== # TEST 9 # # Qingping online, # но CO2 отсутствует. # # Это тоже fallback. # ======================================================== qingping.online = True qingping.state.co2 = None await auto._process() print_state( "TEST 9 — CO2 MISSING", auto, tion, ) assert ( auto.status()["state"] == "fallback" ) assert ( auto.status()["reason"] == "co2_missing" ) assert ( auto.status()["target_speed"] == 2 ) assert ( auto.status()["auto_speed"] is None ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ("set_speed", 2), ] # ======================================================== # TEST 10 # # Qingping восстановился. # # CO2 = 1700. # # После fallback auto_speed был сброшен, # поэтому скорость выбирается заново. # # Ожидаем speed 5. # ======================================================== qingping.online = True qingping.state.co2 = 1700 await auto._process() print_state( "TEST 10 — RECOVERY CO2 1700", auto, tion, ) assert ( auto.status()["state"] == "active" ) assert ( auto.status()["reason"] is None ) assert ( auto.status()["auto_speed"] == 5 ) assert ( auto.status()["target_speed"] == 5 ) assert tion.controller.calls == [ ("set_speed", 1), ("set_speed", 2), ("set_speed", 3), ("set_speed", 2), ("set_speed", 5), ] # ======================================================== # TEST 11 # # Проверяем максимальную скорость 6. # # CO2 = 2200 # ======================================================== qingping.state.co2 = 2200 await auto._process() print_state( "TEST 11 — CO2 2200", auto, tion, ) assert ( auto.status()["auto_speed"] == 6 ) assert ( auto.status()["target_speed"] == 6 ) assert tion.controller.calls[-1] == ( "set_speed", 6, ) # ======================================================== # TEST 12 # # Manual override. # # Пользователь управляет Tion вручную. # # AUTO должен полностью отойти в сторону. # ======================================================== schedule.override_active = True await auto._process() print_state( "TEST 12 — MANUAL OVERRIDE", auto, tion, ) assert ( auto.status()["state"] == "suspended" ) assert ( auto.status()["reason"] == "manual_override" ) assert ( auto.status()["target_speed"] is None ) assert ( auto.status()["auto_speed"] is None ) # Никаких новых команд. assert tion.controller.calls[-1] == ( "set_speed", 6, ) # ======================================================== # TEST 13 # # Manual override закончился. # # AUTO должен заново выбрать скорость # по текущему CO2. # # CO2 остаётся 2200 -> speed 6. # ======================================================== schedule.override_active = False await auto._process() print_state( "TEST 13 — OVERRIDE FINISHED", auto, tion, ) assert ( auto.status()["state"] == "active" ) assert ( auto.status()["auto_speed"] == 6 ) assert ( auto.status()["target_speed"] == 6 ) # target_speed был сброшен во время override, # поэтому команда должна быть отправлена заново. assert tion.controller.calls[-1] == ( "set_speed", 6, ) # ======================================================== # TEST 14 # # Qingping снова падает. # # fallback = 2. # ======================================================== qingping.online = False await auto._process() print_state( "TEST 14 — SECOND FAILURE", auto, tion, ) assert ( auto.status()["state"] == "fallback" ) assert ( auto.status()["target_speed"] == 2 ) assert ( auto.status()["auto_speed"] is None ) assert tion.controller.calls[-1] == ( "set_speed", 2, ) # ======================================================== # TEST 15 # # Началась другая AUTO-точка расписания. # # Новый fallback = 3. # # Qingping по-прежнему offline. # # Ожидаем смену fallback: # # 2 -> 3 # ======================================================== schedule.fallback_speed = 3 await auto._process() print_state( "TEST 15 — FALLBACK CHANGED", auto, tion, ) assert ( auto.status()["state"] == "fallback" ) assert ( auto.status()["target_speed"] == 3 ) assert tion.controller.calls[-1] == ( "set_speed", 3, ) # ======================================================== # TEST 16 # # AUTO закончился. # # AutoController перестаёт управлять Tion. # ======================================================== schedule.auto_active = False schedule.fallback_speed = None await auto._process() print_state( "TEST 16 — AUTO FINISHED", auto, tion, ) assert ( auto.status()["state"] == "inactive" ) assert ( auto.status()["reason"] is None ) assert ( auto.status()["target_speed"] is None ) assert ( auto.status()["auto_speed"] is None ) # ======================================================== # TEST 17 # # CO2 policy отсутствует. # # Например, auto.yaml повреждён. # AUTO обязан использовать fallback. # ======================================================== schedule2 = FakeScheduleService( enabled=True, auto_active=True, fallback_speed=4, ) qingping2 = FakeQingpingService( online=True, co2=2500, ) tion2 = FakeTionService() auto2 = AutoController( schedule_service=schedule2, qingping_service=qingping2, tion_service=tion2, policy=None, ) await auto2._process() print_state( "TEST 17 — POLICY UNAVAILABLE", auto2, tion2, ) assert ( auto2.status()["state"] == "fallback" ) assert ( auto2.status()["reason"] == "auto_policy_unavailable" ) assert ( auto2.status()["target_speed"] == 4 ) assert ( auto2.status()["auto_speed"] is None ) assert tion2.controller.calls == [ ("set_speed", 4), ] print() print("=" * 70) print( "ALL AUTO CONTROLLER TESTS PASSED" ) print("=" * 70) if __name__ == "__main__": asyncio.run(main())