# 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.

"""Unit tests for the models."""

import datetime as dt
from pathlib import Path
from unittest import mock

import django.db.utils
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import PermissionDenied, ValidationError
from django.test import RequestFactory
from django.utils import timezone

from debusine.artifacts.local_artifact import WorkRequestDebugLogs
from debusine.artifacts.models import ArtifactCategory, EmptyArtifactData
from debusine.db.context import ContextConsistencyError, context
from debusine.db.models import (
    Artifact,
    ArtifactRelation,
    File,
    FileInArtifact,
    FileUpload,
    Scope,
    Workspace,
)
from debusine.db.models.permissions import Allow
from debusine.db.permissioncontext import PermissionContext
from debusine.db.playground.scenarios import DefaultContext
from debusine.test.django import (
    AllowAll,
    DenyAll,
    ListFilter,
    TestCase,
    override_permission,
)
from debusine.test.test_utils import data_generator


class ArtifactManagerTests(TestCase):
    """Tests for the ArtifactManager class."""

    scenario = DefaultContext()

    def test_create_from_local_artifact(self) -> None:
        """create_from_local_artifact() creates a matching Artifact."""
        temp_dir = self.create_temporary_directory()
        paths = [temp_dir / "1.log", temp_dir / "2.log"]
        paths[0].write_text("1\n")
        paths[1].write_text("2\n")
        local_artifact = WorkRequestDebugLogs.create(files=paths)
        work_request = self.playground.create_worker_task()

        with context.disable_permission_checks():
            artifact = Artifact.objects.create_from_local_artifact(
                local_artifact,
                self.scenario.workspace,
                created_by_work_request=work_request,
            )

        self.assertEqual(artifact.category, local_artifact.category)
        self.assertEqual(artifact.workspace, self.scenario.workspace)
        self.assertEqual(artifact.data, local_artifact.data.model_dump())
        self.assertEqual(artifact.created_by_work_request, work_request)
        self.assertEqual(artifact.fileinartifact_set.count(), 2)
        for file_in_artifact, path in zip(
            artifact.fileinartifact_set.order_by("id"), paths
        ):
            self.assertEqual(file_in_artifact.path, path.name)
            fileobj = file_in_artifact.file
            file_backend = self.scenario.scope.download_file_backend(fileobj)
            with file_backend.get_stream(fileobj) as file:
                self.assertEqual(file.read(), path.read_bytes())

    @context.disable_permission_checks()
    def test_not_expired_return_not_expired_artifacts(self) -> None:
        """not_expired() return only the not expired artifacts."""
        artifact_1 = Artifact.objects.create(
            workspace=self.scenario.workspace,
            expiration_delay=dt.timedelta(days=1),
        )
        artifact_1.created_at = timezone.now() - dt.timedelta(days=2)
        artifact_1.save()
        artifact_2 = Artifact.objects.create(
            workspace=self.scenario.workspace,
            expiration_delay=dt.timedelta(days=1),
        )
        artifact_3 = Artifact.objects.create(
            workspace=self.scenario.workspace, expiration_delay=dt.timedelta(0)
        )

        self.assertQuerySetEqual(
            Artifact.objects.not_expired(timezone.now()),
            {artifact_2, artifact_3},
            ordered=False,
        )

    @context.disable_permission_checks()
    def test_expired_return_expired_artifacts(self) -> None:
        """expired() return only the expired artifacts."""
        artifact_1 = Artifact.objects.create(
            workspace=self.scenario.workspace,
            expiration_delay=dt.timedelta(days=1),
        )
        artifact_1.created_at = timezone.now() - dt.timedelta(days=2)
        artifact_1.save()

        Artifact.objects.create(
            workspace=self.scenario.workspace,
            expiration_delay=dt.timedelta(days=1),
        )
        Artifact.objects.create(
            workspace=self.scenario.workspace, expiration_delay=dt.timedelta(0)
        )

        self.assertQuerySetEqual(
            Artifact.objects.expired(timezone.now()), {artifact_1}
        )

    def test_in_current_scope(self) -> None:
        """Test the in_current_scope() QuerySet filter."""
        artifact, _ = self.playground.create_artifact(
            workspace=self.scenario.workspace
        )

        with context.local():
            context.set_permission_context_scoped(
                self.scenario.scope, PermissionContext.for_ui_anonymous()
            )
            self.assertQuerySetEqual(
                Artifact.objects.in_current_scope(),
                [artifact],
            )

        scope1 = Scope.objects.create(name="Scope1")
        with context.local():
            context.set_permission_context_scoped(
                scope1, PermissionContext.for_ui_anonymous()
            )
            self.assertQuerySetEqual(
                Artifact.objects.in_current_scope(),
                [],
            )

    def test_in_current_scope_no_context_scope(self) -> None:
        """Test the in_current_scope() QuerySet filter without scope set."""
        with self.assertRaisesRegex(
            ContextConsistencyError, "scope is not set"
        ):
            Artifact.objects.in_current_scope()

    def test_in_current_workspace(self) -> None:
        """Test the in_current_workspace() QuerySet filter."""
        artifact, _ = self.playground.create_artifact(
            workspace=self.scenario.workspace
        )
        context.set_permission_context_scoped(
            self.scenario.scope, self.scenario.context_ui()
        )

        with context.local():
            self.scenario.workspace.set_current()
            self.assertQuerySetEqual(
                Artifact.objects.in_current_workspace(),
                [artifact],
            )

        workspace1 = self.playground.create_workspace(name="other", public=True)
        with context.local():
            workspace1.set_current()
            self.assertQuerySetEqual(
                Artifact.objects.in_current_workspace(),
                [],
            )

    def test_in_current_workspace_no_context_workspace(self) -> None:
        """Test in_current_workspace() without workspace set."""
        with self.assertRaisesRegex(
            ContextConsistencyError, "workspace is not set"
        ):
            Artifact.objects.in_current_workspace()

    def test_visible_to_current_workspace(self) -> None:
        """Test the visible_to_current_workspace() QuerySet filter."""
        artifact, _ = self.playground.create_artifact(
            workspace=self.scenario.workspace
        )

        with context.local():
            context.set_permission_context_scoped(
                self.scenario.scope, self.scenario.context_ui()
            )

            with self.subTest(workspace="direct"), context.local():
                self.scenario.workspace.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [artifact],
                )

            workspace1 = self.playground.create_workspace(
                name="other", public=True
            )
            with (
                self.subTest(workspace="other", inherits=False),
                context.local(),
            ):
                workspace1.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [],
                )
            workspace1.set_inheritance([self.scenario.workspace])
            with (
                self.subTest(workspace="other", inherits=True),
                context.local(),
            ):
                workspace1.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [artifact],
                )

            grandchild_workspace = self.playground.create_workspace(
                name="grandchild", public=True
            )
            grandchild_workspace.set_inheritance([workspace1])
            with self.subTest(workspace="grandchild"), context.local():
                grandchild_workspace.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [artifact],
                )

            workspace2 = self.playground.create_workspace(
                name="multiple-children", public=True
            )
            workspace2_artifact, _ = self.playground.create_artifact(
                workspace=workspace2
            )
            for i in range(2):
                child_workspace = self.playground.create_workspace(
                    name=f"child-{i}", public=True
                )
                child_workspace.set_inheritance([workspace2])
            with self.subTest(workspace="multiple-children"), context.local():
                workspace2.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [workspace2_artifact],
                )

        scope1 = Scope.objects.create(name="Scope1")
        workspace2 = self.playground.create_workspace(
            name="workspace2", scope=scope1, public=True
        )

        with context.local():
            context.set_permission_context_scoped(
                scope1, self.scenario.context_ui()
            )
            workspace2.set_current()
            with self.subTest(scope="other", inherits=False):
                self.assertQuerySetEqual(
                    Artifact.objects.in_current_scope(),
                    [],
                )
            workspace2.set_inheritance([self.scenario.workspace])
            with self.subTest(scope="other", inherits=True):
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [artifact],
                )

    def test_visible_to_current_workspace_work_request_token(self) -> None:
        private_workspace = self.playground.create_workspace(
            name="private", public=False
        )
        work_request = self.playground.create_worker_task(
            assign_new_worker=True, mark_running=True
        )
        self.playground.create_group_role(
            private_workspace,
            Workspace.Roles.VIEWER,
            users=[work_request.created_by],
        )
        token = self.playground.create_work_request_token(work_request)

        public_artifact, _ = self.playground.create_artifact(
            workspace=self.scenario.workspace
        )
        private_artifact, _ = self.playground.create_artifact(
            workspace=private_workspace
        )

        with context.local():
            context.set_permission_context_scoped(
                self.scenario.scope,
                PermissionContext.for_work_request(token=token),
            )

            with self.subTest(workspace="public"), context.local():
                self.scenario.workspace.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [public_artifact],
                )

            with (
                self.subTest(workspace="private", inherits=False),
                context.local(),
            ):
                private_workspace.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [private_artifact],
                )

            private_workspace.set_inheritance([self.scenario.workspace])
            with (
                self.subTest(workspace="private", inherits=True),
                context.local(),
            ):
                private_workspace.set_current()
                self.assertQuerySetEqual(
                    Artifact.objects.visible_to_current_workspace(),
                    [public_artifact, private_artifact],
                    ordered=False,
                )


