432 lines
7.7 KiB
Python
432 lines
7.7 KiB
Python
import asyncio
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from unittest.mock import patch
|
|
|
|
from schedule import (
|
|
ScheduleActionType,
|
|
ScheduleService,
|
|
load_schedule,
|
|
)
|
|
|
|
|
|
SCHEDULE_YAML = """
|
|
version: 1
|
|
|
|
enabled: true
|
|
timezone: local
|
|
|
|
templates:
|
|
|
|
test:
|
|
|
|
- time: "00:00"
|
|
action:
|
|
type: set
|
|
power: off
|
|
|
|
- time: "09:00"
|
|
action:
|
|
type: set
|
|
power: on
|
|
speed: 1
|
|
|
|
- time: "10:00"
|
|
action:
|
|
type: auto
|
|
speed: 3
|
|
|
|
- time: "12:00"
|
|
action:
|
|
type: set
|
|
speed: 2
|
|
|
|
days:
|
|
mon: test
|
|
tue: test
|
|
wed: test
|
|
thu: test
|
|
fri: test
|
|
sat: test
|
|
sun: test
|
|
"""
|
|
|
|
|
|
class FakeTionController:
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def power_on(self):
|
|
self.calls.append(
|
|
("power_on", None)
|
|
)
|
|
|
|
async def power_off(self):
|
|
self.calls.append(
|
|
("power_off", None)
|
|
)
|
|
|
|
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
|
|
)
|
|
|
|
|
|
def load_test_schedule():
|
|
temp_dir = TemporaryDirectory()
|
|
|
|
path = (
|
|
Path(temp_dir.name)
|
|
/ "schedule.yaml"
|
|
)
|
|
|
|
path.write_text(
|
|
SCHEDULE_YAML,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
config = load_schedule(path)
|
|
|
|
return temp_dir, config
|
|
|
|
|
|
def print_resolution(
|
|
title,
|
|
resolution,
|
|
):
|
|
print()
|
|
print("=" * 70)
|
|
print(title)
|
|
print("=" * 70)
|
|
|
|
print(
|
|
"Current:",
|
|
resolution.current.when
|
|
if resolution.current
|
|
else None,
|
|
)
|
|
|
|
print(
|
|
"Action:",
|
|
resolution.current.point.action.type
|
|
if resolution.current
|
|
else None,
|
|
)
|
|
|
|
print(
|
|
"Scheduled speed:",
|
|
resolution.scheduled_settings.speed,
|
|
)
|
|
|
|
print(
|
|
"AUTO:",
|
|
resolution.auto_active,
|
|
)
|
|
|
|
print(
|
|
"Fallback:",
|
|
resolution.auto_fallback_speed,
|
|
)
|
|
|
|
|
|
def test_resolve(service: ScheduleService):
|
|
|
|
# --------------------------------------------------
|
|
# 09:30
|
|
# Обычный SET speed=1
|
|
# --------------------------------------------------
|
|
|
|
resolution = service.resolve(
|
|
datetime(
|
|
2026,
|
|
9,
|
|
14,
|
|
9,
|
|
30,
|
|
)
|
|
)
|
|
|
|
print_resolution(
|
|
"TEST 1 — SET",
|
|
resolution,
|
|
)
|
|
|
|
assert resolution.auto_active is False
|
|
|
|
assert (
|
|
resolution.auto_fallback_speed
|
|
is None
|
|
)
|
|
|
|
assert (
|
|
resolution.scheduled_settings.speed
|
|
== 1
|
|
)
|
|
|
|
# --------------------------------------------------
|
|
# 10:30
|
|
# AUTO speed=3
|
|
#
|
|
# scheduled_settings.speed всё ещё 1,
|
|
# потому что AUTO не участвует
|
|
# в накоплении SET-настроек.
|
|
#
|
|
# Но fallback должен быть 3.
|
|
# --------------------------------------------------
|
|
|
|
resolution = service.resolve(
|
|
datetime(
|
|
2026,
|
|
9,
|
|
14,
|
|
10,
|
|
30,
|
|
)
|
|
)
|
|
|
|
print_resolution(
|
|
"TEST 2 — AUTO",
|
|
resolution,
|
|
)
|
|
|
|
assert (
|
|
resolution.current.point.action.type
|
|
== ScheduleActionType.AUTO
|
|
)
|
|
|
|
assert resolution.auto_active is True
|
|
|
|
assert (
|
|
resolution.auto_fallback_speed
|
|
== 3
|
|
)
|
|
|
|
assert (
|
|
resolution.scheduled_settings.speed
|
|
== 1
|
|
)
|
|
|
|
# --------------------------------------------------
|
|
# 12:30
|
|
# AUTO закончился.
|
|
# SET speed=2.
|
|
# --------------------------------------------------
|
|
|
|
resolution = service.resolve(
|
|
datetime(
|
|
2026,
|
|
9,
|
|
14,
|
|
12,
|
|
30,
|
|
)
|
|
)
|
|
|
|
print_resolution(
|
|
"TEST 3 — AFTER AUTO",
|
|
resolution,
|
|
)
|
|
|
|
assert resolution.auto_active is False
|
|
|
|
assert (
|
|
resolution.auto_fallback_speed
|
|
is None
|
|
)
|
|
|
|
assert (
|
|
resolution.scheduled_settings.speed
|
|
== 2
|
|
)
|
|
|
|
|
|
async def test_process(
|
|
service: ScheduleService,
|
|
tion: FakeTionService,
|
|
):
|
|
|
|
# --------------------------------------------------
|
|
# 09:30
|
|
#
|
|
# Обычный SET.
|
|
# ScheduleService должен применить speed=1.
|
|
# --------------------------------------------------
|
|
|
|
fixed_now = datetime(
|
|
2026,
|
|
9,
|
|
14,
|
|
9,
|
|
30,
|
|
)
|
|
|
|
with patch(
|
|
"schedule.service.datetime",
|
|
wraps=datetime,
|
|
) as mocked_datetime:
|
|
|
|
mocked_datetime.now.return_value = (
|
|
fixed_now
|
|
)
|
|
|
|
await service._process()
|
|
|
|
print()
|
|
print(
|
|
"09:30 calls:",
|
|
tion.controller.calls,
|
|
)
|
|
|
|
assert (
|
|
"set_speed",
|
|
1,
|
|
) in tion.controller.calls
|
|
|
|
# Очищаем историю команд.
|
|
tion.controller.calls.clear()
|
|
|
|
# --------------------------------------------------
|
|
# 10:30
|
|
#
|
|
# AUTO.
|
|
#
|
|
# В расписании fallback speed=3,
|
|
# но ScheduleService НЕ должен
|
|
# отправить ни speed=1, ни speed=3.
|
|
#
|
|
# Скорость теперь принадлежит
|
|
# будущему AutoController.
|
|
# --------------------------------------------------
|
|
|
|
fixed_now = datetime(
|
|
2026,
|
|
9,
|
|
14,
|
|
10,
|
|
30,
|
|
)
|
|
|
|
with patch(
|
|
"schedule.service.datetime",
|
|
wraps=datetime,
|
|
) as mocked_datetime:
|
|
|
|
mocked_datetime.now.return_value = (
|
|
fixed_now
|
|
)
|
|
|
|
await service._process()
|
|
|
|
print()
|
|
print(
|
|
"10:30 AUTO calls:",
|
|
tion.controller.calls,
|
|
)
|
|
|
|
speed_calls = [
|
|
call
|
|
for call in tion.controller.calls
|
|
if call[0] == "set_speed"
|
|
]
|
|
|
|
assert speed_calls == []
|
|
|
|
# Очищаем историю.
|
|
tion.controller.calls.clear()
|
|
|
|
# --------------------------------------------------
|
|
# 12:30
|
|
#
|
|
# AUTO закончился.
|
|
# Расписание снова должно поставить speed=2.
|
|
# --------------------------------------------------
|
|
|
|
fixed_now = datetime(
|
|
2026,
|
|
9,
|
|
14,
|
|
12,
|
|
30,
|
|
)
|
|
|
|
with patch(
|
|
"schedule.service.datetime",
|
|
wraps=datetime,
|
|
) as mocked_datetime:
|
|
|
|
mocked_datetime.now.return_value = (
|
|
fixed_now
|
|
)
|
|
|
|
await service._process()
|
|
|
|
print()
|
|
print(
|
|
"12:30 calls:",
|
|
tion.controller.calls,
|
|
)
|
|
|
|
assert (
|
|
"set_speed",
|
|
2,
|
|
) in tion.controller.calls
|
|
|
|
|
|
async def main():
|
|
|
|
temp_dir, config = (
|
|
load_test_schedule()
|
|
)
|
|
|
|
try:
|
|
# ----------------------------------------------
|
|
# Проверка resolve()
|
|
# ----------------------------------------------
|
|
|
|
service = ScheduleService(
|
|
config
|
|
)
|
|
|
|
test_resolve(service)
|
|
|
|
# ----------------------------------------------
|
|
# Проверка реального _process(),
|
|
# но через FakeTion.
|
|
# ----------------------------------------------
|
|
|
|
tion = FakeTionService()
|
|
|
|
service = ScheduleService(
|
|
config,
|
|
tion=tion,
|
|
)
|
|
|
|
await test_process(
|
|
service,
|
|
tion,
|
|
)
|
|
|
|
finally:
|
|
temp_dir.cleanup()
|
|
|
|
print()
|
|
print("=" * 70)
|
|
print("ALL AUTO SCHEDULE TESTS PASSED")
|
|
print("=" * 70)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |