# 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 the Asset artifacts."""

import base64

from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.db import IntegrityError, transaction

from debusine.assets import (
    AWSProviderAccountConfiguration,
    AWSProviderAccountCredentials,
    AWSProviderAccountData,
    AssetCategory,
    BasicAPTAuthenticationData,
    BasicAPTCredentials,
    KeyPurpose,
    SigningKeyData,
)
from debusine.db.context import context
from debusine.db.models import Asset, Scope, Workspace
from debusine.db.models.assets import (
    AssetRoles,
    AssetUsage,
    AssetUsageRole,
    AssetUsageRoleBase,
    AssetUsageRoles,
)
from debusine.db.models.scopes import ScopeRoles
from debusine.db.models.tests.utils import RolesTestCase
from debusine.db.permissioncontext import PermissionContext
from debusine.db.playground import scenarios
from debusine.tasks.models import WorkerType
from debusine.test.django import TestCase


class AssetUsageRolesTests(RolesTestCase[AssetUsageRoles]):
    """Tests for the AssetUsageRoles class."""

    roles_class = AssetUsageRoles

    def test_invariants(self) -> None:
        self.assertRolesInvariants()

    def test_from_iterable(self) -> None:
        for value, expected in (
            ([], []),
            (["signer"], [AssetUsage.Roles.SIGNER]),
            (["repository_signer"], [AssetUsage.Roles.REPOSITORY_SIGNER]),
            ([AssetUsage.Roles.SIGNER], [AssetUsage.Roles.SIGNER]),
            (
                [AssetUsage.Roles.REPOSITORY_SIGNER],
                [AssetUsage.Roles.REPOSITORY_SIGNER],
            ),
            (["signer", "repository_signer"], [AssetUsage.Roles.SIGNER]),
            (
                [
                    "signer",
                    "signer",
                    AssetUsage.Roles.SIGNER,
                    AssetUsage.Roles.REPOSITORY_SIGNER,
                ],
                [AssetUsage.Roles.SIGNER],
            ),
        ):
            with self.subTest(value=repr(value)):
                self.assertEqual(
                    AssetUsage.Roles.from_iterable(value),
                    frozenset(expected),
                )

    def test_choices(self) -> None:
        self.assertChoices()

    def test_values(self) -> None:
        role = AssetUsage.Roles.SIGNER
        self.assertEqual(role.label, "Signer")
        self.assertEqual(
            role.description,
            "Sign data using this key",
        )
        self.assertEqual(role.implied_by_scope_roles, frozenset([]))
        self.assertEqual(role.implied_by_workspace_roles, frozenset([]))
        self.assertEqual(role.implied_by_asset_usage_roles, frozenset([role]))
        self.assertTrue(role.implies(AssetUsage.Roles.REPOSITORY_SIGNER))

        role = AssetUsage.Roles.REPOSITORY_SIGNER
        self.assertEqual(role.label, "Repository signer")
        self.assertEqual(
            role.description,
            "Sign repositories using this key",
        )
        self.assertEqual(
            role.implied_by_scope_roles, frozenset([Scope.Roles.OWNER])
        )
        self.assertEqual(
            role.implied_by_workspace_roles, frozenset([Workspace.Roles.OWNER])
        )
        self.assertEqual(
            role.implied_by_asset_usage_roles,
            frozenset([role, AssetUsageRoles.SIGNER]),
        )
        self.assertFalse(role.implies(AssetUsage.Roles.SIGNER))

        role = AssetUsage.Roles.APT_AUTHENTICATOR
        self.assertEqual(role.label, "Can use APT authentication")
        self.assertEqual(
            role.description,
            "Authenticate to an external APT repository "
            "using these credentials",
        )
        self.assertEqual(
            role.implied_by_scope_roles,
            frozenset([Scope.Roles.OWNER]),
        )

        self.assertEqual(
            role.implied_by_workspace_roles,
            frozenset([Workspace.Roles.OWNER]),
        )

        self.assertEqual(
            role.implied_by_asset_usage_roles,
            frozenset([role]),
        )

    def test_init_invalid_implication(self) -> None:
        r = AssetUsageRoleBase("test", implied_by=[None])
        with self.assertRaisesRegex(
            ImproperlyConfigured,
            r"AssetUsage roles do not support implications by None",
        ):
            r._setup()


