884 lines
16 KiB
Python
884 lines
16 KiB
Python
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()) |