Peter van Arkel
fe8e65e12e
Move backup execution out of the management command into a reusable backup runner service that can execute an existing BackupRun record. Add queue primitives and a run_pobsync_worker command so manual backup requests can be recorded as queued SQL state and processed outside the web request path. Add a worker Docker service and pobsync worker CLI alias, with tests for queued run creation, worker execution, manual run typing, and command mapping.
56 lines
2.2 KiB
Python
56 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from io import StringIO
|
|
from unittest.mock import patch
|
|
|
|
from django.test import SimpleTestCase
|
|
|
|
from pobsync.cli import main
|
|
|
|
|
|
class ConsoleEntrypointTests(SimpleTestCase):
|
|
def test_maps_backup_alias_to_django_command(self) -> None:
|
|
with patch("pobsync.cli.execute_from_command_line") as execute:
|
|
exit_code = main(["backup", "web-01", "--dry-run"])
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
execute.assert_called_once_with(["pobsync", "run_pobsync_backup", "web-01", "--dry-run"])
|
|
|
|
def test_unknown_command_returns_usage_error(self) -> None:
|
|
stderr = StringIO()
|
|
with patch("sys.stderr", stderr):
|
|
exit_code = main(["run-scheduled", "web-01"])
|
|
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertIn("Unknown pobsync command", stderr.getvalue())
|
|
|
|
def test_django_passthrough_keeps_management_command_name(self) -> None:
|
|
with patch("pobsync.cli.execute_from_command_line") as execute:
|
|
exit_code = main(["django", "check"])
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
execute.assert_called_once_with(["pobsync", "check"])
|
|
|
|
def test_maps_schedule_alias_to_django_command(self) -> None:
|
|
with patch("pobsync.cli.execute_from_command_line") as execute:
|
|
exit_code = main(["schedule", "web-01", "--cron", "15 2 * * *"])
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
execute.assert_called_once_with(
|
|
["pobsync", "configure_pobsync_schedule", "web-01", "--cron", "15 2 * * *"]
|
|
)
|
|
|
|
def test_maps_discover_snapshots_alias_to_django_command(self) -> None:
|
|
with patch("pobsync.cli.execute_from_command_line") as execute:
|
|
exit_code = main(["discover-snapshots", "--host", "web-01"])
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
execute.assert_called_once_with(["pobsync", "discover_pobsync_snapshots", "--host", "web-01"])
|
|
|
|
def test_maps_worker_alias_to_django_command(self) -> None:
|
|
with patch("pobsync.cli.execute_from_command_line") as execute:
|
|
exit_code = main(["worker", "--once"])
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
execute.assert_called_once_with(["pobsync", "run_pobsync_worker", "--once"])
|