class ArtifactTests(TestCase):
    """Tests for the Artifact class."""

    scenario = DefaultContext()

    def assertPermissionDelegatesToWorkspace(
        self, artifact: Artifact, perm_name: str, workspace_perm_name: str
    ) -> None:
        pred_check = getattr(artifact, perm_name)
        print(pred_check, pred_check.anonymous)
        allows_anonymous = pred_check.anonymous == Allow.PASS

        with override_permission(Workspace, workspace_perm_name, AllowAll):
            if allows_anonymous:
                self.assertPermission(
                    perm_name,
                    users=(AnonymousUser(), self.scenario.user),
                    allowed=artifact,
                )
            else:
                self.assertPermission(
                    perm_name,
                    users=AnonymousUser(),
                    denied=artifact,
                )
                self.assertPermission(
                    perm_name,
                    users=self.scenario.user,
                    allowed=artifact,
                )
        with override_permission(Workspace, workspace_perm_name, DenyAll):
            self.assertPermission(
                perm_name,
                users=(AnonymousUser(), self.scenario.user),
                denied=artifact,
            )

    def test_save_permissions(self) -> None:
        """Test permission checks on save."""
        for user, username in (
            (None, "AnonymousUser"),
            (self.scenario.user, str(self.scenario.user)),
        ):
            with (
                self.subTest(user=str(user)),
                context.local(),
                override_permission(Workspace, "can_create_artifacts", DenyAll),
                self.assertRaisesRegex(
                    PermissionDenied,
                    (
                        rf"{username} cannot create artifacts in "
                        rf"{self.scenario.workspace}"
                    ),
                ),
            ):
                if user:
                    context.set_permission_context_scoped(
                        self.scenario.scope, PermissionContext.for_ui_user(user)
                    )
                else:
                    context.set_permission_context_scoped(
                        self.scenario.scope,
                        PermissionContext.for_ui_anonymous(),
                    )
                Artifact.objects.create(workspace=self.scenario.workspace)

    @context.disable_permission_checks()
    def test_default_values_of_fields(self) -> None:
        """Define the fields of the models and test their default values."""
        artifact = Artifact()

        self.assertEqual(artifact.category, "")

        artifact.category = "sample-type"

        self.assertEqual(artifact.data, {})

        artifact.workspace = self.playground.create_workspace(name="test")

        artifact.clean_fields()

        date_time_before_save = timezone.now()

        artifact.save()

        # date_time_before_save <= artifact.created_at <= timezone.now()
        self.assertLessEqual(date_time_before_save, artifact.created_at)
        self.assertLessEqual(artifact.created_at, timezone.now())

        # By default, the expiration_delay is None
        self.assertIsNone(artifact.expiration_delay)

        # By default, the created_by_work_request is None
        self.assertIsNone(artifact.created_by_work_request)

        # By default, the created_by is None
        self.assertIsNone(artifact.created_by)

    @context.disable_permission_checks()
    def test_expired(self) -> None:
        """Test Artifact.expired() method."""
        artifact = Artifact()

        artifact.created_at = timezone.now() - dt.timedelta(days=2)
        artifact.expiration_delay = dt.timedelta(days=1)
        self.assertTrue(artifact.expired(timezone.now()))

        artifact.expiration_delay = dt.timedelta(0)
        self.assertFalse(artifact.expired(timezone.now()))

        artifact.created_at = timezone.now()
        artifact.expiration_delay = dt.timedelta(days=1)
        self.assertFalse(artifact.expired(timezone.now()))

    @context.disable_permission_checks()
    def test_default_expiration(self) -> None:
        """Test workspace's default_expiration_delay."""
        self.assertEqual(
            self.scenario.workspace.default_expiration_delay, dt.timedelta(0)
        )
        artifact = Artifact.objects.create(workspace=self.scenario.workspace)
        self.assertIsNone(artifact.expiration_delay)

        expiring_ws = self.playground.create_workspace(name="test")
        expiring_ws.default_expiration_delay = dt.timedelta(days=7)
        test_creation_time = timezone.now()
        artifact = Artifact.objects.create(workspace=expiring_ws)
        delta = expiring_ws.default_expiration_delay
        assert artifact.expire_at is not None
        self.assertEqual(
            (artifact.expire_at - test_creation_time).days, delta.days
        )

    @context.disable_permission_checks()
    def test_str(self) -> None:
        """Test for Artifact.__str__."""
        artifact = Artifact.objects.create(workspace=self.scenario.workspace)
        self.assertEqual(
            artifact.__str__(),
            f"Id: {artifact.id} "
            f"Category: {artifact.category} "
            f"Workspace: {artifact.workspace.id}",
        )

    def test_ui(self) -> None:
        from debusine.web.views.ui.artifacts import ArtifactUI

        artifact, _ = self.playground.create_artifact()
        request = RequestFactory().get("/")
        ui = artifact.ui(request)
        self.assertIsInstance(ui, ArtifactUI)

    @context.disable_permission_checks()
    def test_get_absolute_url(self) -> None:
        """Test for Artifact.get_absolute_url."""
        artifact = Artifact.objects.create(workspace=self.scenario.workspace)
        self.assertEqual(
            artifact.get_absolute_url(),
            f"/{self.scenario.scope.name}/{self.scenario.workspace.name}"
            f"/artifact/{artifact.pk}/",
        )

    @context.disable_permission_checks()
    def test_get_absolute_url_different_scope(self) -> None:
        """Artifact.get_absolute_url works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        artifact = Artifact.objects.create(workspace=other_workspace)
        self.assertEqual(
            artifact.get_absolute_url(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/artifact/{artifact.pk}/",
        )

    @context.disable_permission_checks()
    def test_get_absolute_url_download(self) -> None:
        """Test for Artifact.get_absolute_url_download."""
        artifact = Artifact.objects.create(workspace=self.scenario.workspace)
        self.assertEqual(
            artifact.get_absolute_url_download(),
            f"/{self.scenario.scope.name}/{self.scenario.workspace.name}"
            f"/artifact/{artifact.pk}/download/",
        )

    @context.disable_permission_checks()
    def test_get_absolute_url_download_different_scope(self) -> None:
        """Artifact.get_absolute_url_download works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        artifact = Artifact.objects.create(workspace=other_workspace)
        self.assertEqual(
            artifact.get_absolute_url_download(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/artifact/{artifact.pk}/download/",
        )

    def test_data_validation_error(self) -> None:
        """Assert that ValidationError is raised if data is not valid."""
        with context.disable_permission_checks():
            artifact = Artifact.objects.create(
                category=ArtifactCategory.WORK_REQUEST_DEBUG_LOGS,
                workspace=self.scenario.workspace,
                data=[],
            )
        with self.assertRaises(ValidationError) as raised:
            artifact.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {"data": ["data must be a dictionary"]},
        )

        artifact.category = "missing:does-not-exist"
        artifact.data = {}
        with self.assertRaises(ValidationError) as raised:
            artifact.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {"category": ["missing:does-not-exist: invalid artifact category"]},
        )

        artifact.category = ArtifactCategory.PACKAGE_BUILD_LOG
        with self.assertRaises(ValidationError) as raised:
            artifact.full_clean()
        messages = raised.exception.message_dict
        self.assertCountEqual(messages.keys(), ["data"])
        self.assertRegex(
            messages["data"][0],
            r"invalid artifact data:"
            r" 3 validation errors for DebianPackageBuildLog\n",
        )

    @context.disable_permission_checks()
    def test_get_label_from_data(self) -> None:
        """Test getting artifact label from its data."""
        artifact = Artifact.objects.create(
            category=ArtifactCategory.SOURCE_PACKAGE,
            workspace=self.scenario.workspace,
            data={
                "name": "test",
                "version": "1.0-1",
                "type": "dpkg",
                "dsc_fields": {},
            },
        )
        self.assertEqual(artifact.get_label(), "test_1.0-1")

    @context.disable_permission_checks()
    def test_get_label_from_category(self) -> None:
        """Test defaults for artifact label."""
        artifact = Artifact.objects.create(
            category=ArtifactCategory.WORK_REQUEST_DEBUG_LOGS,
            workspace=self.scenario.workspace,
            data={},
        )
        self.assertEqual(
            artifact.get_label(), "debusine:work-request-debug-logs"
        )

    @context.disable_permission_checks()
    def test_get_label_from_data_reuse_model(self) -> None:
        """Test getting artifact label reusing the model instance."""
        artifact = Artifact.objects.create(
            category=ArtifactCategory.SOURCE_PACKAGE,
            workspace=self.scenario.workspace,
            data={
                "name": "test",
                "version": "1.0-1",
                "type": "dpkg",
                "dsc_fields": {},
            },
        )
        self.assertEqual(
            artifact.get_label(EmptyArtifactData()), "debian:source-package"
        )

    def test_can_display_delegate_to_workspace(self) -> None:
        """Test the can_display predicate on public workspaces."""
        artifact, _ = self.playground.create_artifact()
        self.assertPermissionDelegatesToWorkspace(
            artifact, "can_display", "can_display"
        )

    def test_can_display_work_request_created_artifacts(self) -> None:
        """Test the can_display predicate on public workspaces."""
        work_request = self.playground.create_worker_task(mark_running=True)
        token = self.playground.create_work_request_token(work_request)

        artifact, _ = self.playground.create_artifact()
        artifact.created_by_work_request = work_request
        artifact.save()

        other, _ = self.playground.create_artifact()

        with override_permission(Workspace, "can_display", AllowAll):
            self.assertPermission(
                "can_display",
                users=[AnonymousUser(), self.scenario.user],
                allowed=artifact,
                denied=other,
                token=token,
            )
        with override_permission(Workspace, "can_display", DenyAll):
            self.assertPermission(
                "can_display",
                users=(AnonymousUser(), self.scenario.user),
                denied=[artifact, other],
            )

    def test_can_display_work_request_input_allowlist(self) -> None:
        """Test the can_display predicate on public workspaces."""
        task_input, _ = self.playground.create_artifact()
        other, _ = self.playground.create_artifact()
        work_request = self.playground.create_worker_task(mark_running=True)
        token = self.playground.create_work_request_token(work_request)
        self.enterContext(
            mock.patch(
                "debusine.tasks.models.BaseDynamicTaskData"
                ".get_input_artifacts_ids",
                return_value=[task_input.pk],
            )
        )
        with override_permission(Workspace, "can_display", AllowAll):
            self.assertPermission(
                "can_display",
                users=[AnonymousUser(), self.scenario.user],
                allowed=task_input,
                denied=other,
                token=token,
            )
        with override_permission(Workspace, "can_display", DenyAll):
            self.assertPermission(
                "can_display",
                users=(AnonymousUser(), self.scenario.user),
                denied=[task_input, other],
            )

    def test_can_add_files_delegate_to_workspace(self) -> None:
        artifact, _ = self.playground.create_artifact()
        self.assertPermissionDelegatesToWorkspace(
            artifact, "can_add_files", "can_create_artifacts"
        )

    def test_can_add_files_work_request(self) -> None:
        work_request = self.playground.create_worker_task(mark_running=True)
        token = self.playground.create_work_request_token(work_request)

        artifact, _ = self.playground.create_artifact()
        artifact.created_by_work_request = work_request
        artifact.save()

        other, _ = self.playground.create_artifact()

        with override_permission(Workspace, "can_create_artifacts", AllowAll):
            self.assertPermission(
                "can_add_files",
                users=AnonymousUser(),
                denied=[artifact, other],
            )
            self.assertPermission(
                "can_add_files",
                users=[self.scenario.user],
                allowed=artifact,
                denied=other,
                token=token,
            )
        with override_permission(Workspace, "can_create_artifacts", DenyAll):
            self.assertPermission(
                "can_add_files",
                users=(AnonymousUser(), self.scenario.user),
                denied=[artifact, other],
            )

    def test_can_add_relation_delegate_to_workspace(self) -> None:
        artifact, _ = self.playground.create_artifact()
        self.assertPermissionDelegatesToWorkspace(
            artifact, "can_add_relation", "can_create_artifacts"
        )

    def test_can_add_relation_work_request(self) -> None:
        work_request = self.playground.create_worker_task(mark_running=True)
        token = self.playground.create_work_request_token(work_request)

        artifact, _ = self.playground.create_artifact()
        artifact.created_by_work_request = work_request
        artifact.save()

        other, _ = self.playground.create_artifact()

        with override_permission(Workspace, "can_create_artifacts", AllowAll):
            self.assertPermission(
                "can_add_relation",
                users=AnonymousUser(),
                denied=[artifact, other],
            )
            self.assertPermission(
                "can_add_relation",
                users=[self.scenario.user],
                allowed=artifact,
                denied=other,
                token=token,
            )
        with override_permission(Workspace, "can_create_artifacts", DenyAll):
            self.assertPermission(
                "can_add_relation",
                users=(AnonymousUser(), self.scenario.user),
                denied=[artifact, other],
            )


