diff --git a/README.md b/README.md index 00bc560..5c46739 100644 --- a/README.md +++ b/README.md @@ -137,6 +137,7 @@ loaded before Django starts: ``` sudo -u pobsync pobsync-manage showmigrations pobsync_backend sudo -u pobsync pobsync-manage check +sudo -u pobsync pobsync-manage check_pobsync_install ``` The UI includes: @@ -198,6 +199,7 @@ Then check: ``` systemctl status pobsync-web pobsync-worker pobsync-scheduler sudo -u pobsync pobsync-manage check +sudo -u pobsync pobsync-manage check_pobsync_install ``` ## Development diff --git a/scripts/install-systemd b/scripts/install-systemd index 05cbc23..f309834 100755 --- a/scripts/install-systemd +++ b/scripts/install-systemd @@ -586,3 +586,4 @@ echo "Useful commands:" echo " systemctl status pobsync-web pobsync-worker pobsync-scheduler" echo " journalctl -u pobsync-worker -f" echo " sudo -u $SERVICE_USER pobsync-manage check" +echo " sudo -u $SERVICE_USER pobsync-manage check_pobsync_install" diff --git a/src/pobsync_backend/management/commands/check_pobsync_install.py b/src/pobsync_backend/management/commands/check_pobsync_install.py new file mode 100644 index 0000000..20156f2 --- /dev/null +++ b/src/pobsync_backend/management/commands/check_pobsync_install.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from django.core.management.base import BaseCommand, CommandError + +from pobsync_backend.self_check import collect_self_checks, summarize_self_checks + + +class Command(BaseCommand): + help = "Run pobsync runtime self checks for native installs and updates." + + def add_arguments(self, parser): + parser.add_argument( + "--fail-on-warning", + action="store_true", + help="Exit with an error when warnings are present.", + ) + + def handle(self, *args, **options): + checks = collect_self_checks() + summary = summarize_self_checks(checks) + + for check in checks: + line = f"[{check.status.upper()}] {check.name}: {check.message}" + if check.detail: + line = f"{line} ({check.detail})" + if check.status == "failed": + self.stderr.write(line) + elif check.status == "warning": + self.stderr.write(line) + else: + self.stdout.write(line) + + self.stdout.write( + "Summary: " + f"{summary['ok']} ok, " + f"{summary['warning']} warning(s), " + f"{summary['failed']} failed, " + f"{summary['skipped']} skipped" + ) + + if summary["failed"]: + raise CommandError("pobsync install self check failed.") + if options["fail_on_warning"] and summary["warning"]: + raise CommandError("pobsync install self check reported warnings.") diff --git a/src/pobsync_backend/tests/test_self_check.py b/src/pobsync_backend/tests/test_self_check.py index c985e2e..76dff6c 100644 --- a/src/pobsync_backend/tests/test_self_check.py +++ b/src/pobsync_backend/tests/test_self_check.py @@ -1,11 +1,14 @@ from __future__ import annotations import subprocess +from io import StringIO from unittest.mock import patch +from django.core.management import call_command +from django.core.management.base import CommandError from django.test import SimpleTestCase -from pobsync_backend.self_check import _systemd_checks +from pobsync_backend.self_check import SelfCheck, _systemd_checks class SystemdSelfCheckTests(SimpleTestCase): @@ -40,3 +43,44 @@ class SystemdSelfCheckTests(SimpleTestCase): journal_check = next(check for check in checks if check.name == "Journal access") self.assertEqual(journal_check.status, "failed") self.assertEqual(journal_check.message, "pobsync cannot read service logs.") + + +class CheckPobsyncInstallCommandTests(SimpleTestCase): + def test_command_prints_summary_for_successful_checks(self) -> None: + stdout = StringIO() + stderr = StringIO() + checks = [ + SelfCheck("Database connection", "ok", "django.db.backends.sqlite3"), + SelfCheck("Systemd services", "skipped", "systemd is not available in this runtime."), + ] + + with patch("pobsync_backend.management.commands.check_pobsync_install.collect_self_checks", return_value=checks): + call_command("check_pobsync_install", stdout=stdout, stderr=stderr) + + self.assertIn("[OK] Database connection", stdout.getvalue()) + self.assertIn("[SKIPPED] Systemd services", stdout.getvalue()) + self.assertIn("Summary: 1 ok, 0 warning(s), 0 failed, 1 skipped", stdout.getvalue()) + self.assertEqual(stderr.getvalue(), "") + + def test_command_fails_when_checks_fail(self) -> None: + stdout = StringIO() + stderr = StringIO() + checks = [ + SelfCheck("POBSYNC_BACKUP_ROOT", "failed", "/backups does not exist."), + ] + + with patch("pobsync_backend.management.commands.check_pobsync_install.collect_self_checks", return_value=checks): + with self.assertRaisesMessage(CommandError, "pobsync install self check failed."): + call_command("check_pobsync_install", stdout=stdout, stderr=stderr) + + self.assertIn("[FAILED] POBSYNC_BACKUP_ROOT", stderr.getvalue()) + self.assertIn("Summary: 0 ok, 0 warning(s), 1 failed, 0 skipped", stdout.getvalue()) + + def test_command_can_fail_on_warnings(self) -> None: + checks = [ + SelfCheck("Global config", "warning", "Default global config has not been created yet."), + ] + + with patch("pobsync_backend.management.commands.check_pobsync_install.collect_self_checks", return_value=checks): + with self.assertRaisesMessage(CommandError, "pobsync install self check reported warnings."): + call_command("check_pobsync_install", "--fail-on-warning", stdout=StringIO(), stderr=StringIO())