Peter van Arkel
86873bd035
Drop the unused GlobalConfig.data field and remove the remaining YAML config path helpers from PobsyncPaths. Keep HostConfig.config as runtime state for preflight data, and relabel it in the admin so it no longer reads as legacy compatibility storage.
141 lines
6.5 KiB
Python
141 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
import stat
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from django.test import TestCase, override_settings
|
|
|
|
from pobsync_backend.config_source import DjangoConfigSource
|
|
from pobsync_backend.models import GlobalConfig, HostConfig, SshCredential
|
|
|
|
|
|
class DjangoConfigSourceTests(TestCase):
|
|
def test_returns_effective_config_from_database(self) -> None:
|
|
GlobalConfig.objects.create(
|
|
name="default",
|
|
backup_root="/backups",
|
|
rsync_args=["--archive"],
|
|
rsync_extra_args=["--numeric-ids"],
|
|
excludes_default=["/proc/***"],
|
|
retention_daily=7,
|
|
retention_weekly=4,
|
|
retention_monthly=3,
|
|
retention_yearly=1,
|
|
)
|
|
HostConfig.objects.create(
|
|
host="web-01",
|
|
address="web-01.example.test",
|
|
excludes_add=["/tmp/***"],
|
|
rsync_extra_args=["--delete"],
|
|
retention_daily=7,
|
|
retention_weekly=4,
|
|
retention_monthly=3,
|
|
retention_yearly=1,
|
|
config={
|
|
"retention": {"daily": 99, "weekly": 99, "monthly": 99, "yearly": 99},
|
|
"excludes_add": ["/ignored/***"],
|
|
"rsync": {"extra_args": ["--ignored"]},
|
|
},
|
|
)
|
|
|
|
cfg = DjangoConfigSource().effective_config_for_host("web-01")
|
|
|
|
self.assertEqual(cfg["backup_root"], "/backups")
|
|
self.assertEqual(cfg["host"], "web-01")
|
|
self.assertEqual(cfg["address"], "web-01.example.test")
|
|
self.assertEqual(cfg["excludes_effective"], ["/proc/***", "/tmp/***"])
|
|
self.assertEqual(cfg["rsync"]["args_effective"], ["--archive", "--numeric-ids", "--delete"])
|
|
|
|
def test_materializes_global_ssh_credential_for_runtime_config(self) -> None:
|
|
credential = SshCredential.objects.create(
|
|
name="backup-key",
|
|
private_key="PRIVATE KEY",
|
|
known_hosts="web-01.example.test ssh-ed25519 AAAATEST",
|
|
)
|
|
GlobalConfig.objects.create(
|
|
name="default",
|
|
backup_root="/backups",
|
|
default_ssh_credential=credential,
|
|
ssh_options=["-oBatchMode=yes"],
|
|
)
|
|
HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
|
|
|
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
|
cfg = DjangoConfigSource().effective_config_for_host("web-01")
|
|
|
|
identity_file = Path(tmp) / "home" / "state" / "ssh-credentials" / str(credential.pk) / "identity"
|
|
known_hosts = identity_file.parent / "known_hosts"
|
|
self.assertEqual(identity_file.read_text(encoding="utf-8"), "PRIVATE KEY\n")
|
|
self.assertEqual(known_hosts.read_text(encoding="utf-8"), "web-01.example.test ssh-ed25519 AAAATEST\n")
|
|
self.assertEqual(stat.S_IMODE(identity_file.stat().st_mode), 0o600)
|
|
|
|
self.assertIn("-oBatchMode=yes", cfg["ssh"]["options"])
|
|
self.assertIn(f"-oIdentityFile={identity_file}", cfg["ssh"]["options"])
|
|
self.assertIn(f"-oUserKnownHostsFile={known_hosts}", cfg["ssh"]["options"])
|
|
self.assertNotIn("-oStrictHostKeyChecking=accept-new", cfg["ssh"]["options"])
|
|
self.assertEqual(cfg["ssh_credential"]["storage"], "database")
|
|
|
|
def test_host_ssh_credential_overrides_global_credential(self) -> None:
|
|
global_credential = SshCredential.objects.create(name="global-key", private_key="GLOBAL")
|
|
host_credential = SshCredential.objects.create(name="host-key", private_key="HOST")
|
|
GlobalConfig.objects.create(
|
|
name="default",
|
|
backup_root="/backups",
|
|
default_ssh_credential=global_credential,
|
|
)
|
|
HostConfig.objects.create(
|
|
host="web-01",
|
|
address="web-01.example.test",
|
|
ssh_credential=host_credential,
|
|
)
|
|
|
|
with TemporaryDirectory() as tmp, override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
|
cfg = DjangoConfigSource().effective_config_for_host("web-01")
|
|
|
|
host_identity_file = Path(tmp) / "home" / "state" / "ssh-credentials" / str(host_credential.pk) / "identity"
|
|
global_identity_file = Path(tmp) / "home" / "state" / "ssh-credentials" / str(global_credential.pk) / "identity"
|
|
self.assertEqual(host_identity_file.read_text(encoding="utf-8"), "HOST\n")
|
|
self.assertFalse(global_identity_file.exists())
|
|
|
|
self.assertIn(f"-oIdentityFile={host_identity_file}", cfg["ssh"]["options"])
|
|
|
|
def test_filesystem_ssh_credential_uses_existing_key_path(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
identity_file = Path(tmp) / "identity"
|
|
identity_file.write_text("PRIVATE KEY\n", encoding="utf-8")
|
|
identity_file.chmod(0o600)
|
|
credential = SshCredential.objects.create(name="backup-key", key_path=str(identity_file), public_key="PUBLIC")
|
|
GlobalConfig.objects.create(
|
|
name="default",
|
|
backup_root="/backups",
|
|
default_ssh_credential=credential,
|
|
)
|
|
HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
|
|
|
cfg = DjangoConfigSource().effective_config_for_host("web-01")
|
|
|
|
self.assertIn(f"-oIdentityFile={identity_file}", cfg["ssh"]["options"])
|
|
self.assertEqual(cfg["ssh_credential"]["storage"], "filesystem")
|
|
|
|
def test_missing_known_hosts_uses_service_accept_new_file(self) -> None:
|
|
with TemporaryDirectory() as tmp:
|
|
identity_file = Path(tmp) / "identity"
|
|
identity_file.write_text("PRIVATE KEY\n", encoding="utf-8")
|
|
identity_file.chmod(0o600)
|
|
credential = SshCredential.objects.create(name="backup-key", key_path=str(identity_file), public_key="PUBLIC")
|
|
GlobalConfig.objects.create(
|
|
name="default",
|
|
backup_root="/backups",
|
|
default_ssh_credential=credential,
|
|
)
|
|
HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
|
|
|
with override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
|
|
cfg = DjangoConfigSource().effective_config_for_host("web-01")
|
|
service_known_hosts = Path(tmp) / "home" / "state" / "known_hosts"
|
|
|
|
self.assertTrue(service_known_hosts.exists())
|
|
self.assertIn(f"-oUserKnownHostsFile={service_known_hosts}", cfg["ssh"]["options"])
|
|
self.assertIn("-oStrictHostKeyChecking=accept-new", cfg["ssh"]["options"])
|