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", pobsync_home="/opt/pobsync", rsync_args=["--archive"], rsync_extra_args=["--numeric-ids"], excludes_default=["/proc/***"], retention_daily=7, retention_weekly=4, retention_monthly=3, retention_yearly=1, data={ "backup_root": "/ignored", "pobsync_home": "/ignored", "ssh": {"user": "root", "port": 22, "options": []}, "rsync": { "binary": "rsync", "args": ["--archive"], "timeout_seconds": 0, "bwlimit_kbps": 0, "extra_args": ["--numeric-ids"], }, "defaults": {"source_root": "/", "destination_subdir": ""}, "excludes_default": ["/proc/***"], "retention_defaults": {"daily": 7, "weekly": 4, "monthly": 3, "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", pobsync_home="/opt/pobsync", 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", pobsync_home="/opt/pobsync", 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", pobsync_home="/opt/pobsync", 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", pobsync_home="/opt/pobsync", 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"])