Add check_pobsync_install so native deployments can run the same runtime diagnostics from the terminal that are available in the Django Self Check view. The command prints every check with status, returns a failing exit code when install-critical checks fail, supports fail-on-warning for stricter automation, and is documented in the installer output and README update flow.
87 lines
3.7 KiB
Python
87 lines
3.7 KiB
Python
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 SelfCheck, _systemd_checks
|
|
|
|
|
|
class SystemdSelfCheckTests(SimpleTestCase):
|
|
def test_journal_permission_hint_is_reported_as_failure(self) -> None:
|
|
def which(binary: str) -> str | None:
|
|
if binary in {"systemctl", "journalctl"}:
|
|
return f"/usr/bin/{binary}"
|
|
return None
|
|
|
|
active_result = subprocess.CompletedProcess(
|
|
args=["systemctl"],
|
|
returncode=0,
|
|
stdout="active\n",
|
|
stderr="",
|
|
)
|
|
journal_result = subprocess.CompletedProcess(
|
|
args=["journalctl"],
|
|
returncode=0,
|
|
stdout="",
|
|
stderr="No journal files were opened due to insufficient permissions.",
|
|
)
|
|
|
|
with patch("pobsync_backend.self_check.Path.exists", return_value=True), patch(
|
|
"pobsync_backend.self_check.shutil.which",
|
|
side_effect=which,
|
|
), patch(
|
|
"pobsync_backend.self_check.subprocess.run",
|
|
side_effect=[active_result, active_result, active_result, journal_result],
|
|
):
|
|
checks = _systemd_checks()
|
|
|
|
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())
|