class AssetManagerTests(TestCase):
    """Tests for AssetManager and AssetQuerySet."""

    def test_in_current_scope(self) -> None:
        """Test in_current_scope()."""
        scope2 = self.playground.get_or_create_scope("scope2")
        default_workspace = self.playground.get_default_workspace()
        self.playground.create_signing_key_asset()

        with context.local():
            context.set_permission_context_scoped(
                default_workspace.scope, PermissionContext.for_ui_anonymous()
            )
            self.assertTrue(Asset.objects.in_current_scope().exists())

        with context.local():
            context.set_permission_context_scoped(
                scope2, PermissionContext.for_ui_anonymous()
            )
            self.assertFalse(Asset.objects.in_current_scope().exists())

    def test_can_display(self) -> None:
        """Test can_display() delegates to the workspace."""
        default_workspace = self.playground.get_default_workspace()
        private_scope = self.playground.get_or_create_scope("private")
        private_workspace = self.playground.create_workspace(
            public=False, scope=private_scope
        )
        # No access to private_workspace
        user = self.playground.create_user(username="unprivileged")
        asset = self.playground.create_signing_key_asset()
        self.playground.create_signing_key_asset(workspace=private_workspace)

        with context.local():
            context.set_permission_context_scoped(
                default_workspace.scope, PermissionContext.for_ui_user(user)
            )

            self.assertEqual(Asset.objects.can_display(context.pc).count(), 1)
            self.assertEqual(
                Asset.objects.can_display(context.pc).first(), asset
            )

    def test_can_display_cloud_provider_accounts(self) -> None:
        default_workspace = self.playground.get_default_workspace()
        user = self.playground.get_default_user()
        asset = self.playground.create_asset(
            category=AssetCategory.CLOUD_PROVIDER_ACCOUNT,
            workspace=default_workspace,
            data=AWSProviderAccountData(
                name="test",
                configuration=AWSProviderAccountConfiguration(
                    region_name="test-region"
                ),
                credentials=AWSProviderAccountCredentials(
                    access_key_id="access-key", secret_access_key="secret-key"
                ),
            ).model_dump(),
            created_by=user,
        )
        self.playground.create_group_role(asset, AssetRoles.OWNER, users=[user])

        with context.local():
            context.set_permission_context_scoped(
                default_workspace.scope, PermissionContext.for_ui_user(user)
            )
            self.assertQuerySetEqual(Asset.objects.can_display(context.pc), [])

    def test_can_manage_permissions(self) -> None:
        """Test can_manage_permissions()."""
        user = self.playground.get_default_user()
        asset = self.playground.create_signing_key_asset()
        self.assertFalse(
            Asset.objects.can_manage_permissions(
                PermissionContext.for_ui_user(user)
            ).exists()
        )

        self.playground.create_group_role(asset, AssetRoles.OWNER, users=[user])
        self.assertEqual(
            Asset.objects.can_manage_permissions(
                PermissionContext.for_ui_user(user)
            ).first(),
            asset,
        )

    def test_get_by_slug_signing_key(self) -> None:
        asset = self.playground.create_signing_key_asset()
        self.assertEqual(
            Asset.objects.get_by_slug(category=asset.category, slug=asset.slug),
            asset,
        )

    def test_get_by_slug_apt_authentication_by_name(self) -> None:
        asset = self.playground.create_apt_authentication_asset(name="test-1")
        self.playground.create_apt_authentication_asset(name="test-2")
        self.assertEqual(
            Asset.objects.get_by_slug(
                category=asset.category,
                slug="test-1",
                workspace=self.playground.get_default_workspace(),
            ),
            asset,
        )

    def test_get_by_slug_apt_authentication_by_workspace_and_name(self) -> None:
        other_workspace = self.playground.create_workspace(name="other")
        asset = self.playground.create_apt_authentication_asset(
            workspace=other_workspace, name="test"
        )
        self.playground.create_apt_authentication_asset(name="test")
        self.assertEqual(
            Asset.objects.get_by_slug(
                category=asset.category,
                slug="other:test",
                workspace=self.playground.get_default_workspace(),
            ),
            asset,
        )

    def test_get_by_slug_apt_authentication_system(self) -> None:
        asset = self.playground.create_apt_authentication_asset(
            name="test", workspace=None
        )
        self.playground.create_apt_authentication_asset(name="test")
        self.assertEqual(
            Asset.objects.get_by_slug(
                category=asset.category,
                slug=":test",
            ),
            asset,
        )

    def test_get_by_slug_unknown_category(self) -> None:
        with self.assertRaisesRegex(
            ValueError, r"No slug defined for category 'unknown'"
        ):
            Asset.objects.get_by_slug(category="unknown", slug="abc123")


