pobsync/src/pobsync_backend/tests/test_django_config_source.py

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"],
            rsync_bwlimit_kbps=10000,
            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"],
            rsync_bwlimit_kbps=2500,
            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"])
        self.assertEqual(cfg["rsync"]["bwlimit_kbps"], 2500)

    def test_host_can_disable_global_rsync_bandwidth_limit(self) -> None:
        GlobalConfig.objects.create(
            name="default",
            backup_root="/backups",
            rsync_args=["--archive"],
            rsync_bwlimit_kbps=5000,
        )
        HostConfig.objects.create(
            host="web-01",
            address="web-01.example.test",
            rsync_bwlimit_kbps=0,
        )

        cfg = DjangoConfigSource().effective_config_for_host("web-01")

        self.assertEqual(cfg["rsync"]["bwlimit_kbps"], 0)

    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")

            with override_settings(POBSYNC_HOME=str(Path(tmp) / "home")):
                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"])