work: добавил режим ручного управления. Теперь можно отключать расписание!

tests/test_schedule_pause_persistence.py

@@ -0,0 +1,179 @@
+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())