class AssetUsageManagerTests(TestCase):
    """Tests for AssetUsageManager."""

    scenario = scenarios.DefaultContext()

    def test_get_roles_model(self) -> None:
        self.assertEqual(AssetUsage.objects.get_roles_model(), AssetUsageRole)

    def test_with_role(self) -> None:
        asset = self.playground.create_signing_key_asset()
        asset_usage = self.playground.create_asset_usage(resource=asset)
        signers = self.playground.create_group_role(
            asset_usage, AssetUsageRoles.SIGNER, users=[self.scenario.user]
        )
        other = self.playground.create_user("other")

        for pc, expected in (
            (PermissionContext.for_ui_anonymous(), []),
            (PermissionContext.for_ui_user(other), []),
            (
                PermissionContext.for_ui_user(other).with_extra_groups(signers),
                [asset_usage],
            ),
            (self.scenario.context_ui(), [asset_usage]),
        ):
            with self.subTest(pc=str(pc)):
                self.assertQuerySetEqual(
                    AssetUsage.objects.with_role(pc, AssetUsage.Roles.SIGNER),
                    expected,
                )


class AssetTests(TestCase):
    """Tests for Asset."""

    def create_signing_key_asset(self) -> Asset:
        """Create an unsaved signing-key Asset."""
        return Asset(
            category=AssetCategory.SIGNING_KEY,
            workspace=self.playground.get_default_workspace(),
            data=SigningKeyData(
                purpose=KeyPurpose.OPENPGP_UPLOAD,
                fingerprint="ABC123",
                public_key=base64.b64encode(b"PUBLIC KEY").decode(),
                description="A Description",
            ).model_dump(),
            created_by=self.playground.get_default_user(),
        )

    def create_cloud_provider_account_asset(
        self, workspace: Workspace | None
    ) -> Asset:
        """Create an unsaved cloud-provider-account Asset."""
        return Asset(
            category=AssetCategory.CLOUD_PROVIDER_ACCOUNT,
            workspace=workspace,
            data=AWSProviderAccountData(
                name="test",
                configuration=AWSProviderAccountConfiguration(
                    region_name="test-region"
                ),
                credentials=AWSProviderAccountCredentials(
                    access_key_id="access-key", secret_access_key="secret-key"
                ),
            ).model_dump(),
            created_by=self.playground.get_default_user(),
        )

    def create_apt_authentication_asset(self, system: bool = False) -> Asset:
        """Create an unsaved apt-authentication Asset."""
        return Asset(
            category=AssetCategory.APT_AUTHENTICATION,
            workspace=(
                None if system else self.playground.get_default_workspace()
            ),
            data=BasicAPTAuthenticationData(
                name="test",
                credentials=BasicAPTCredentials(
                    username="test-username", password="test-password"
                ),
            ).model_dump(),
            created_by=self.playground.get_default_user(),
        )

    def test_signing_key_unique_constraint(self) -> None:
        """Test that signing-key assets have a unique constraint."""
        self.playground.create_signing_key_asset(fingerprint="ABC123")
        with self.assertRaises(IntegrityError):
            self.playground.create_signing_key_asset(fingerprint="ABC123")

    def test_apt_authentication_name(self) -> None:
        """apt-authentication assets must have valid names."""
        valid_names = ("secret", "foo_bar", "foo.bar", "foo+bar", "tail_")
        invalid_names = (
            "+foo",
            "-foo",
            ".foo",
            "_foo",
            "foo:bar",
            "/etc/passwd",
            "../foo",
        )

        for name in valid_names:
            with self.subTest(name=name):
                self.playground.create_apt_authentication_asset(name=name)

        for name in invalid_names:
            with (
                self.subTest(name=name),
                transaction.atomic(),
                self.assertRaises(IntegrityError),
            ):
                self.playground.create_apt_authentication_asset(name=name)

    def test_apt_authentication_unique_constraint(self) -> None:
        """Test that apt-authentication assets have a unique constraint."""
        self.playground.create_apt_authentication_asset(name="test-apt")
        with self.assertRaises(IntegrityError):
            self.playground.create_apt_authentication_asset(name="test-apt")

    def test_clean(self) -> None:
        """Test that clean() validates the model."""
        asset = self.playground.create_signing_key_asset()
        asset.clean()
        asset.data["something"] = 42
        with self.assertRaisesRegex(
            ValidationError, r"invalid asset category or data"
        ):
            asset.clean()

    def test_slug_signing_key(self) -> None:
        asset = self.create_signing_key_asset()
        self.assertEqual(asset.slug, "openpgp:ABC123")

    def test_slug_apt_authentication(self) -> None:
        asset = self.create_apt_authentication_asset()
        workspace = self.playground.get_default_workspace()
        self.assertEqual(asset.slug, f"{workspace.name}:test")

    def test_slug_apt_authentication_system(self) -> None:
        asset = self.create_apt_authentication_asset(system=True)
        self.assertEqual(asset.slug, ":test")

    def test_slug_unknown_category(self) -> None:
        asset = self.create_signing_key_asset()
        asset.category = "unknown"
        with self.assertRaisesRegex(
            NotImplementedError, r"No slug defined for category 'unknown'"
        ):
            asset.slug

    def test_can_edit(self) -> None:
        """Test that can_edit() only permits direct owners to edit."""
        asset = self.playground.create_signing_key_asset()
        self.assertFalse(asset.can_edit(PermissionContext.for_ui_anonymous()))
        user = self.playground.get_default_user()
        pc = PermissionContext.for_ui_user(user)
        self.assertFalse(asset.can_edit(pc))

        self.playground.create_group_role(asset, AssetRoles.OWNER, users=[user])
        self.assertTrue(asset.can_edit(pc))

    def test_can_create(self) -> None:
        """Test that can_create() only permits scope owners to create."""
        asset = self.create_signing_key_asset()
        assert asset.workspace

        self.assertFalse(asset.can_create(PermissionContext.for_ui_anonymous()))
        user = self.playground.get_default_user()
        pc = PermissionContext.for_ui_user(user)
        self.assertFalse(asset.can_create(pc))

        self.playground.create_group_role(
            asset.workspace.scope, ScopeRoles.OWNER, users=[user]
        )
        self.assertTrue(asset.can_create(pc))

    def test_can_create_worker(self) -> None:
        """Test that can_create() can be used by signing workers."""
        asset = self.create_signing_key_asset()
        assert asset.workspace

        self.assertFalse(asset.can_create(PermissionContext.for_ui_anonymous()))
        ext_worker_token = self.playground.create_work_request_token(
            self.playground.create_worker_task(
                mark_running=True, assign_new_worker=True
            )
        )
        sig_worker_token = self.playground.create_work_request_token(
            self.playground.create_worker_task(
                mark_running=True,
                worker=self.playground.create_worker(
                    worker_type=WorkerType.SIGNING
                ),
            )
        )
        with context.local():
            context.set_permission_context_scoped(
                asset.workspace.scope,
                PermissionContext.for_work_request(token=ext_worker_token),
            )
            self.assertFalse(asset.can_create(context.pc))

        with context.local():
            context.set_permission_context_scoped(
                asset.workspace.scope,
                PermissionContext.for_work_request(token=sig_worker_token),
            )
            self.assertTrue(asset.can_create(context.pc))

    def test_can_create_no_workspace(self) -> None:
        asset = self.create_cloud_provider_account_asset(workspace=None)
        user = self.playground.get_default_user()
        self.assertFalse(asset.can_create(PermissionContext.for_ui_user(user)))

    def test_can_create_cloud_provider_account(self) -> None:
        asset = self.create_cloud_provider_account_asset(
            workspace=self.playground.get_default_workspace()
        )
        user = self.playground.get_default_user()
        assert asset.workspace
        self.playground.create_group_role(
            asset.workspace.scope, ScopeRoles.OWNER, users=[user]
        )
        self.assertFalse(asset.can_create(PermissionContext.for_ui_user(user)))

    def test_data_model(self) -> None:
        """Test data_model validations: No data."""
        asset = self.playground.create_signing_key_asset()

        self.assertIsInstance(asset.data_model, SigningKeyData)

    def test_data_model_validation_none(self) -> None:
        """Test data_model validations: No data."""
        asset = self.create_signing_key_asset()
        asset.data = None

        with self.assertRaisesRegex(
            ValidationError, r"data must be a dictionary"
        ):
            asset.data_model

    def test_data_model_validation_category(self) -> None:
        """Test data_model validations: Unknown category."""
        asset = self.create_signing_key_asset()
        asset.category = "unknown"

        with self.assertRaisesRegex(ValidationError, r"invalid asset category"):
            asset.data_model

    def test_save_create(self) -> None:
        """Test that save requires can_create permission for creation."""
        user = self.playground.get_default_user()
        asset = self.create_signing_key_asset()
        assert asset.workspace
        self.playground.create_group_role(
            asset.workspace.scope, ScopeRoles.OWNER, users=[user]
        )
        with context.local():
            context.set_permission_context_scoped(
                asset.workspace.scope, PermissionContext.for_ui_user(user)
            )
            asset.workspace.set_current()
            self.assertTrue(asset.can_create(context.pc))

            asset.save()

    def test_save_edit(self) -> None:
        """Test that save requires can_edit permission for modification."""
        user = self.playground.get_default_user()
        asset = self.playground.create_signing_key_asset()
        assert asset.workspace
        self.playground.create_group_role(asset, AssetRoles.OWNER, users=[user])
        with context.local():
            context.set_permission_context_scoped(
                asset.workspace.scope, PermissionContext.for_ui_user(user)
            )
            asset.workspace.set_current()
            self.assertTrue(asset.can_edit(context.pc))

            asset.save()

    def test_str(self) -> None:
        asset = self.create_signing_key_asset()
        self.assertEqual(
            str(asset),
            (
                "Id: None Category: debusine:signing-key "
                "Workspace: debusine/System"
            ),
        )