class FileInArtifactTests(TestCase):
    """Tests for the FileInArtifact class."""

    def create_file_in_artifact(
        self,
        fileobj: File | None = None,
        workspace: Workspace | None = None,
        artifact: Artifact | None = None,
        path: str | None = None,
        complete: bool = False,
    ) -> FileInArtifact:
        """Return FileInArtifact to be used in tests."""
        if fileobj is None:
            fileobj = self.playground.create_file()

        if artifact is None:
            artifact, _ = self.playground.create_artifact(workspace=workspace)

        if path is None:
            path = "/usr/bin/test"

        return FileInArtifact.objects.create(
            artifact=artifact,
            file=fileobj,
            path=path,
            complete=complete,
        )

    def test_artifact_path_unique_constraint(self) -> None:
        """Test two FileInArtifact cannot have the same artifact and path."""
        artifact, _ = self.playground.create_artifact()

        file1 = self.playground.create_file(b"contents1")
        file2 = self.playground.create_file(b"contents2")

        self.create_file_in_artifact(
            file1, artifact=artifact, path="/usr/bin/test"
        )

        with self.assertRaisesRegex(
            django.db.utils.IntegrityError,
            "db_fileinartifact_unique_artifact_path",
        ):
            self.create_file_in_artifact(
                file2, artifact=artifact, path="/usr/bin/test"
            )

    def test_str(self) -> None:
        """Test FileInArtifact.__str__."""
        file_in_artifact = self.create_file_in_artifact()

        self.assertEqual(
            file_in_artifact.__str__(),
            f"Id: {file_in_artifact.id} "
            f"Artifact: {file_in_artifact.artifact.id} "
            f"Path: {file_in_artifact.path} "
            f"File: {file_in_artifact.file.id}",
        )

    def test_ui(self) -> None:
        from debusine.web.views.ui.artifacts import FileInArtifactUI

        file_in_artifact = self.create_file_in_artifact()
        request = RequestFactory().get("/")
        ui = file_in_artifact.ui(request)
        self.assertIsInstance(ui, FileInArtifactUI)

    def test_get_absolute_url(self) -> None:
        """Test FileInArtifact.get_absolute_url."""
        file_in_artifact = self.create_file_in_artifact()
        self.assertEqual(
            file_in_artifact.get_absolute_url(),
            f"/debusine/System/artifact"
            f"/{file_in_artifact.artifact.id}/file//usr/bin/test",
        )

    def test_get_absolute_url_different_scope(self) -> None:
        """FileInArtifact.get_absolute_url works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        file_in_artifact = self.create_file_in_artifact(
            workspace=other_workspace
        )
        self.assertEqual(
            file_in_artifact.get_absolute_url(),
            f"/{other_scope.name}/{other_workspace.name}/artifact"
            f"/{file_in_artifact.artifact.id}/file//usr/bin/test",
        )

    def test_get_absolute_url_raw(self) -> None:
        """Test FileInArtifact.get_absolute_url_raw."""
        file_in_artifact = self.create_file_in_artifact()
        self.assertEqual(
            file_in_artifact.get_absolute_url_raw(),
            f"/debusine/System/artifact"
            f"/{file_in_artifact.artifact.id}/raw//usr/bin/test",
        )

    def test_get_absolute_url_raw_different_scope(self) -> None:
        """FileInArtifact.get_absolute_url_raw works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        file_in_artifact = self.create_file_in_artifact(
            workspace=other_workspace
        )
        self.assertEqual(
            file_in_artifact.get_absolute_url_raw(),
            f"/{other_scope.name}/{other_workspace.name}/artifact"
            f"/{file_in_artifact.artifact.id}/raw//usr/bin/test",
        )

    def test_get_absolute_url_download(self) -> None:
        """Test FileInArtifact.get_absolute_url_download."""
        file_in_artifact = self.create_file_in_artifact()
        self.assertEqual(
            file_in_artifact.get_absolute_url_download(),
            f"/debusine/System/artifact"
            f"/{file_in_artifact.artifact.id}/download//usr/bin/test",
        )

    def test_get_absolute_url_download_different_scope(self) -> None:
        """FileInArtifact.get_absolute_url_download works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        file_in_artifact = self.create_file_in_artifact(
            workspace=other_workspace
        )
        self.assertEqual(
            file_in_artifact.get_absolute_url_download(),
            f"/{other_scope.name}/{other_workspace.name}/artifact"
            f"/{file_in_artifact.artifact.id}/download//usr/bin/test",
        )


class FileUploadTests(TestCase):
    """Tests for FileUpload class."""

    def setUp(self) -> None:
        """Set up basic objects for the tests."""
        super().setUp()
        self.file_upload = self.playground.create_file_upload()
        self.artifact = self.file_upload.file_in_artifact.artifact
        self.workspace = self.artifact.workspace
        self.file_path = self.file_upload.file_in_artifact.path
        self.file_size = self.file_upload.file_in_artifact.file.size

    def test_current_size_raise_no_file_in_artifact(self) -> None:
        """Test current_size() raise ValueError (FileInArtifact not found)."""
        wrong_path = "no-exist"
        with self.assertRaisesRegex(
            ValueError,
            f'^No FileInArtifact for Artifact {self.artifact.id} '
            f'and path "{wrong_path}"$',
        ):
            FileUpload.current_size(self.artifact, wrong_path)

    @context.disable_permission_checks()
    def test_current_size_raise_no_fileupload_for_file_in_artifact(
        self,
    ) -> None:
        """Test current_size() raise ValueError (FileUpload not found)."""
        artifact = Artifact.objects.create(
            category="test", workspace=self.workspace
        )
        file_in_artifact = FileInArtifact.objects.create(
            artifact=artifact,
            path="something",
            file=self.playground.create_file(),
        )

        with self.assertRaisesRegex(
            ValueError,
            f"^No FileUpload for FileInArtifact {file_in_artifact.id}$",
        ):
            FileUpload.current_size(artifact, "something")

    def test_current_size_return_last_position_received(self) -> None:
        """Test current_size() method return size of the file."""
        write_to_position = 30
        self.file_upload.absolute_file_path().write_bytes(
            next(data_generator(write_to_position))
        )
        self.assertEqual(
            FileUpload.current_size(
                artifact=self.artifact, path_in_artifact=self.file_path
            ),
            write_to_position,
        )

    def test_no_two_fileobj_to_same_path(self) -> None:
        """Test cannot create two FileUpload with same path."""
        data = next(data_generator(self.file_size))
        file = self.playground.create_file(data)
        new_file_in_artifact = FileInArtifact.objects.create(
            artifact=self.artifact, path="README-2", file=file
        )

        with self.assertRaisesRegex(
            django.db.utils.IntegrityError, "db_fileupload_path_key"
        ):
            FileUpload.objects.create(
                file_in_artifact=new_file_in_artifact,
                path=self.file_upload.path,
            )

    def test_delete(self) -> None:
        """Test FileUpload delete() try to unlink file."""
        file_path = self.file_upload.absolute_file_path()
        Path(file_path).write_bytes(next(data_generator(self.file_size)))

        self.assertTrue(file_path.exists())

        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            self.file_upload.delete()

        self.assertFalse(file_path.exists())

        self.assertEqual(len(callbacks), 1)

    def test_delete_file_does_not_exist(self) -> None:
        """Test FileUpload delete() try to unlink file but did not exist."""
        file_path = self.file_upload.absolute_file_path()

        self.assertFalse(Path(file_path).exists())

        with self.captureOnCommitCallbacks(execute=True) as callbacks:
            self.file_upload.delete()

        self.assertEqual(len(callbacks), 1)

    def test_absolute_file_path(self) -> None:
        """Test absolute_file_path() is in DEBUSINE_UPLOAD_DIRECTORY."""
        self.assertEqual(
            self.file_upload.absolute_file_path(),
            Path(settings.DEBUSINE_UPLOAD_DIRECTORY) / self.file_upload.path,
        )

    def test_str(self) -> None:
        """Test __str__."""
        self.assertEqual(self.file_upload.__str__(), f"{self.file_upload.id}")


class ArtifactRelationManagerTests(TestCase):
    """Tests for the ArtifactRelationManager class."""

    scenario = DefaultContext()

    def test_in_current_scope(self) -> None:
        """Test the in_current_scope() QuerySet filter."""
        artifact, _ = self.playground.create_artifact(
            workspace=self.scenario.workspace
        )
        target, _ = self.playground.create_artifact(
            workspace=self.scenario.workspace
        )
        relation = self.playground.create_artifact_relation(
            artifact=artifact, target=target
        )

        with context.local():
            self.scenario.set_current()
            self.assertQuerySetEqual(
                ArtifactRelation.objects.in_current_scope(), [relation]
            )

        scope1 = Scope.objects.create(name="Scope1")
        with context.local():
            context.set_permission_context_scoped(
                scope1, PermissionContext.for_ui_anonymous()
            )
            self.assertQuerySetEqual(
                ArtifactRelation.objects.in_current_scope(), []
            )

    def test_in_current_scope_no_context_scope(self) -> None:
        """Test the in_current_scope() QuerySet filter without scope set."""
        with self.assertRaisesRegex(
            ContextConsistencyError, "scope is not set"
        ):
            ArtifactRelation.objects.in_current_scope()


class ArtifactRelationTests(TestCase):
    """Implement ArtifactRelation tests."""

    scenario = DefaultContext()

    def setUp(self) -> None:
        """Initialize test object."""
        super().setUp()
        self.artifact_1, _ = self.playground.create_artifact()
        self.artifact_2, _ = self.playground.create_artifact(
            workspace=self.playground.create_workspace(name="other-workspace")
        )

    def test_type_not_valid_raise_validation_error(self) -> None:
        """Assert that ValidationError is raised if type is not valid."""
        artifact_relation = ArtifactRelation.objects.create(
            artifact=self.artifact_1,
            target=self.artifact_2,
            type="wrong-type",
        )
        with self.assertRaises(ValidationError):
            artifact_relation.full_clean()

    def test_str(self) -> None:
        """Assert __str__ return expected value."""
        artifact_relation = self.playground.create_artifact_relation(
            artifact=self.artifact_1, target=self.artifact_2
        )
        self.assertEqual(
            str(artifact_relation),
            f"{self.artifact_1.id} {ArtifactRelation.Relations.RELATES_TO} "
            f"{self.artifact_2.id}",
        )

    def test_can_display_delegate_to_workspace(self) -> None:
        """Test the can_display predicate on public workspaces."""
        artifact_relation = self.playground.create_artifact_relation(
            artifact=self.artifact_1, target=self.artifact_2
        )
        with override_permission(Workspace, "can_display", AllowAll):
            self.assertPermission(
                "can_display",
                users=(AnonymousUser(), self.scenario.user),
                allowed=artifact_relation,
            )
        for allowed_artifact in (self.artifact_1, self.artifact_2):
            with override_permission(
                Workspace,
                "can_display",
                ListFilter,
                include=[allowed_artifact.workspace],
            ):
                self.assertPermission(
                    "can_display",
                    users=(AnonymousUser(), self.scenario.user),
                    denied=artifact_relation,
                )
        with override_permission(Workspace, "can_display", DenyAll):
            self.assertPermission(
                "can_display",
                users=(AnonymousUser(), self.scenario.user),
                denied=artifact_relation,
            )

    def test_can_display_both_artifacts_same_wr(self) -> None:
        wr = self.playground.create_worker_task()
        self.artifact_1.created_by_work_request = wr
        self.artifact_1.save()
        self.artifact_2.workspace = self.scenario.workspace
        self.artifact_2.created_by_work_request = wr
        self.artifact_2.save()
        artifact_relation = self.playground.create_artifact_relation(
            artifact=self.artifact_1, target=self.artifact_2
        )
        pc = PermissionContext.for_work_request(work_request=wr)
        self.assertTrue(artifact_relation.can_display(pc))
        self.assertQuerySetEqual(
            ArtifactRelation.objects.can_display(pc), [artifact_relation]
        )

    def test_can_display_both_artifacts_as_input(self) -> None:
        wr = self.playground.create_worker_task()
        self.artifact_1.created_by_work_request = wr
        self.artifact_1.save()
        self.artifact_2.workspace = self.scenario.workspace
        self.artifact_2.save()
        self.enterContext(
            mock.patch(
                "debusine.tasks.BaseExternalTask.get_input_artifacts_ids",
                return_value=[self.artifact_1.pk, self.artifact_2.pk],
            )
        )
        artifact_relation = self.playground.create_artifact_relation(
            artifact=self.artifact_1, target=self.artifact_2
        )
        pc = PermissionContext.for_work_request(work_request=wr)
        self.assertTrue(artifact_relation.can_display(pc))
        self.assertQuerySetEqual(
            ArtifactRelation.objects.can_display(pc), [artifact_relation]
        )

    def test_can_delete(self) -> None:
        """Test the can_delete predicate."""
        artifact_relation = self.playground.create_artifact_relation(
            artifact=self.artifact_1, target=self.artifact_2
        )
        # Requires OWNER access in the artifact's workspace:
        with (
            self.subTest(workspace_owner=False, can_display=True),
            override_permission(Workspace, "can_display", DenyAll),
        ):
            self.assertPermission(
                "can_delete",
                users=(AnonymousUser(), self.scenario.user),
                denied=artifact_relation,
            )

        self.playground.create_group_role(
            self.artifact_1.workspace,
            Workspace.Roles.OWNER,
            users=[self.scenario.user],
        )

        # Also requires can_display
        with self.subTest(can_display=False):
            self.assertPermission(
                "can_delete",
                users=(AnonymousUser(), self.scenario.user),
                denied=artifact_relation,
            )
        with (
            self.subTest(can_display=True),
            override_permission(Workspace, "can_display", AllowAll),
        ):
            self.assertPermission(
                "can_delete",
                users=AnonymousUser(),
                denied=artifact_relation,
            )
            self.assertPermission(
                "can_delete",
                users=self.scenario.user,
                allowed=artifact_relation,
            )
