203 lines
3.4 KiB
Python
203 lines
3.4 KiB
Python
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from schedule import (
|
|
ScheduleService,
|
|
ScheduledSettings,
|
|
load_schedule,
|
|
)
|
|
|
|
|
|
SCHEDULE_YAML = """
|
|
version: 1
|
|
|
|
enabled: true
|
|
timezone: local
|
|
|
|
templates:
|
|
|
|
test:
|
|
|
|
- time: "00: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 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
|
|
|
|
|
|
async def test_paused_schedule_does_not_control_tion():
|
|
|
|
temp_dir, config = (
|
|
load_test_schedule()
|
|
)
|
|
|
|
try:
|
|
tion = FakeTionService()
|
|
|
|
service = ScheduleService(
|
|
config,
|
|
tion=tion,
|
|
)
|
|
|
|
assert service.paused is False
|
|
|
|
await service.pause()
|
|
|
|
assert service.paused is True
|
|
|
|
# Даже если фоновый цикл вызывает _process(),
|
|
# никаких команд быть не должно.
|
|
await service._process()
|
|
|
|
assert tion.controller.calls == []
|
|
|
|
# Возвращаем расписание.
|
|
#
|
|
# resume() должен сразу применить
|
|
# текущее состояние расписания.
|
|
await service.resume()
|
|
|
|
assert service.paused is False
|
|
|
|
assert (
|
|
"set_speed",
|
|
2,
|
|
) in tion.controller.calls
|
|
|
|
finally:
|
|
temp_dir.cleanup()
|
|
|
|
|
|
async def test_pause_clears_override():
|
|
|
|
temp_dir, config = (
|
|
load_test_schedule()
|
|
)
|
|
|
|
try:
|
|
service = ScheduleService(
|
|
config
|
|
)
|
|
|
|
# Имитируем уже существующий
|
|
# temporary override.
|
|
service._override_settings = (
|
|
ScheduledSettings(
|
|
speed=5,
|
|
)
|
|
)
|
|
|
|
service._override_until = (
|
|
datetime.now()
|
|
+ timedelta(hours=1)
|
|
)
|
|
|
|
service._override_pending = True
|
|
|
|
assert (
|
|
service.override_settings.speed
|
|
== 5
|
|
)
|
|
|
|
await service.pause()
|
|
|
|
assert service.paused is True
|
|
|
|
assert (
|
|
service.override_settings.speed
|
|
is None
|
|
)
|
|
|
|
assert (
|
|
service.override_until
|
|
is None
|
|
)
|
|
|
|
assert (
|
|
service._override_pending
|
|
is False
|
|
)
|
|
|
|
finally:
|
|
temp_dir.cleanup()
|
|
|
|
|
|
async def main():
|
|
|
|
await test_paused_schedule_does_not_control_tion()
|
|
|
|
await test_pause_clears_override()
|
|
|
|
print()
|
|
print("=" * 70)
|
|
print(
|
|
"ALL SCHEDULE PAUSE TESTS PASSED"
|
|
)
|
|
print("=" * 70)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |