# 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 workflow views."""

import re
from typing import Any, ClassVar, assert_never, override
from unittest import mock

from django.conf import settings
from django.db.models import Max
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient

from debusine.artifacts.models import ArtifactCategory, TaskTypes
from debusine.client.models import (
    Link,
    RuntimeParameter,
    RuntimeParameters,
    StrictBaseModel,
    WorkflowTemplateData,
    WorkflowTemplateDataNew,
)
from debusine.db.models import (
    Scope,
    WorkRequest,
    WorkflowTemplate,
    Workspace,
    default_workspace,
)
from debusine.db.playground import scenarios
from debusine.server.scopes import urlconf_scope
from debusine.server.serializers import WorkRequestSerializer
from debusine.server.views.tests.base import TestCase
from debusine.tasks import TaskConfigError
from debusine.test.django import (
    AllowAll,
    DenyAll,
    TestResponseType,
    override_permission,
)


class WorkflowTemplateViewSetTests(TestCase):
    """Tests for :py:class:`WorkflowTemplateViewSet`."""

    scenario = scenarios.DefaultContextAPI()
    sample: ClassVar[WorkflowTemplate]
    sample_client: ClassVar[WorkflowTemplateData]
    sample_client_with_links: ClassVar[WorkflowTemplateData]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.sample = cls.playground.create_workflow_template(
            "sample", task_name="noop"
        )
        cls.sample_client = WorkflowTemplateData(
            id=cls.sample.id,
            name=cls.sample.name,
            task_name=cls.sample.task_name,
            priority=cls.sample.priority,
            static_parameters=cls.sample.static_parameters,
            runtime_parameters=cls.sample.runtime_parameters,
        )
        cls.sample_client_with_links = cls.sample_client.model_copy()
        cls.sample_client_with_links.links = cls.get_links(cls.sample)

    @classmethod
    def get_links(
        cls, workflow_template: WorkflowTemplate
    ) -> dict[str, str | Link]:
        return {
            "self": reverse(
                "api:workflow-template-detail",
                kwargs={
                    "workspace": workflow_template.workspace.name,
                    "pk": workflow_template.pk,
                },
            ),
            "webui_self": workflow_template.get_absolute_url(),
        }

    def _workspace_url_kwarg(self, workspace: Workspace | str | None) -> str:
        match workspace:
            case None:
                return self.scenario.workspace.name
            case str():
                return workspace
            case _:
                return workspace.name

    def _request_data(
        self, data: dict[str, Any] | StrictBaseModel | None = None
    ) -> dict[str, Any]:
        match data:
            case dict():
                return data
            case StrictBaseModel():
                return data.model_dump(mode="json")
            case None:
                return {}
            case _ as unreachable:
                assert_never(unreachable)

    def _reverse_list(
        self,
        workspace: Workspace | str | None,
        **kwargs: str,
    ) -> str:
        """Build the URL for a test request."""
        kwargs["workspace"] = self._workspace_url_kwarg(workspace)
        return reverse("api:workflow-template-list", kwargs=kwargs)

    def _reverse_detail(
        self,
        workflow_template: int | str | WorkflowTemplate | None,
        workspace: Workspace | str | None,
        **kwargs: str,
    ) -> str:
        """Build the URL for a test request."""
        kwargs["workspace"] = self._workspace_url_kwarg(workspace)

        match workflow_template:
            case None:
                kwargs["pk"] = str(self.sample.pk)
            case int():
                kwargs["pk"] = str(workflow_template)
            case str():
                kwargs["pk"] = workflow_template
            case WorkflowTemplate():
                kwargs["pk"] = str(workflow_template.pk)
            case _ as unreachable:
                assert_never(unreachable)

        return reverse("api:workflow-template-detail", kwargs=kwargs)

    def _headers(self, user_token: bool) -> dict[str, str]:
        """Build headers for a test request."""
        match user_token:
            case True:
                return {"token": self.scenario.user_token.key}
            case False:
                return {}
            case _ as unreachable:
                assert_never(unreachable)

    def get(
        self,
        workflow_template: int | str | WorkflowTemplate | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for a workflow template."""
        return self.client.get(
            self._reverse_detail(workflow_template, workspace),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def list(
        self,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for the workflow template list."""
        return self.client.get(
            self._reverse_list(workspace),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def post(
        self,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | StrictBaseModel | None = None,
    ) -> TestResponseType:
        """POST request."""
        return self.client.post(
            self._reverse_list(workspace),
            data=self._request_data(data),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def put(
        self,
        workflow_template: int | WorkflowTemplate | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | StrictBaseModel | None = None,
    ) -> TestResponseType:
        """PUT request for a workflow template."""
        return self.client.put(
            self._reverse_detail(workflow_template, workspace),
            data=self._request_data(data),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def patch(
        self,
        workflow_template: int | WorkflowTemplate | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | StrictBaseModel | None = None,
    ) -> TestResponseType:
        """PATCH request for a workflow template."""
        return self.client.patch(
            self._reverse_detail(workflow_template, workspace),
            data=self._request_data(data),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def assertSampleUnchanged(self) -> None:
        """Ensure self.sample is unchanged."""
        self.sample.refresh_from_db()
        self.assertEqual(self.sample.name, "sample")
        self.assertEqual(self.sample.task_name, "noop")
        self.assertEqual(self.sample.priority, 0)
        self.assertEqual(self.sample.static_parameters, {})
        self.assertEqual(self.sample.runtime_parameters, RuntimeParameter.ANY)

    def test_unauthenticated(self) -> None:
        """Authentication is required."""
        for method in ("get", "list", "post", "put", "patch"):
            with self.subTest(method=method):
                response = getattr(self, method)(user_token=False)
                self.assertResponseProblem(
                    response,
                    "Error",
                    detail_pattern=(
                        "Authentication credentials were not provided."
                    ),
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_workspace_not_found(self) -> None:
        """Workspace must exist."""
        for method in ("get", "list", "post", "put", "patch"):
            with self.subTest(method=method):
                response = getattr(self, method)(workspace="does-not-exist")
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace does-not-exist not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_workspace_not_accessible(self) -> None:
        """User must be able to display the workspace."""
        workspace = self.playground.create_workspace(name="private")
        for method in ("get", "list", "post", "put", "patch"):
            with self.subTest(method=method):
                response = getattr(self, method)(workspace=workspace)
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace private not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_not_found(self) -> None:
        """The workflow template must exist."""
        max_id = WorkflowTemplate.objects.aggregate(Max('id'))['id__max']
        for method in ("get", "put", "patch"):
            with self.subTest(method=method):
                response = getattr(self, method)(workflow_template=max_id + 1)
                self.assertResponseProblem(
                    response,
                    "Object not found",
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_list_unauthorized(self) -> None:
        """The user must be able to display the collection."""
        with override_permission(WorkflowTemplate, "can_display", DenyAll):
            response = self.list()
            data = self.assertAPIResponseOk(response)
            self.assertEqual(
                data,
                {"count": 0, "next": None, "previous": None, "results": []},
            )

    def test_detail_unauthorized(self) -> None:
        """The user must be able to display the collection."""
        for method in ("get", "put", "patch"):
            with (
                self.subTest(method=method),
                override_permission(WorkflowTemplate, "can_display", DenyAll),
            ):
                response = getattr(self, method)()
                self.assertResponseProblem(
                    response,
                    "Object not found",
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_create_unauthorized(self) -> None:
        """The user must be able to create the collection."""
        with override_permission(Workspace, "can_configure", DenyAll):
            response = self.post(
                data={
                    "name": "test",
                    "task_name": "noop",
                    "static_parameters": {},
                    "priority": 0,
                }
            )
            self.assertResponseProblem(
                response,
                "playground cannot configure workspace"
                f" {self.scenario.workspace}",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_create_unauthorized_create_child_workspace(self) -> None:
        """Only scope owners may create ``create_child_workspace`` templates."""
        with (
            override_permission(Workspace, "can_configure", AllowAll),
            override_permission(Scope, "can_create_workspace", DenyAll),
        ):
            response = self.post(
                data=WorkflowTemplateDataNew(
                    name="test",
                    task_name="create_child_workspace",
                    static_parameters={},
                )
            )
            self.assertResponseProblem(
                response,
                f"playground cannot create workspaces in {self.scenario.scope}",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_update_unauthorized(self) -> None:
        """The user must be able to update the collection."""
        for method in ("put", "patch"):
            with (
                self.subTest(method=method),
                override_permission(Workspace, "can_configure", DenyAll),
            ):
                response = getattr(self, method)(
                    data=WorkflowTemplateDataNew(
                        name="test", task_name="noop", static_parameters={}
                    )
                )
                self.assertResponseProblem(
                    response,
                    "playground cannot configure sample",
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_update_unauthorized_create_child_workspace(self) -> None:
        """Only scope owners may update ``create_child_workspace`` templates."""
        template = self.playground.create_workflow_template(
            name="test", task_name="create_child_workspace"
        )

        for method in ("put", "patch"):
            with (
                self.subTest(method=method),
                override_permission(
                    WorkflowTemplate, "can_configure", AllowAll
                ),
                override_permission(Scope, "can_create_workspace", DenyAll),
            ):
                response = getattr(self, method)(
                    workflow_template=template,
                    data=WorkflowTemplateDataNew(
                        name="test",
                        task_name="create_child_workspace",
                        static_parameters={},
                    ),
                )
                self.assertResponseProblem(
                    response,
                    "playground cannot create workspaces in"
                    f" {self.scenario.scope}",
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_post_unauthorized(self) -> None:
        """The user must be able to create the workflow template."""
        with override_permission(Workspace, "can_configure", DenyAll):
            response = self.post(
                data=WorkflowTemplateDataNew(
                    name="test", task_name="noop", static_parameters={}
                )
            )
            self.assertResponseProblem(
                response,
                "playground cannot configure workspace debusine/System",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_patch_unauthorized(self) -> None:
        """The user must be able to update the workflow template."""
        with override_permission(Workspace, "can_configure", DenyAll):
            response = self.patch(
                data=WorkflowTemplateDataNew(
                    name="test", task_name="noop", static_parameters={}
                )
            )
            self.assertResponseProblem(
                response,
                "playground cannot configure sample",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_get_success(self) -> None:
        response = self.get()
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data,
            {
                "id": self.sample.id,
                "name": self.sample.name,
                "task_name": self.sample.task_name,
                "static_parameters": self.sample.static_parameters,
                "runtime_parameters": self.sample.runtime_parameters,
                "priority": self.sample.priority,
                "links": self.get_links(self.sample),
                "extra_groups": [],
                "restricted": False,
            },
        )

    def test_get_name_success(self) -> None:
        response = self.get(workflow_template="sample")
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data,
            {
                "id": self.sample.id,
                "name": self.sample.name,
                "task_name": self.sample.task_name,
                "static_parameters": self.sample.static_parameters,
                "runtime_parameters": self.sample.runtime_parameters,
                "priority": self.sample.priority,
                "links": self.get_links(self.sample),
                "extra_groups": [],
                "restricted": False,
            },
        )

    def test_get_honours_workspace(self) -> None:
        workspace = self.playground.create_workspace(
            name="workspace", public=True
        )
        template = WorkflowTemplate.objects.create(
            name=self.sample.name,
            workspace=workspace,
            task_name="noop",
            static_parameters={},
        )

        response = self.get(workspace=workspace, workflow_template=template)
        data = self.assertAPIResponseOk(response)
        self.assertEqual(data["id"], template.id)

        response = self.get(workflow_template=template)
        self.assertResponseProblem(
            response,
            title="Object not found",
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_list_success(self) -> None:
        response = self.list()
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data,
            {
                "count": 1,
                "next": None,
                "previous": None,
                "results": [
                    {
                        "id": self.sample.pk,
                        "name": "sample",
                        "priority": 0,
                        "static_parameters": {},
                        "runtime_parameters": RuntimeParameter.ANY,
                        "task_name": "noop",
                        "links": self.get_links(self.sample),
                        "extra_groups": [],
                        "restricted": False,
                    }
                ],
            },
        )

    def test_list_different_workspace(self) -> None:
        workspace = self.playground.create_workspace(name="other", public=True)
        other2 = self.playground.create_workflow_template(
            task_name="noop", name="other2", workspace=workspace
        )
        other1 = self.playground.create_workflow_template(
            task_name="noop", name="other1", workspace=workspace
        )
        response = self.list(workspace=workspace)
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data,
            {
                "count": 2,
                "next": None,
                "previous": None,
                "results": [
                    {
                        "id": other1.pk,
                        "name": other1.name,
                        "priority": 0,
                        "static_parameters": {},
                        "runtime_parameters": RuntimeParameter.ANY,
                        "task_name": "noop",
                        "links": self.get_links(other1),
                        "extra_groups": [],
                        "restricted": False,
                    },
                    {
                        "id": other2.pk,
                        "name": other2.name,
                        "priority": 0,
                        "static_parameters": {},
                        "runtime_parameters": RuntimeParameter.ANY,
                        "task_name": "noop",
                        "links": self.get_links(other2),
                        "extra_groups": [],
                        "restricted": False,
                    },
                ],
            },
        )

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_success(self) -> None:
        """Create a new workflow template."""
        static_parameters = {
            "target_distribution": "debian:bookworm",
            "architectures": ["amd64", "arm64"],
        }

        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="sbuild",
                static_parameters=static_parameters,
                runtime_parameters=RuntimeParameter.ANY,
            )
        )
        self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        template = WorkflowTemplate.objects.get(
            name="test", workspace=self.scenario.workspace
        )
        self.assertEqual(template.task_name, "sbuild")
        self.assertEqual(template.static_parameters, static_parameters)
        self.assertEqual(template.runtime_parameters, RuntimeParameter.ANY)

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_duplicate_name(self) -> None:
        self.playground.create_workflow_template("duplicate", task_name="noop")

        with self.assert_model_count_unchanged(WorkflowTemplate):
            response = self.post(
                data=WorkflowTemplateDataNew(
                    name="duplicate",
                    task_name="noop",
                    static_parameters={},
                )
            )

        self.assertResponseProblem(
            response,
            "Workflow template already exists",
            detail_pattern=(
                "A workflow template called 'duplicate' already exists"
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    @override_permission(Workspace, "can_configure", AllowAll)
    @override_permission(Scope, "can_create_workspace", AllowAll)
    def test_post_create_child_workspace(self) -> None:
        """Create a new ``create_child_workspace`` workflow template."""
        static_parameters = {"prefix": "experiment"}
        runtime_parameters: RuntimeParameters = {
            "suffix": RuntimeParameter.ANY,
            "public": RuntimeParameter.ANY,
            "owner_group": RuntimeParameter.ANY,
            "workflow_template_names": RuntimeParameter.ANY,
            "expiration_delay": RuntimeParameter.ANY,
        }

        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="create_child_workspace",
                static_parameters=static_parameters,
                runtime_parameters=runtime_parameters,
            )
        )
        self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        template = WorkflowTemplate.objects.get(
            name="test", workspace=self.scenario.workspace
        )
        self.assertEqual(template.task_name, "create_child_workspace")
        self.assertEqual(template.static_parameters, static_parameters)
        self.assertEqual(template.runtime_parameters, runtime_parameters)

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_different_workspace(self) -> None:
        workspace = self.playground.create_workspace(
            name="test-workspace", public=True
        )
        static_parameters = {"target_distribution": "debian:bookworm"}

        response = self.post(
            workspace=workspace,
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="sbuild",
                static_parameters=static_parameters,
            ),
        )
        self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        template = WorkflowTemplate.objects.get(
            name="test", workspace=workspace
        )
        self.assertEqual(template.task_name, "sbuild")
        self.assertEqual(template.static_parameters, static_parameters)

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_positive_priority_without_permissions(self) -> None:
        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="noop",
                static_parameters={},
                priority=1,
            ),
        )
        self.assertResponseProblem(
            response,
            "Error",
            detail_pattern="You are not permitted to set positive priorities",
            status_code=status.HTTP_403_FORBIDDEN,
        )

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_positive_priority_with_permissions(self) -> None:
        self.playground.add_user_permission(
            self.scenario.user, WorkRequest, "manage_workrequest_priorities"
        )
        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="noop",
                static_parameters={},
                priority=1,
            ),
        )
        self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        template = WorkflowTemplate.objects.get(
            name="test", workspace=self.scenario.workspace
        )
        self.assertEqual(template.priority, 1)

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_extra_groups(self) -> None:
        group1 = self.playground.create_group("group1")
        group2 = self.playground.create_group("group2")
        group1.users.add(self.scenario.user)
        group2.users.add(self.scenario.user)
        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="noop",
                static_parameters={},
                extra_groups=[group1.pk, str(group2)],
            ),
        )
        self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        template = WorkflowTemplate.objects.get(
            name="test", workspace=self.scenario.workspace
        )
        self.assertQuerySetEqual(
            template.extra_groups.all(), [group1, group2], ordered=False
        )

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_extra_groups_not_accessible(self) -> None:
        group1 = self.playground.create_group("group1")
        group2 = self.playground.create_group("group2")
        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="noop",
                static_parameters={},
                extra_groups=[group1.pk, str(group2)],
            ),
        )
        self.assertResponseProblem(
            response,
            "Error",
            detail_pattern=r"You are not allowed to grant"
            rf" extra group {re.escape(str(group1))}",
            status_code=status.HTTP_403_FORBIDDEN,
        )
        self.assertQuerySetEqual(
            WorkflowTemplate.objects.filter(name="test"), []
        )

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_post_restricted(self) -> None:
        response = self.post(
            data=WorkflowTemplateDataNew(
                name="test",
                task_name="noop",
                static_parameters={},
                restricted=True,
            ),
        )
        self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        template = WorkflowTemplate.objects.get(
            name="test", workspace=self.scenario.workspace
        )
        self.assertTrue(template.restricted)

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_noop(self) -> None:
        response = self.patch(data={})
        new_data = self.assertAPIResponseOk(response)
        assert WorkflowTemplateData(**new_data) == self.sample_client_with_links
        self.assertSampleUnchanged()

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_noop_all_set(self) -> None:
        response = self.patch(
            data={
                "name": "sample",
                "task_name": "noop",
                "priority": 0,
                "static_parameters": {},
                "runtime_parameters": RuntimeParameter.ANY,
            },
        )
        new_data = self.assertAPIResponseOk(response)
        assert WorkflowTemplateData(**new_data) == self.sample_client_with_links
        self.assertSampleUnchanged()

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_rename(self) -> None:
        response = self.patch(data={"name": "renamed"})
        new_data = self.assertAPIResponseOk(response)
        changed = self.sample_client.model_dump()
        changed["name"] = "renamed"
        self.sample.refresh_from_db()
        changed["links"] = self.get_links(self.sample)
        assert WorkflowTemplateData(**new_data).model_dump() == changed
        self.assertEqual(self.sample.name, "renamed")
        self.assertEqual(self.sample.task_name, "noop")
        self.assertEqual(self.sample.priority, 0)
        self.assertEqual(self.sample.static_parameters, {})

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_static_parameters(self) -> None:
        static_parameters = {"changed": True}
        response = self.patch(
            workflow_template=self.sample,
            data={"static_parameters": static_parameters},
        )
        self.assertAPIResponseOk(response)
        self.sample.refresh_from_db()
        self.assertEqual(self.sample.static_parameters, static_parameters)

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    @override_permission(Scope, "can_create_workspace", AllowAll)
    def test_patch_create_child_workspace(self) -> None:
        template = self.playground.create_workflow_template(
            "test",
            task_name="create_child_workspace",
            static_parameters={"prefix": "experiment"},
        )
        static_parameters = {"prefix": "new-experiment"}
        response = self.patch(
            workflow_template=template,
            data={"static_parameters": static_parameters},
        )
        self.assertAPIResponseOk(response)
        template.refresh_from_db()
        self.assertEqual(template.static_parameters, static_parameters)

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_positive_priority_without_permissions(self) -> None:
        response = self.patch(
            workflow_template=self.sample, data={"priority": 1}
        )
        self.assertResponseProblem(
            response,
            "Error",
            detail_pattern="You are not permitted to set positive priorities",
            status_code=status.HTTP_403_FORBIDDEN,
        )

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_positive_priority_with_permissions(self) -> None:
        self.playground.add_user_permission(
            self.scenario.user, WorkRequest, "manage_workrequest_priorities"
        )
        response = self.patch(
            workflow_template=self.sample, data={"priority": 1}
        )
        self.assertAPIResponseOk(response)
        self.sample.refresh_from_db()
        self.assertEqual(self.sample.priority, 1)

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_restricted(self) -> None:
        response = self.patch(
            workflow_template=self.sample, data={"restricted": True}
        )
        self.assertAPIResponseOk(response)
        self.sample.refresh_from_db()
        self.assertTrue(self.sample.restricted)

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_change_all(self) -> None:
        response = self.patch(
            data={
                "name": "renamed",
                "priority": -3,
                "task_name": "noop",
                "static_parameters": {"changed": True},
                "runtime_parameters": {},
                "restricted": True,
            }
        )
        new_data = self.assertAPIResponseOk(response)
        self.sample.refresh_from_db()
        assert new_data == {
            "id": self.sample.pk,
            "name": "renamed",
            "task_name": "noop",
            "priority": -3,
            "static_parameters": {"changed": True},
            "runtime_parameters": {},
            "links": self.get_links(self.sample),
            "extra_groups": [],
            "restricted": True,
        }
        self.assertEqual(self.sample.name, "renamed")
        self.assertEqual(self.sample.task_name, "noop")
        self.assertEqual(self.sample.priority, -3)
        self.assertEqual(self.sample.static_parameters, {"changed": True})
        self.assertEqual(self.sample.runtime_parameters, {})
        self.assertTrue(self.sample.restricted)

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_cannot_change_task_name(self) -> None:
        response = self.patch(
            workflow_template=self.sample, data={"task_name": "sbuild"}
        )
        self.assertResponseProblem(
            response,
            "The task name of a workflow template cannot be changed",
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        self.assertSampleUnchanged()

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_rename_conflict(self) -> None:
        self.playground.create_workflow_template("renamed", task_name="noop")

        response = self.patch(data={"name": "renamed"})
        self.assertResponseProblem(
            response,
            "New name conflicts with an existing workflow template",
            detail_pattern=(
                "A workflow template called 'renamed' already exists"
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        self.assertSampleUnchanged()

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_rename_no_conflict_in_other_workspaces(self) -> None:
        other = self.playground.create_workspace(name="other")
        self.playground.create_workflow_template(
            "renamed", task_name="noop", workspace=other
        )

        response = self.patch(data={"name": "renamed"})
        new_data = self.assertAPIResponseOk(response)
        changed = self.sample_client.model_copy()
        changed.name = "renamed"
        self.sample.refresh_from_db()
        changed.links = self.get_links(self.sample)
        assert WorkflowTemplateData(**new_data) == changed

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_extra_groups(self) -> None:
        group1 = self.playground.create_group("group1")
        group2 = self.playground.create_group("group2")
        group1.users.add(self.scenario.user)
        group2.users.add(self.scenario.user)
        response = self.patch(
            workflow_template=self.sample,
            data={"extra_groups": [group1.pk, str(group2)]},
        )
        new_data = self.assertAPIResponseOk(response)
        self.sample.refresh_from_db()
        assert new_data == {
            "id": self.sample.pk,
            "name": "sample",
            "task_name": "noop",
            "priority": 0,
            "static_parameters": {},
            "runtime_parameters": "any",
            "links": self.get_links(self.sample),
            "extra_groups": [str(group1), str(group2)],
            "restricted": False,
        }
        self.assertEqual(self.sample.name, "sample")
        self.assertEqual(self.sample.task_name, "noop")
        self.assertEqual(self.sample.priority, 0)
        self.assertEqual(self.sample.static_parameters, {})
        self.assertEqual(self.sample.runtime_parameters, "any")
        self.assertQuerySetEqual(
            self.sample.extra_groups.all(), [group1, group2], ordered=False
        )

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_patch_cannot_set_unaccessible_extra_groups(self) -> None:
        group = self.playground.create_group("group")
        response = self.patch(
            workflow_template=self.sample, data={"extra_groups": [group.pk]}
        )
        self.assertResponseProblem(
            response,
            "Error",
            detail_pattern=r"You are not allowed to grant"
            rf" extra group {re.escape(str(group))}",
            status_code=status.HTTP_403_FORBIDDEN,
        )
        self.assertSampleUnchanged()

    @override_permission(Workspace, "can_configure", AllowAll)
    def test_debusine_workflow_template_create(self) -> None:
        debusine = self.get_debusine()
        request = WorkflowTemplateDataNew(
            name="test",
            task_name="noop",
            static_parameters={},
            priority=0,
        )
        result = debusine.workflow_template_create(
            self.scenario.workspace.name, request
        )
        created = WorkflowTemplate.objects.get(
            workspace=self.scenario.workspace, name="test"
        )
        self.assertEqual(result.id, created.id)
        self.assertEqual(result.name, "test")
        self.assertEqual(result.static_parameters, {})
        self.assertEqual(result.priority, 0)

    def test_debusine_workflow_template_get_pk(self) -> None:
        debusine = self.get_debusine()
        result = debusine.workflow_template_get(
            self.scenario.workspace.name, self.sample.pk
        )
        self.assertEqual(result, self.sample_client_with_links)

    def test_debusine_workflow_template_get_name(self) -> None:
        debusine = self.get_debusine()
        result = debusine.workflow_template_get(
            self.scenario.workspace.name, self.sample.name
        )
        self.assertEqual(result, self.sample_client_with_links)

    def test_debusine_workflow_template_iter(self) -> None:
        debusine = self.get_debusine()
        result = debusine.workflow_template_iter(self.scenario.workspace.name)
        self.assertEqual([c.name for c in result], ["sample"])

    @override_permission(WorkflowTemplate, "can_configure", AllowAll)
    def test_debusine_workflow_template_update(self) -> None:
        debusine = self.get_debusine()
        changed = self.sample_client.model_copy()
        changed.name = "renamed"
        changed.priority = -2
        changed.static_parameters = {"changed": True}
        changed.runtime_parameters = RuntimeParameter.ANY
        result = debusine.workflow_template_update(
            self.scenario.workspace.name, changed
        )
        self.sample.refresh_from_db()
        changed_with_links = changed.model_copy()
        changed_with_links.links = self.get_links(self.sample)
        self.assertEqual(result, changed_with_links)
        self.assertEqual(self.sample.name, "renamed")
        self.assertEqual(self.sample.task_name, "noop")
        self.assertEqual(self.sample.priority, -2)
        self.assertEqual(self.sample.static_parameters, {"changed": True})
        self.assertEqual(self.sample.runtime_parameters, RuntimeParameter.ANY)


class WorkflowViewTests(TestCase):
    """Tests for WorkflowView."""

    scenario = scenarios.DefaultContextAPI()

    @override
    def setUp(self) -> None:
        super().setUp()
        self.client = APIClient()
        self.token = self.scenario.user_token

    def post_workflow(
        self, data: dict[str, Any], scope: str | None = None
    ) -> TestResponseType:
        """Post a workflow creation request to api:workflows."""
        headers = {"Token": self.token.key}
        if scope is not None:
            headers["X-Debusine-Scope"] = scope

        return self.client.post(
            reverse("api:workflows"), data=data, headers=headers, format="json"
        )

    def test_authentication_credentials_not_provided(self) -> None:
        """A Token is required to use the endpoint."""
        response = self.client.post(reverse("api:workflows"))
        self.assertResponseProblem(
            response,
            "Error",
            detail_pattern="Authentication credentials were not provided.",
            status_code=status.HTTP_403_FORBIDDEN,
        )

    def test_create_workflow_success(self) -> None:
        """Create a workflow."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        artifact = self.playground.create_source_artifact()
        for architecture in ("amd64", "arm64"):
            self.playground.create_debian_environment(
                codename="bookworm", architecture=architecture
            )

        WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name="sbuild",
            static_parameters={"architectures": ["amd64", "arm64"]},
            runtime_parameters=RuntimeParameter.ANY,
        )

        response = self.post_workflow(
            {
                "template_name": "test",
                "task_data": {
                    "input": {"source_artifact": artifact.id},
                    "target_distribution": "debian:bookworm",
                },
            }
        )

        data = self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        workflow = WorkRequest.objects.latest("created_at")
        self.assertEqual(data, WorkRequestSerializer(workflow).data)
        self.assertDictContainsAll(
            data,
            {
                "workspace": settings.DEBUSINE_DEFAULT_WORKSPACE,
                "created_by": self.token.user_id,
                "status": WorkRequest.Statuses.PENDING,
                "task_type": TaskTypes.WORKFLOW,
                "task_name": "sbuild",
                "task_data": {
                    "input": {"source_artifact": artifact.id},
                    "target_distribution": "debian:bookworm",
                    "architectures": ["amd64", "arm64"],
                },
                "priority_base": 0,
            },
        )

    def test_create_workflow_different_workspace(self) -> None:
        """Create a workflow in a non-default workspace."""
        workspace = self.playground.create_workspace(
            name="test-workspace", public=True
        )
        self.playground.create_group_role(
            workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        artifact = self.playground.create_source_artifact(workspace=workspace)
        for architecture in ("amd64", "arm64"):
            self.playground.create_debian_environment(
                workspace=workspace,
                codename="bookworm",
                architecture=architecture,
            )
        WorkflowTemplate.objects.create(
            name="test",
            workspace=workspace,
            task_name="sbuild",
            static_parameters={"architectures": ["amd64", "arm64"]},
            runtime_parameters=RuntimeParameter.ANY,
        )

        response = self.post_workflow(
            {
                "template_name": "test",
                "workspace": workspace.name,
                "task_data": {
                    "input": {"source_artifact": artifact.id},
                    "target_distribution": "debian:bookworm",
                },
            }
        )

        data = self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        workflow = WorkRequest.objects.latest("created_at")
        self.assertEqual(data, WorkRequestSerializer(workflow).data)
        self.assertEqual(workflow.workspace, workspace)

    def test_create_workflow_without_template_success(self) -> None:
        """Create a workflow without a template."""
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )
        artifact = self.playground.create_source_artifact()
        for architecture in ("amd64", "arm64"):
            self.playground.create_debian_environment(
                codename="bookworm", architecture=architecture
            )

        response = self.post_workflow(
            {
                "task_name": "sbuild",
                "task_data": {
                    "input": {"source_artifact": artifact.id},
                    "target_distribution": "debian:bookworm",
                    "architectures": ["amd64", "arm64"],
                },
            }
        )

        data = self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        workflow = WorkRequest.objects.latest("created_at")
        self.assertEqual(data, WorkRequestSerializer(workflow).data)
        self.assertDictContainsAll(
            data,
            {
                "workspace": settings.DEBUSINE_DEFAULT_WORKSPACE,
                "created_by": self.token.user_id,
                "status": WorkRequest.Statuses.PENDING,
                "task_type": TaskTypes.WORKFLOW,
                "task_name": "sbuild",
                "task_data": {
                    "input": {"source_artifact": artifact.id},
                    "target_distribution": "debian:bookworm",
                    "architectures": ["amd64", "arm64"],
                },
                "priority_base": 0,
            },
        )

    @override_permission(Scope, "can_create_workspace", AllowAll)
    def test_create_child_workspace_without_template_success(self) -> None:
        """Create a ``create_child_workspace`` workflow without a template."""
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )

        response = self.post_workflow(
            {
                "task_name": "create_child_workspace",
                "task_data": {"suffix": "test"},
            }
        )

        data = self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        workflow = WorkRequest.objects.latest("created_at")
        self.assertEqual(data, WorkRequestSerializer(workflow).data)
        self.assertDictContainsAll(
            data,
            {
                "workspace": settings.DEBUSINE_DEFAULT_WORKSPACE,
                "created_by": self.token.user_id,
                "status": WorkRequest.Statuses.PENDING,
                "task_type": TaskTypes.WORKFLOW,
                "task_name": "create_child_workspace",
                "task_data": {"suffix": "test"},
                "priority_base": 0,
            },
        )

    def test_create_workflow_nonexistent_template(self) -> None:
        """The view returns HTTP 404 if the workflow template does not exist."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        with self.assert_model_count_unchanged(WorkRequest):
            response = self.post_workflow(
                {"template_name": "test", "task_data": {}}
            )

        self.assertResponseProblem(
            response,
            "Workflow template not found",
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_create_workflow_invalid_task_data(self) -> None:
        """Creating a workflow validates task data."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        self.playground.create_workflow_template(name="noop", task_name="noop")

        with self.assert_model_count_unchanged(WorkRequest):
            response = self.post_workflow(
                {"template_name": "noop", "task_data": {"nonexistent": True}}
            )

        self.assertResponseProblem(
            response,
            "Cannot create workflow",
            detail_pattern=r"Extra inputs are not permitted",
        )

    def test_create_workflow_thorough_input_validation(self) -> None:
        """Creating a workflow does thorough validation of input data."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        self.playground.create_workflow_template(
            name="sbuild", task_name="sbuild"
        )
        artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST
        )

        with self.assert_model_count_unchanged(WorkRequest):
            response = self.post_workflow(
                {
                    "template_name": "sbuild",
                    "task_data": {
                        "input": {"source_artifact": artifact.id},
                        "target_distribution": "debian:bookworm",
                        "architectures": ["amd64"],
                    },
                }
            )

        self.assertResponseProblem(
            response,
            "Cannot create workflow",
            detail_pattern=(
                r"^input.source_artifact: unexpected artifact category: "
                r"'debusine:test'. Valid categories: "
                r"\['debian:source-package', 'debian:upload'\]$"
            ),
        )

    def test_create_workflow_template_validation_untruncated(self) -> None:
        """Pydantic validation messages are not truncated."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        self.playground.create_workflow_template(
            name="sbuild",
            task_name="sbuild",
            static_parameters={
                "very_long_unknown_parameter_name_foo_bar_baz_quux_corge": "1"
            },
        )

        response = self.post_workflow(
            {"template_name": "sbuild", "task_data": {}}
        )

        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
        # pydantic validation messages can get truncated if rended by str()
        self.assertNotIn("...", response.json()["detail"])

    def test_create_workflow_template_task_config_error(self) -> None:
        """non-Pydantic validation messages are handled correctly."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        self.playground.create_workflow_template(
            name="sbuild",
            task_name="sbuild",
            static_parameters={},
        )
        self.playground.create_debian_environment(
            codename="bookworm", architecture="amd64"
        )
        artifact = self.playground.create_source_artifact()

        with mock.patch.object(
            WorkRequest,
            "get_task",
            side_effect=TaskConfigError("Unknown failure"),
        ):
            response = self.post_workflow(
                {
                    "template_name": "sbuild",
                    "task_data": {
                        "input": {"source_artifact": artifact.id},
                        "target_distribution": "debian:bookworm",
                        "architectures": ["amd64"],
                    },
                }
            )

        self.assertResponseProblem(
            response,
            "Cannot create workflow",
            detail_pattern=r"Unknown failure",
        )

    def test_create_workflow_honours_scope(self) -> None:
        """Creating a workflow looks up workspace in current scope."""
        artifact = self.playground.create_source_artifact()
        for architecture in ("amd64", "i386"):
            self.playground.create_debian_environment(
                codename="bookworm", architecture=architecture
            )

        scope1 = self.playground.get_or_create_scope("scope1")
        scope2 = self.playground.get_or_create_scope("scope2")
        scope3 = self.playground.get_or_create_scope("scope3")
        workspace1 = self.playground.create_workspace(
            scope=scope1, name="common-name", public=True
        )
        workspace2 = self.playground.create_workspace(
            scope=scope2, name="common-name", public=True
        )
        for workspace, architectures in (
            (workspace1, ["amd64"]),
            (workspace2, ["i386"]),
        ):
            workspace.set_inheritance([self.scenario.workspace])
            self.playground.create_workflow_template(
                name="test",
                task_name="sbuild",
                workspace=workspace,
                static_parameters={"architectures": architectures},
            )
        task_data = {
            "input": {"source_artifact": artifact.id},
            "target_distribution": "debian:bookworm",
        }

        for workspace, architectures in (
            (workspace1, ["amd64"]),
            (workspace2, ["i386"]),
        ):
            self.playground.create_group_role(
                workspace,
                Workspace.Roles.CONTRIBUTOR,
                [self.scenario.user],
            )

            response = self.post_workflow(
                {
                    "template_name": "test",
                    "workspace": "common-name",
                    "task_data": task_data,
                },
                scope=workspace.scope.name,
            )

            data = self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
            workflow = WorkRequest.objects.latest("created_at")
            with urlconf_scope(workflow.workspace.scope.name):
                self.assertEqual(data, WorkRequestSerializer(workflow).data)
            self.assertEqual(workflow.workspace, workspace)
            self.assertEqual(workflow.task_data["architectures"], architectures)

        with self.assert_model_count_unchanged(WorkRequest):
            response = self.post_workflow(
                {
                    "template_name": "test",
                    "workspace": "common-name",
                    "task_data": task_data,
                },
                scope=scope3.name,
            )
        self.assertResponseProblem(
            response,
            "Workspace not found",
            detail_pattern="Workspace common-name not found in scope scope3",
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_create_workflow_no_default_workspace(self) -> None:
        """POST with no workspace in a scope without a default workspace."""
        scope = self.playground.get_or_create_scope("empty-scope")

        with self.assert_model_count_unchanged(WorkRequest):
            response = self.post_workflow(
                {"template_name": "test"}, scope=scope.name
            )

        self.assertResponseProblem(
            response,
            "Cannot deserialize workflow creation request",
            validation_errors_pattern=(
                r"'workspace': \['This field is required\.'\]"
            ),
        )

    def test_create_workflow_private_workspace_unauthorized(self) -> None:
        """POST to private workspaces 404s to the unauthorized."""
        private_workspace = self.playground.create_workspace(name="Private")
        self.playground.create_workflow_template(
            name="test", task_name="noop", workspace=private_workspace
        )

        with self.assert_model_count_unchanged(WorkRequest):
            response = self.post_workflow(
                {"template_name": "test", "workspace": private_workspace.name},
                scope=private_workspace.scope.name,
            )

        self.assertResponseProblem(
            response,
            "Workspace not found",
            detail_pattern=(
                f"Workspace {private_workspace.name} not found in scope "
                f"{private_workspace.scope.name}"
            ),
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_create_workflow_private_workspace_authorized(self) -> None:
        """POST to private workspaces succeeds for the authorized."""
        private_workspace = self.playground.create_workspace(name="Private")
        self.playground.create_group_role(
            private_workspace, Workspace.Roles.OWNER, users=[self.scenario.user]
        )
        self.playground.create_workflow_template(
            name="test", task_name="noop", workspace=private_workspace
        )

        response = self.post_workflow(
            {"template_name": "test", "workspace": private_workspace.name},
            scope=private_workspace.scope.name,
        )

        data = self.assertAPIResponseOk(response, status.HTTP_201_CREATED)
        workflow = WorkRequest.objects.latest("created_at")
        self.assertEqual(data, WorkRequestSerializer(workflow).data)
        self.assertEqual(workflow.workspace, private_workspace)

    def test_create_workflow_without_template_not_owner(self) -> None:
        """Only workspace owners can create workflows without a template."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            [self.scenario.user],
        )
        artifact = self.playground.create_source_artifact()

        response = self.post_workflow(
            {
                "task_name": "sbuild",
                "task_data": {
                    "input": {"source_artifact": artifact.id},
                    "target_distribution": "debian:bookworm",
                    "architectures": ["amd64", "arm64"],
                },
            }
        )

        self.assertResponseProblem(
            response,
            f"playground cannot configure workspace {self.scenario.workspace}",
            status_code=status.HTTP_403_FORBIDDEN,
        )

    @override_permission(Scope, "can_create_workspace", DenyAll)
    def test_create_child_workspace_without_template_not_scope_owner(
        self,
    ) -> None:
        """Only scope owners may create ``create_child_workspace`` templates."""
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )

        response = self.post_workflow(
            {
                "task_name": "create_child_workspace",
                "task_data": {"suffix": "test"},
            }
        )

        self.assertResponseProblem(
            response,
            f"playground cannot create workspaces in {self.scenario.scope}",
            status_code=status.HTTP_403_FORBIDDEN,
        )