class AssetUsageTests(TestCase):
    """Tests for AssetUsage."""

    scenario = scenarios.DefaultContext()

    def test_has_role(self) -> None:
        asset = self.playground.create_signing_key_asset()
        asset_usage = self.playground.create_asset_usage(resource=asset)
        signers = self.playground.create_group_role(
            asset_usage, AssetUsageRoles.SIGNER, users=[self.scenario.user]
        )
        other = self.playground.create_user("other")

        for pc, expected in (
            (PermissionContext.for_ui_anonymous(), False),
            (PermissionContext.for_ui_user(other), False),
            (
                PermissionContext.for_ui_user(other).with_extra_groups(signers),
                True,
            ),
            (self.scenario.context_ui(), True),
        ):
            with self.subTest(pc=str(pc)):
                self.assertEqual(
                    asset_usage.has_role(pc, AssetUsage.Roles.SIGNER),
                    expected,
                )

    def test_can_sign_with(self) -> None:
        user = self.scenario.user
        asset = self.playground.create_signing_key_asset()
        asset_usage = self.playground.create_asset_usage(resource=asset)
        for u, workspace_roles, asset_usage_roles, expected in (
            (AnonymousUser(), [], [], False),
            (user, [], [], False),
            (user, [Workspace.Roles.OWNER], [], False),
            (user, [], [AssetUsage.Roles.SIGNER], True),
            (user, [], [AssetUsage.Roles.REPOSITORY_SIGNER], False),
        ):
            with (
                self.subTest(
                    user=repr(u),
                    workspace_roles=repr(workspace_roles),
                    asset_usage_roles=repr(asset_usage_roles),
                ),
                self.scenario.assign_role(
                    self.scenario.workspace, *workspace_roles
                ),
                self.scenario.assign_role(asset_usage, *asset_usage_roles),
            ):
                if u.is_authenticated:
                    pc = PermissionContext.for_ui_user(u)
                else:
                    pc = PermissionContext.for_ui_anonymous()
                self.assertPermissionPredicate(
                    asset_usage.can_sign_with,
                    pc,
                    expected,
                )

    def test_can_sign_repository_with(self) -> None:
        user = self.scenario.user
        asset = self.playground.create_signing_key_asset()
        asset_usage = self.playground.create_asset_usage(resource=asset)
        for u, workspace_roles, asset_usage_roles, expected in (
            (AnonymousUser(), [], [], False),
            (user, [], [], False),
            (user, [Workspace.Roles.OWNER], [], True),
            (user, [], [AssetUsage.Roles.SIGNER], True),
            (user, [], [AssetUsage.Roles.REPOSITORY_SIGNER], True),
        ):
            with (
                self.subTest(
                    user=repr(u),
                    workspace_roles=repr(workspace_roles),
                    asset_usage_roles=repr(asset_usage_roles),
                ),
                self.scenario.assign_role(
                    self.scenario.workspace, *workspace_roles
                ),
                self.scenario.assign_role(asset_usage, *asset_usage_roles),
            ):
                if u.is_authenticated:
                    pc = PermissionContext.for_ui_user(u)
                else:
                    pc = PermissionContext.for_ui_anonymous()
                self.assertPermissionPredicate(
                    asset_usage.can_sign_repository_with, pc, expected
                )

    def test_can_use_apt_authentication_with(self) -> None:
        user = self.scenario.user
        asset = self.playground.create_apt_authentication_asset()
        asset_usage = self.playground.create_asset_usage(resource=asset)
        for u, workspace_roles, asset_usage_roles, expected in (
            (AnonymousUser(), [], [], False),
            (user, [], [], False),
            (user, [Workspace.Roles.OWNER], [], True),
            (user, [], [AssetUsage.Roles.APT_AUTHENTICATOR], True),
        ):
            with (
                self.subTest(
                    user=repr(u),
                    workspace_roles=repr(workspace_roles),
                    asset_usage_roles=repr(asset_usage_roles),
                ),
                self.scenario.assign_role(
                    self.scenario.workspace, *workspace_roles
                ),
                self.scenario.assign_role(asset_usage, *asset_usage_roles),
            ):
                if u.is_authenticated:
                    pc = PermissionContext.for_ui_user(user)
                else:
                    pc = PermissionContext.for_ui_anonymous()
                self.assertPermissionPredicate(
                    asset_usage.can_use_apt_authentication_with, pc, expected
                )
