179 lines
3.2 KiB
Python
179 lines
3.2 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from schedule import (
|
|
ScheduleService,
|
|
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
|
|
"""
|
|
|
|
|
|
def create_config(
|
|
directory: Path,
|
|
):
|
|
|
|
schedule_path = (
|
|
directory
|
|
/ "schedule.yaml"
|
|
)
|
|
|
|
schedule_path.write_text(
|
|
SCHEDULE_YAML,
|
|
encoding="utf-8",
|
|
)
|
|
|
|
return load_schedule(
|
|
schedule_path
|
|
)
|
|
|
|
|
|
async def test_pause_survives_restart():
|
|
|
|
with TemporaryDirectory() as temp_dir:
|
|
|
|
directory = Path(temp_dir)
|
|
|
|
config = create_config(
|
|
directory
|
|
)
|
|
|
|
state_path = (
|
|
directory
|
|
/ "schedule_state.json"
|
|
)
|
|
|
|
# --------------------------------------
|
|
# Первый экземпляр сервиса.
|
|
# --------------------------------------
|
|
|
|
service1 = ScheduleService(
|
|
config,
|
|
state_path=state_path,
|
|
)
|
|
|
|
assert service1.paused is False
|
|
|
|
await service1.pause()
|
|
|
|
assert service1.paused is True
|
|
assert state_path.exists()
|
|
|
|
# --------------------------------------
|
|
# Имитируем перезапуск приложения:
|
|
# создаём новый ScheduleService.
|
|
# --------------------------------------
|
|
|
|
service2 = ScheduleService(
|
|
config,
|
|
state_path=state_path,
|
|
)
|
|
|
|
assert service2.paused is True
|
|
|
|
|
|
async def test_resume_survives_restart():
|
|
|
|
with TemporaryDirectory() as temp_dir:
|
|
|
|
directory = Path(temp_dir)
|
|
|
|
config = create_config(
|
|
directory
|
|
)
|
|
|
|
state_path = (
|
|
directory
|
|
/ "schedule_state.json"
|
|
)
|
|
|
|
service1 = ScheduleService(
|
|
config,
|
|
state_path=state_path,
|
|
)
|
|
|
|
await service1.pause()
|
|
|
|
assert service1.paused is True
|
|
|
|
await service1.resume()
|
|
|
|
assert service1.paused is False
|
|
|
|
# Новый экземпляр после resume
|
|
# тоже должен быть активным.
|
|
service2 = ScheduleService(
|
|
config,
|
|
state_path=state_path,
|
|
)
|
|
|
|
assert service2.paused is False
|
|
|
|
|
|
async def test_missing_state_file_defaults_to_active():
|
|
|
|
with TemporaryDirectory() as temp_dir:
|
|
|
|
directory = Path(temp_dir)
|
|
|
|
config = create_config(
|
|
directory
|
|
)
|
|
|
|
state_path = (
|
|
directory
|
|
/ "missing_state.json"
|
|
)
|
|
|
|
assert not state_path.exists()
|
|
|
|
service = ScheduleService(
|
|
config,
|
|
state_path=state_path,
|
|
)
|
|
|
|
assert service.paused is False
|
|
|
|
|
|
async def main():
|
|
|
|
await test_pause_survives_restart()
|
|
await test_resume_survives_restart()
|
|
await test_missing_state_file_defaults_to_active()
|
|
|
|
print()
|
|
print("=" * 70)
|
|
print(
|
|
"ALL SCHEDULE PAUSE "
|
|
"PERSISTENCE TESTS PASSED"
|
|
)
|
|
print("=" * 70)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |