# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Tests for cloud-init script generation."""

from base64 import b64decode
from textwrap import dedent

import yaml
from django.test import SimpleTestCase

from debusine.assets.models import (
    BasicAPTAuthenticationData,
    BasicAPTCredentials,
)
from debusine.db.models import Token
from debusine.server.worker_pools.cloud_init import (
    AptConfig,
    CloudInitConfig,
    FilesystemSetup,
    WriteFile,
    apt_source,
    worker_bootstrap_cloud_init,
)
from debusine.server.worker_pools.models import (
    DebianRelease,
    DebusineInstallSource,
)
from debusine.tasks.models import ExtraExternalRepository
from debusine.test.django import TestCase


class APTSourceTests(SimpleTestCase):
    """Tests for apt_source."""

    def test_daily_builds(self) -> None:
        source = apt_source(DebusineInstallSource.DAILY_BUILDS)
        self.assertEqual(source.components, ["main"])
        self.assertEqual(source.suite, "trixie-devel")
        self.assertEqual(
            str(source.url),
            "https://deb.debusine.freexian.com/freexian/debusine",
        )
        assert source.signing_key is not None
        self.assertTrue(
            source.signing_key.startswith(
                "-----BEGIN PGP PUBLIC KEY BLOCK-----"
            )
        )

    def test_daily_builds_specified_release(self) -> None:
        source = apt_source(
            DebusineInstallSource.DAILY_BUILDS, release=DebianRelease.FORKY
        )
        self.assertEqual(source.suite, "forky-devel")

    def test_enterprise_staging(self) -> None:
        source = apt_source(DebusineInstallSource.ENTERPRISE_STAGING)
        self.assertEqual(source.components, ["main"])
        self.assertEqual(source.suite, "trixie-staging")
        self.assertEqual(
            str(source.url),
            "https://deb.debusine.freexian.com/freexian/debusine-enterprise",
        )
        assert source.signing_key is not None
        self.assertTrue(
            source.signing_key.startswith(
                "-----BEGIN PGP PUBLIC KEY BLOCK-----"
            )
        )

    def test_enterprise(self) -> None:
        source = apt_source(DebusineInstallSource.ENTERPRISE)
        self.assertEqual(source.components, ["main"])
        self.assertEqual(source.suite, "trixie-enterprise")
        self.assertEqual(
            str(source.url),
            "https://deb.debusine.freexian.com/freexian/debusine-enterprise",
        )
        assert source.signing_key is not None
        self.assertTrue(
            source.signing_key.startswith(
                "-----BEGIN PGP PUBLIC KEY BLOCK-----"
            )
        )

    def test_custom(self) -> None:
        custom_source = ExtraExternalRepository(
            url="https://example.net/",
            components=["main"],
            suite="debusine",
            signing_key="PUBLIC KEY",
        )
        source = apt_source(DebusineInstallSource.CUSTOM, custom_source)
        self.assertEqual(source, custom_source)

    def test_custom_source_unspecified(self) -> None:
        with self.assertRaisesRegex(
            ValueError, "debusine_install_source_custom is not specified"
        ):
            apt_source(DebusineInstallSource.CUSTOM)


class CloudInitConfigTests(SimpleTestCase):
    """Tests for CloudInitConfig."""

    def test_full_instantiation(self) -> None:
        config = CloudInitConfig(
            apt=AptConfig(
                preserve_sources_list=False,
                sources_list="deb $MIRROR $RELEASE main",
            ),
            device_aliases={"foo": "/dev/nvme0n1"},
            fs_setup=[
                FilesystemSetup(device="foo", filesystem="ext4"),
            ],
            mounts=[
                ["foo", "/mnt/foo", "ext4", "defaults"],
            ],
            package_reboot_if_required=True,
            package_update=True,
            package_upgrade=True,
            packages=["build-essential", "vim-tiny"],
            bootcmd=["/bin/true"],
            runcmd=["/bin/true", "/bin/false"],
            write_files=[
                WriteFile(
                    path="/etc/hostname", content="test", permissions="0o644"
                ),
            ],
        )
        rendered = config.as_str()
        parsed = yaml.safe_load(rendered)
        self.assertIsInstance(parsed, dict)
        self.assertEqual(
            parsed["apt"],
            {
                "preserve_sources_list": False,
                "sources_list": "deb $MIRROR $RELEASE main",
            },
        )

        self.assertEqual(
            parsed["fs_setup"], [{"device": "foo", "filesystem": "ext4"}]
        )
        self.assertTrue(parsed["package_reboot_if_required"])

    def test_empty_as_str(self) -> None:
        config = CloudInitConfig()
        self.assertEqual(config.as_str(), "#cloud-config\n{}\n")

    def test_simple_as_str(self) -> None:
        config = CloudInitConfig(packages=["foobar"], runcmd=["/bin/true"])
        self.assertEqual(
            config.as_str(),
            dedent(
                """\
                #cloud-config
                packages:
                - foobar
                runcmd:
                - /bin/true
                """
            ),
        )

    def test_simple_as_base64_str(self) -> None:
        config = CloudInitConfig(packages=["foobar"], runcmd=["/bin/true"])
        encoded = config.as_base64_str()
        parsed = yaml.safe_load(b64decode(encoded))
        self.assertEqual(
            parsed, {"packages": ["foobar"], "runcmd": ["/bin/true"]}
        )


class TestWorkerBootstrapCloudInit(TestCase):
    """Tests for worker_bootstrap_cloud_init."""

    def get_file_by_path_if_exists(
        self, files: list[WriteFile], path: str
    ) -> WriteFile | None:
        """Return the file with path in files, if it exists."""
        for file in files:
            if file.path == path:
                return file
        return None

    def get_file_by_path(self, files: list[WriteFile], path: str) -> WriteFile:
        """Return the file with path in files."""
        file = self.get_file_by_path_if_exists(files, path)
        assert file is not None, f"File with {path=} not found in {files=}"
        return file

    def assertFileNotCreated(self, files: list[WriteFile], path: str) -> None:
        """Assert that path is not in files."""
        file = self.get_file_by_path_if_exists(files, path)
        self.assertIsNone(file)

    def test_worker_bootstrap_cloud_init_simple(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.BACKPORTS,
        )
        self.assertTrue(config.package_update)
        self.assertTrue(config.package_upgrade)
        self.assertTrue(config.package_reboot_if_required)
        self.assertEqual(len(config.bootcmd), 2)
        self.assertIn("eatmydata", config.bootcmd[0])
        self.assertEqual("worker", config.hostname)
        self.assertIn("debusine-worker", config.packages)
        self.assertIn("incus", config.packages)
        backports_pin_file = self.get_file_by_path(
            config.write_files, "/etc/apt/preferences.d/90-debusine-backports"
        )
        self.assertIn("src:debusine", backports_pin_file.content)
        self.assertIn("a=stable-backports", backports_pin_file.content)
        self.assertIn("a=stable\n", backports_pin_file.content)
        self.assertFileNotCreated(
            config.write_files,
            "/etc/apt/preferences.d/90-debusine-ignore-debian",
        )
        token_file = self.get_file_by_path(
            config.write_files, "/etc/debusine/worker/activation-token"
        )
        self.assertEqual(token_file.content, token.key)
        self.assertIsNone(token_file.owner)
        self.get_file_by_path(config.write_files, "/etc/nftables.conf")
        self.assertFileNotCreated(
            config.write_files, "/etc/apt/sources.list.d/debusine.sources"
        )

    def test_worker_bootstrap_cloud_init_daily(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.DAILY_BUILDS,
        )
        self.assertFileNotCreated(
            config.write_files, "/etc/apt/preferences.d/90-debusine-backports"
        )
        ignore_debian_pin_file = self.get_file_by_path(
            config.write_files,
            "/etc/apt/preferences.d/90-debusine-ignore-debian",
        )
        self.assertIn("release o=Debian", ignore_debian_pin_file.content)
        apt_source = self.get_file_by_path(
            config.write_files, "/etc/apt/sources.list.d/debusine.sources"
        )
        self.assertIn("Suites: trixie-devel", apt_source.content.splitlines())

    def test_worker_bootstrap_cloud_init_enterprise_staging(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.ENTERPRISE_STAGING,
            apt_authentication=BasicAPTAuthenticationData(
                name="enterprise",
                credentials=BasicAPTCredentials(
                    username="test",
                    password="secret-key",
                ),
            ),
        )
        self.assertFileNotCreated(
            config.write_files, "/etc/apt/preferences.d/90-debusine-backports"
        )
        ignore_debian_pin_file = self.get_file_by_path(
            config.write_files,
            "/etc/apt/preferences.d/90-debusine-ignore-debian",
        )
        self.assertIn("release o=Debian", ignore_debian_pin_file.content)
        apt_source = self.get_file_by_path(
            config.write_files, "/etc/apt/sources.list.d/debusine.sources"
        )
        self.assertIn("Suites: trixie-staging", apt_source.content.splitlines())
        apt_auth = self.get_file_by_path(
            config.write_files, "/etc/apt/auth.conf.d/enterprise.conf"
        )
        self.assertIn("password secret-key", apt_auth.content.splitlines())

    def test_worker_bootstrap_cloud_init_enterprise_no_authentication(
        self,
    ) -> None:
        token = self.playground.create_worker_activation_token()
        with self.assertRaisesRegex(
            ValueError,
            r"apt_authentication is required for ENTERPRISE sources\.",
        ):
            worker_bootstrap_cloud_init(
                "worker",
                token,
                debusine_install_source=DebusineInstallSource.ENTERPRISE_STAGING,
            )

    def test_worker_bootstrap_cloud_init_custom(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.CUSTOM,
            debusine_install_source_custom=ExtraExternalRepository(
                url="https://example.net/",
                components=["main"],
                suite="debusine",
                signing_key="PUBLIC KEY",
            ),
            apt_authentication=BasicAPTAuthenticationData(
                name="examplenet",
                credentials=BasicAPTCredentials(
                    username="test",
                    password="secret-key",
                ),
            ),
        )
        self.assertFileNotCreated(
            config.write_files, "/etc/apt/preferences.d/90-debusine-backports"
        )
        ignore_debian_pin_file = self.get_file_by_path(
            config.write_files,
            "/etc/apt/preferences.d/90-debusine-ignore-debian",
        )
        self.assertIn("release o=Debian", ignore_debian_pin_file.content)
        apt_source = self.get_file_by_path(
            config.write_files, "/etc/apt/sources.list.d/debusine.sources"
        )
        self.assertIn("Suites: debusine", apt_source.content.splitlines())
        apt_auth = self.get_file_by_path(
            config.write_files, "/etc/apt/auth.conf.d/examplenet.conf"
        )
        self.assertIn(
            "machine https://example.net/", apt_auth.content.splitlines()
        )

    def test_worker_bootstrap_cloud_init_pre_installed(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.PRE_INSTALLED,
        )
        self.assertTrue(config.package_update)
        self.assertTrue(config.package_upgrade)
        self.assertTrue(config.package_reboot_if_required)
        self.assertEqual(config.bootcmd, [])
        self.assertEqual("worker", config.hostname)
        self.assertEqual(
            config.write_files,
            [
                WriteFile(
                    path="/etc/debusine/worker/activation-token",
                    permissions="0o600",
                    content=token.key,
                    owner="debusine-worker:debusine-worker",
                )
            ],
        )
        self.assertEqual(
            config.runcmd, ["systemctl restart debusine-worker.service"]
        )
        self.assertEqual(config.packages, [])

    def test_worker_bootstrap_trixie(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.RELEASE,
            debian_release=DebianRelease.TRIXIE,
        )
        self.assertIsNone(config.apt)
        self.assertFileNotCreated(
            config.write_files, "/etc/apt/preferences.d/90-debusine-backports"
        )
        self.assertFileNotCreated(
            config.write_files,
            "/etc/apt/preferences.d/90-debusine-ignore-debian",
        )

    def test_worker_bootstrap_trixie_bpo(self) -> None:
        token = self.playground.create_worker_activation_token()
        config = worker_bootstrap_cloud_init(
            "worker",
            token,
            debusine_install_source=DebusineInstallSource.BACKPORTS,
            debian_release=DebianRelease.TRIXIE,
        )
        self.assertIsNone(config.apt)
        self.get_file_by_path(
            config.write_files, "/etc/apt/preferences.d/90-debusine-backports"
        )

    def test_worker_bootstrap_cloud_init_without_key(self) -> None:
        token = Token()
        with self.assertRaisesRegex(
            ValueError,
            (
                r"Requires a fresh token with a key. Token<None> does not have "
                r"one\."
            ),
        ):
            worker_bootstrap_cloud_init(
                "worker",
                token,
                debusine_install_source=DebusineInstallSource.BACKPORTS,
            )
