(feature) Add full native installer and self-check page
Expand the systemd installer so it can perform a complete native installation with sensible defaults: copy the checkout into the target app directory, create runtime directories, write the environment file, install dependencies, configure systemd units, and optionally configure nginx. Add a staff-only Django self-check page that verifies runtime settings, required binaries, writable paths, database connectivity, global config state, and systemd service status when available. Document installer overrides and expose the self-check from the main navigation.
This commit is contained in:
200
src/pobsync_backend/self_check.py
Normal file
200
src/pobsync_backend/self_check.py
Normal file
@@ -0,0 +1,200 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import connection
|
||||
|
||||
from .models import GlobalConfig
|
||||
|
||||
|
||||
CheckStatus = Literal["ok", "warning", "failed", "skipped"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SelfCheck:
|
||||
name: str
|
||||
status: CheckStatus
|
||||
message: str
|
||||
detail: str = ""
|
||||
|
||||
|
||||
def collect_self_checks() -> list[SelfCheck]:
|
||||
checks: list[SelfCheck] = []
|
||||
checks.extend(_django_checks())
|
||||
checks.extend(_path_checks())
|
||||
checks.extend(_binary_checks())
|
||||
checks.extend(_database_checks())
|
||||
checks.extend(_config_checks())
|
||||
checks.extend(_systemd_checks())
|
||||
return checks
|
||||
|
||||
|
||||
def summarize_self_checks(checks: list[SelfCheck]) -> dict[str, int]:
|
||||
return {
|
||||
"ok": sum(1 for check in checks if check.status == "ok"),
|
||||
"warning": sum(1 for check in checks if check.status == "warning"),
|
||||
"failed": sum(1 for check in checks if check.status == "failed"),
|
||||
"skipped": sum(1 for check in checks if check.status == "skipped"),
|
||||
}
|
||||
|
||||
|
||||
def _django_checks() -> list[SelfCheck]:
|
||||
checks = [
|
||||
SelfCheck(
|
||||
"Django debug",
|
||||
"warning" if settings.DEBUG else "ok",
|
||||
"DEBUG is enabled." if settings.DEBUG else "DEBUG is disabled.",
|
||||
),
|
||||
SelfCheck(
|
||||
"Django secret key",
|
||||
"failed" if settings.SECRET_KEY == "dev-only-change-me" else "ok",
|
||||
"Default development secret key is still active."
|
||||
if settings.SECRET_KEY == "dev-only-change-me"
|
||||
else "Secret key is configured.",
|
||||
),
|
||||
SelfCheck(
|
||||
"Allowed hosts",
|
||||
"ok" if settings.ALLOWED_HOSTS else "failed",
|
||||
", ".join(settings.ALLOWED_HOSTS) if settings.ALLOWED_HOSTS else "No allowed hosts configured.",
|
||||
),
|
||||
]
|
||||
return checks
|
||||
|
||||
|
||||
def _path_checks() -> list[SelfCheck]:
|
||||
checks = []
|
||||
checks.append(_path_check("POBSYNC_HOME", Path(settings.POBSYNC_HOME), must_be_absolute=True, must_be_writable=True))
|
||||
checks.append(
|
||||
_path_check(
|
||||
"POBSYNC_BACKUP_ROOT",
|
||||
Path(settings.POBSYNC_BACKUP_ROOT),
|
||||
must_be_absolute=True,
|
||||
must_exist=True,
|
||||
must_be_writable=True,
|
||||
)
|
||||
)
|
||||
checks.append(
|
||||
_path_check(
|
||||
"Static root",
|
||||
Path(settings.STATIC_ROOT),
|
||||
must_be_absolute=True,
|
||||
must_exist=False,
|
||||
must_be_writable=True,
|
||||
)
|
||||
)
|
||||
db_settings = settings.DATABASES["default"]
|
||||
if db_settings["ENGINE"] == "django.db.backends.sqlite3":
|
||||
checks.append(
|
||||
_path_check(
|
||||
"SQLite directory",
|
||||
Path(str(db_settings["NAME"])).parent,
|
||||
must_be_absolute=True,
|
||||
must_exist=True,
|
||||
must_be_writable=True,
|
||||
)
|
||||
)
|
||||
return checks
|
||||
|
||||
|
||||
def _path_check(
|
||||
name: str,
|
||||
path: Path,
|
||||
*,
|
||||
must_be_absolute: bool,
|
||||
must_exist: bool = False,
|
||||
must_be_writable: bool,
|
||||
) -> SelfCheck:
|
||||
if must_be_absolute and not path.is_absolute():
|
||||
return SelfCheck(name, "failed", f"{path} is not absolute.")
|
||||
if must_exist and not path.exists():
|
||||
return SelfCheck(name, "failed", f"{path} does not exist.")
|
||||
|
||||
target = path if path.exists() else path.parent
|
||||
if not target.exists():
|
||||
return SelfCheck(name, "failed", f"{target} does not exist.")
|
||||
if must_be_writable and not os.access(target, os.W_OK):
|
||||
return SelfCheck(name, "failed", f"{target} is not writable by this process.")
|
||||
return SelfCheck(name, "ok", str(path))
|
||||
|
||||
|
||||
def _binary_checks() -> list[SelfCheck]:
|
||||
checks = []
|
||||
for binary in ("rsync", "ssh", "ssh-keygen", "gunicorn"):
|
||||
path = shutil.which(binary)
|
||||
checks.append(
|
||||
SelfCheck(
|
||||
f"Binary: {binary}",
|
||||
"ok" if path else "failed",
|
||||
path or f"{binary} was not found in PATH.",
|
||||
)
|
||||
)
|
||||
return checks
|
||||
|
||||
|
||||
def _database_checks() -> list[SelfCheck]:
|
||||
try:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("SELECT 1")
|
||||
cursor.fetchone()
|
||||
except Exception as exc:
|
||||
return [SelfCheck("Database connection", "failed", f"{type(exc).__name__}: {exc}")]
|
||||
return [SelfCheck("Database connection", "ok", settings.DATABASES["default"]["ENGINE"])]
|
||||
|
||||
|
||||
def _config_checks() -> list[SelfCheck]:
|
||||
try:
|
||||
global_config = GlobalConfig.objects.get(name="default")
|
||||
except GlobalConfig.DoesNotExist:
|
||||
return [SelfCheck("Global config", "warning", "Default global config has not been created yet.")]
|
||||
|
||||
status: CheckStatus = "ok"
|
||||
message = "Default global config exists."
|
||||
if global_config.backup_root != settings.POBSYNC_BACKUP_ROOT:
|
||||
status = "warning"
|
||||
message = "Global config backup root differs from runtime POBSYNC_BACKUP_ROOT."
|
||||
return [
|
||||
SelfCheck(
|
||||
"Global config",
|
||||
status,
|
||||
message,
|
||||
f"database={global_config.backup_root} runtime={settings.POBSYNC_BACKUP_ROOT}",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _systemd_checks() -> list[SelfCheck]:
|
||||
if not Path("/run/systemd/system").exists() or shutil.which("systemctl") is None:
|
||||
return [
|
||||
SelfCheck(
|
||||
"Systemd services",
|
||||
"skipped",
|
||||
"systemd is not available in this runtime.",
|
||||
"This is expected inside Docker.",
|
||||
)
|
||||
]
|
||||
|
||||
checks = []
|
||||
for service in ("pobsync-web.service", "pobsync-worker.service", "pobsync-scheduler.service"):
|
||||
result = subprocess.run(
|
||||
["systemctl", "is-active", service],
|
||||
check=False,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
active_state = result.stdout.strip() or result.stderr.strip()
|
||||
checks.append(
|
||||
SelfCheck(
|
||||
service,
|
||||
"ok" if result.returncode == 0 else "failed",
|
||||
active_state,
|
||||
)
|
||||
)
|
||||
return checks
|
||||
Reference in New Issue
Block a user