Peter van Arkel
994f7f66c4
Update the scheduler to reflect the actual scheduled BackupRun status after a run completes, so prune warnings are shown as schedule warnings instead of being reported as successful schedule executions.
94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from django.test import SimpleTestCase, TestCase
|
|
|
|
from pobsync_backend.management.commands.run_pobsync_scheduler import Command
|
|
from pobsync_backend.models import BackupRun, HostConfig, ScheduleConfig
|
|
from pobsync_backend.scheduler import due_key, is_due, next_due_after
|
|
|
|
|
|
class SchedulerTests(SimpleTestCase):
|
|
def test_daily_time_is_due_only_on_matching_minute(self) -> None:
|
|
moment = datetime(2026, 5, 19, 2, 15, tzinfo=ZoneInfo("UTC"))
|
|
|
|
self.assertTrue(is_due("15 2 * * *", moment))
|
|
self.assertFalse(is_due("16 2 * * *", moment))
|
|
|
|
def test_step_values_are_supported(self) -> None:
|
|
moment = datetime(2026, 5, 19, 2, 30, tzinfo=ZoneInfo("UTC"))
|
|
|
|
self.assertTrue(is_due("*/15 * * * *", moment))
|
|
self.assertFalse(is_due("*/20 * * * *", moment))
|
|
|
|
def test_sunday_allows_zero_and_seven(self) -> None:
|
|
sunday = datetime(2026, 5, 24, 2, 0, tzinfo=ZoneInfo("UTC"))
|
|
|
|
self.assertTrue(is_due("0 2 * * 0", sunday))
|
|
self.assertTrue(is_due("0 2 * * 7", sunday))
|
|
|
|
def test_due_key_has_minute_granularity(self) -> None:
|
|
moment = datetime(2026, 5, 19, 2, 15, 45, tzinfo=ZoneInfo("UTC"))
|
|
|
|
self.assertEqual(due_key(moment), "202605190215")
|
|
|
|
def test_next_due_after_returns_next_matching_minute(self) -> None:
|
|
moment = datetime(2026, 5, 19, 2, 15, 45, tzinfo=ZoneInfo("UTC"))
|
|
|
|
self.assertEqual(next_due_after("30 2 * * *", moment), datetime(2026, 5, 19, 2, 30, tzinfo=ZoneInfo("UTC")))
|
|
self.assertEqual(next_due_after("15 2 * * *", moment), datetime(2026, 5, 20, 2, 15, tzinfo=ZoneInfo("UTC")))
|
|
|
|
|
|
class SchedulerCommandTests(TestCase):
|
|
def test_run_due_executes_schedule_once_per_minute(self) -> None:
|
|
host = HostConfig.objects.create(
|
|
host="web-01",
|
|
address="web-01.example.test",
|
|
config={
|
|
"retention": {"daily": 7, "weekly": 4, "monthly": 3, "yearly": 1},
|
|
},
|
|
)
|
|
ScheduleConfig.objects.create(host=host, cron_expr="* * * * *")
|
|
|
|
command = Command()
|
|
with patch("pobsync_backend.management.commands.run_pobsync_scheduler.call_command") as call:
|
|
first_count = command._run_due(prefix=Path("/opt/pobsync"), dry_run=True)
|
|
second_count = command._run_due(prefix=Path("/opt/pobsync"), dry_run=True)
|
|
|
|
self.assertEqual(first_count, 1)
|
|
self.assertEqual(second_count, 0)
|
|
self.assertEqual(call.call_count, 1)
|
|
schedule = ScheduleConfig.objects.get(host=host)
|
|
self.assertEqual(schedule.last_status, "success")
|
|
|
|
def test_run_due_records_warning_status_from_scheduled_backup_run(self) -> None:
|
|
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
|
ScheduleConfig.objects.create(host=host, cron_expr="* * * * *", prune=True, prune_max_delete=1)
|
|
|
|
def create_warning_run(*args, **kwargs) -> None:
|
|
BackupRun.objects.create(
|
|
host=host,
|
|
run_type=BackupRun.RunType.SCHEDULED,
|
|
status=BackupRun.Status.WARNING,
|
|
result={
|
|
"ok": True,
|
|
"prune": {
|
|
"ok": False,
|
|
"type": "ConfigError",
|
|
"error": "Refusing to delete 2 snapshots (exceeds --max-delete=1)",
|
|
},
|
|
},
|
|
)
|
|
|
|
command = Command()
|
|
with patch("pobsync_backend.management.commands.run_pobsync_scheduler.call_command", side_effect=create_warning_run):
|
|
count = command._run_due(prefix=Path("/opt/pobsync"), dry_run=False)
|
|
|
|
self.assertEqual(count, 1)
|
|
schedule = ScheduleConfig.objects.get(host=host)
|
|
self.assertEqual(schedule.last_status, "warning")
|