# 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 work request models."""

import datetime as dt
import io
import logging
import re
from enum import ReprEnum
from functools import partial
from typing import Any, ClassVar, Literal, cast, override
from unittest import mock

import pydantic
import pydantic_core
from asgiref.sync import sync_to_async
from django.conf import settings
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import (
    ImproperlyConfigured,
    PermissionDenied,
    ValidationError,
)
from django.db import IntegrityError, transaction
from django.test import RequestFactory, override_settings
from django.utils import timezone
from django_prometheus.testutils import assert_metric_diff, save_registry

from debusine.artifacts.models import (
    ArtifactCategory,
    BareDataCategory,
    CollectionCategory,
    DebusinePromise,
    RuntimeStatistics,
    TaskTypes,
)
from debusine.client.models import (
    LookupChildType,
    RuntimeParameter,
    RuntimeParameters,
)
from debusine.db.context import ContextConsistencyError, context
from debusine.db.models import (
    Collection,
    CollectionItem,
    Group,
    Scope,
    Token,
    WorkRequest,
    Worker,
    WorkerPoolTaskExecutionStatistics,
    WorkflowTemplate,
    Workspace,
    default_workspace,
)
from debusine.db.models.permissions import Role, Roles
from debusine.db.models.tests.utils import RolesTestCase
from debusine.db.models.work_requests import (
    CannotAbort,
    CannotRetry,
    DeleteUnused,
    InternalTaskError,
    MAX_AUTOMATIC_RETRIES,
    SkipWorkRequest,
    StatusChangeError,
    WorkRequestRetryReason,
    WorkRequestRoleBase,
    WorkRequestRoles,
    compute_workflow_runtime_status,
    start_latency,
)
from debusine.db.permissioncontext import PermissionContext
from debusine.db.playground import scenarios
from debusine.server.collections import CollectionManagerInterface
from debusine.server.management.commands.delete_expired import (
    DeleteExpiredWorkRequests,
    DeleteOperation,
)
from debusine.server.scheduler import update_workflows
from debusine.server.tasks import ServerNoop
from debusine.server.tasks.tests.helpers import SampleBaseWaitTask
from debusine.server.tasks.wait import Delay
from debusine.server.tasks.wait.models import DelayData, ExternalDebsignData
from debusine.server.workflows import NoopWorkflow
from debusine.server.workflows.models import (
    BaseWorkflowData,
    ChildWorkspaceData,
    WorkRequestWorkflowData,
)
from debusine.server.workflows.tests.helpers import SampleWorkflow
from debusine.signing.tasks import SigningNoop
from debusine.tasks import Noop, TaskConfigError
from debusine.tasks.executors import ExecutorImageCategory, ExecutorInterface
from debusine.tasks.models import (
    ActionRecordInTaskHistory,
    ActionRetryWithDelays,
    ActionSendNotification,
    ActionSkipIfLookupResultChanged,
    ActionTypes,
    ActionUpdateCollectionWithArtifacts,
    ActionUpdateCollectionWithData,
    BackendType,
    BaseDynamicTaskData,
    BaseTaskData,
    Confirmation,
    EventReaction,
    EventReactions,
    LookupMultiple,
    NoopData,
    OutputData,
    OutputDataError,
    SbuildData,
    SbuildInput,
)
from debusine.test.django import (
    AllowAll,
    BaseTestWorkerTask,
    ChannelsHelpersMixin,
    DenyAll,
    TestCase,
    TransactionTestCase,
    override_permission,
    preserve_db_task_registry,
)


class WorkRequestRolesTests(RolesTestCase[WorkRequestRoles]):
    """Tests for the WorkRequestRoles class."""

    scenario = scenarios.DefaultContext()
    roles_class = WorkRequestRoles

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

    def test_from_iterable(self) -> None:
        for value, expected in (
            ([], []),
            (["owner"], [WorkRequest.Roles.OWNER]),
            (["creator"], [WorkRequest.Roles.CREATOR]),
            (["contributor"], [WorkRequest.Roles.CONTRIBUTOR]),
            (["viewer"], [WorkRequest.Roles.VIEWER]),
            (
                ["owner", WorkRequest.Roles.CREATOR],
                [WorkRequest.Roles.OWNER],
            ),
            (
                ["contributor", "creator"],
                [WorkRequest.Roles.CONTRIBUTOR, WorkRequest.Roles.CREATOR],
            ),
            (
                ["viewer", "creator"],
                [WorkRequest.Roles.VIEWER, WorkRequest.Roles.CREATOR],
            ),
            (
                ["contributor", "viewer", "creator"],
                [WorkRequest.Roles.CONTRIBUTOR, WorkRequest.Roles.CREATOR],
            ),
            (["viewer", "owner"], [WorkRequest.Roles.OWNER]),
            (["creator", "owner"], [WorkRequest.Roles.OWNER]),
            (
                [
                    "owner",
                    "owner",
                    WorkRequest.Roles.OWNER,
                    WorkRequest.Roles.VIEWER,
                    WorkRequest.Roles.CREATOR,
                ],
                [WorkRequest.Roles.OWNER],
            ),
        ):
            with self.subTest(value=repr(value)):
                self.assertEqual(
                    WorkRequest.Roles.from_iterable(value),
                    frozenset(expected),
                )

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

    def test_init(self) -> None:
        r = WorkRequestRoleBase("test")
        r._setup()
        self.assertEqual(r.implied_by_scope_roles, frozenset())
        self.assertEqual(r.implied_by_workspace_roles, frozenset([]))
        self.assertEqual(r.implied_by_public, False)
        self.assertEqual(r.implied_by_creator, False)

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

    def test_q_non_authenticated(self) -> None:
        """Test generating Q objects with anonymous users."""
        wr = self.playground.create_worker_task()
        for role, expected in (
            (WorkRequestRoles.OWNER, False),
            (WorkRequestRoles.CREATOR, False),
            (WorkRequestRoles.CONTRIBUTOR, False),
            (WorkRequestRoles.VIEWER, True),
        ):
            with self.subTest(role=repr(role)):
                expected_qs = [wr] if expected else []
                self.assertQuerySetEqual(
                    WorkRequest.objects.filter(pk=wr.pk).filter(
                        role.q(PermissionContext.for_ui_anonymous())
                    ),
                    expected_qs,
                )

    def test_values(self) -> None:
        role = WorkRequest.Roles.OWNER
        self.assertEqual(role.label, "Owner")
        self.assertEqual(
            role.implied_by_scope_roles, frozenset([Scope.Roles.OWNER])
        )
        self.assertEqual(
            role.implied_by_workspace_roles, frozenset([Workspace.Roles.OWNER])
        )
        self.assertFalse(role.implied_by_public)
        self.assertFalse(role.implied_by_creator)
        self.assertTrue(role.implies(WorkRequest.Roles.CREATOR))
        self.assertTrue(role.implies(WorkRequest.Roles.VIEWER))

        role = WorkRequest.Roles.CREATOR
        self.assertEqual(role.label, "Creator")
        self.assertEqual(
            role.implied_by_scope_roles, frozenset([Scope.Roles.OWNER])
        )
        self.assertEqual(
            role.implied_by_workspace_roles,
            frozenset([Workspace.Roles.OWNER]),
        )
        self.assertFalse(role.implied_by_public)
        self.assertTrue(role.implied_by_creator)
        self.assertFalse(role.implies(WorkRequest.Roles.VIEWER))

        role = WorkRequest.Roles.CONTRIBUTOR
        self.assertEqual(role.label, "Contributor")
        self.assertEqual(
            role.implied_by_scope_roles, frozenset([Scope.Roles.OWNER])
        )
        self.assertEqual(
            role.implied_by_workspace_roles,
            frozenset([Workspace.Roles.OWNER, Workspace.Roles.CONTRIBUTOR]),
        )
        self.assertFalse(role.implied_by_public)
        self.assertFalse(role.implied_by_creator)
        self.assertTrue(role.implies(WorkRequest.Roles.VIEWER))

        role = WorkRequest.Roles.VIEWER
        self.assertEqual(role.label, "Viewer")
        self.assertEqual(
            Workspace.Roles.VIEWER.implied_by_scope_roles,
            frozenset([Scope.Roles.OWNER, Scope.Roles.OBSERVER]),
        )
        self.assertEqual(
            role.implied_by_workspace_roles,
            frozenset(
                [
                    Workspace.Roles.OWNER,
                    Workspace.Roles.CONTRIBUTOR,
                    Workspace.Roles.VIEWER,
                ]
            ),
        )
        self.assertTrue(role.implied_by_public)
        self.assertFalse(role.implied_by_creator)
        self.assertFalse(role.implies(WorkRequest.Roles.CREATOR))

    def test_values_implication_order(self) -> None:
        class TestRoles(Roles, WorkRequestRoleBase, ReprEnum):
            OWNER = Role("owner", implied_by=[Workspace.Roles.OWNER])
            VIEWER = Role("viewer", implied_by=[Workspace.Roles.VIEWER, OWNER])

        TestRoles.setup()
        role = TestRoles.VIEWER
        self.assertEqual(
            role.implied_by_scope_roles,
            frozenset([Scope.Roles.OWNER, Scope.Roles.OBSERVER]),
        )
        self.assertEqual(
            role.implied_by_workspace_roles,
            frozenset(
                [
                    Workspace.Roles.OWNER,
                    Workspace.Roles.CONTRIBUTOR,
                    Workspace.Roles.VIEWER,
                ]
            ),
        )
        self.assertTrue(role.implied_by_public)
        self.assertFalse(role.implied_by_creator)


class SampleWorkflowData(BaseWorkflowData):
    """Workflow data to test instantiation."""

    ivar: int = 1
    dvar: dict[str, Any] = pydantic.Field(default_factory=dict)
    svar: str | None = None


class WorkRequestManagerTestsTestWorkflow(
    SampleWorkflow[SampleWorkflowData, BaseDynamicTaskData]
):
    """Workflow used to test instantiation of data."""

    # This needs a unique name to avoid conflicting with other tests, because
    # it gets registered as a Workflow subclass

    @override
    def validate_input(self) -> None:
        """Allow testing thorough validation of input data."""
        if "sentinel" in self.data.dvar:
            raise ValueError("Sentinel found")

    @override
    def populate(self) -> None:
        raise NotImplementedError()


class WorkRequestManagerTests(TestCase):
    """Tests for WorkRequestManager."""

    scenario = scenarios.DefaultContext()

    def assertRoleAnnotations(
        self,
        pc: PermissionContext,
        work_requests: dict[str, WorkRequest],
        expected: dict[str, dict[str, bool]],
    ) -> None:
        """
        Ensure the roles computed by the database match the expected set.

        :param expected: a dict in the form ``{pk: {role: bool}}``
        """
        names = {v: k for k, v in work_requests.items()}
        by_wr: dict[str, dict[str, bool]] = {}
        for work_request in WorkRequest.objects.with_role_annotations(pc):
            has_roles: dict[str, bool] = {}
            for role in WorkRequest.Roles:
                has_roles[role] = getattr(work_request, f"is_{role}")
            by_wr[names[work_request]] = has_roles

            roles_for: str = getattr(work_request, "roles_for")
            if not pc.user.is_authenticated:
                self.assertEqual(roles_for, "")
            else:
                self.assertEqual(roles_for, str(pc.user.pk))

        self.assertEqual(by_wr, expected)

    def test_create_workflow(self) -> None:
        user = self.playground.get_default_user()

        workflow_name = "workrequestmanagerteststestworkflow"

        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name=workflow_name,
            static_parameters={},
            priority=10,
        )

        wr = self.playground.create_workflow(template)
        self.assertEqual(wr.workspace, template.workspace)
        self.assertIsNone(wr.parent)
        self.assertEqual(wr.workflow_root, wr)
        self.assertEqual(wr.workflow_template, template)
        self.assertEqual(
            wr.workflow_data_json, {"workflow_template_name": "test"}
        )
        self.assertEqual(wr.created_by, user)
        self.assertEqual(wr.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(wr.task_type, TaskTypes.WORKFLOW)
        self.assertEqual(wr.task_name, workflow_name)
        self.assertEqual(wr.task_data, {})
        self.assertEqual(wr.priority_base, 10)
        self.assertIsNone(wr.output_data)
        self.assertIsNotNone(wr.configured_task_data)
        self.assertQuerySetEqual(wr.dependencies.all(), [])
        assert wr.internal_collection is not None
        self.assertEqual(wr.internal_collection.name, f"workflow-{wr.id}")
        self.assertEqual(
            wr.internal_collection.category,
            CollectionCategory.WORKFLOW_INTERNAL,
        )
        self.assertEqual(wr.internal_collection.workspace, template.workspace)
        self.assertEqual(
            wr.internal_collection.retains_artifacts,
            Collection.RetainsArtifacts.WORKFLOW,
        )
        self.assertEqual(wr.internal_collection.workflow, wr)

    def test_create_workflow_blocked(self) -> None:
        workflow_name = "workrequestmanagerteststestworkflow"

        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name=workflow_name,
            static_parameters={},
            priority=10,
        )

        wr = self.playground.create_workflow(template, mark_pending=False)
        self.assertIsNone(wr.configured_task_data)

    def test_create_workflow_with_parent(self) -> None:
        """A sub-workflow does not have its own internal collection."""
        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name="workrequestmanagerteststestworkflow",
        )
        root = self.playground.create_workflow(template, task_data={})
        sub_template = WorkflowTemplate.objects.create(
            name="subtest", workspace=default_workspace(), task_name="noop"
        )
        sub = self.playground.create_workflow(
            sub_template, task_data={}, parent=root
        )

        self.assertEqual(sub.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(sub.parent, root)
        self.assertEqual(sub.workflow_root, root)
        self.assertEqual(sub.workflow_template, sub_template)
        self.assertEqual(
            sub.workflow_data_json, {"workflow_template_name": "subtest"}
        )
        self.assertQuerySetEqual(sub.dependencies.all(), [])
        self.assertIsNone(sub.internal_collection)

    def test_create_workflow_with_parms(self) -> None:
        """Check that WorkflowTemplate parameters have precedence."""
        workflow_name = "workrequestmanagerteststestworkflow"

        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name=workflow_name,
            static_parameters={"ivar": 2, "dvar": {"a": 1}},
            runtime_parameters=RuntimeParameter.ANY,
        )

        wr = self.playground.create_workflow(template, task_data={})
        self.assertEqual(wr.task_data, {"ivar": 2, "dvar": {"a": 1}})

        wr = self.playground.create_workflow(
            template, task_data={"ivar": 3, "dvar": {"b": 2}}
        )
        self.assertEqual(wr.task_data, {"ivar": 3, "dvar": {"b": 2}})

    def test_create_workflow_without_template(self) -> None:
        user = self.playground.get_default_user()
        workspace = default_workspace()
        workflow_name = "workrequestmanagerteststestworkflow"

        wr = self.playground.create_workflow(
            task_name=workflow_name, workspace=workspace
        )

        self.assertEqual(wr.workspace, workspace)
        self.assertIsNone(wr.parent)
        self.assertEqual(wr.workflow_root, wr)
        self.assertIsNone(wr.workflow_template)
        self.assertEqual(wr.workflow_data_json, {})
        self.assertEqual(wr.created_by, user)
        self.assertEqual(wr.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(wr.task_type, TaskTypes.WORKFLOW)
        self.assertEqual(wr.task_name, workflow_name)
        self.assertEqual(wr.task_data, {})
        self.assertEqual(wr.priority_base, 0)
        self.assertIsNone(wr.output_data)
        self.assertIsNotNone(wr.configured_task_data)
        self.assertQuerySetEqual(wr.dependencies.all(), [])
        assert wr.internal_collection is not None
        self.assertEqual(wr.internal_collection.name, f"workflow-{wr.id}")
        self.assertEqual(
            wr.internal_collection.category,
            CollectionCategory.WORKFLOW_INTERNAL,
        )
        self.assertEqual(wr.internal_collection.workspace, workspace)
        self.assertEqual(
            wr.internal_collection.retains_artifacts,
            Collection.RetainsArtifacts.WORKFLOW,
        )
        self.assertEqual(wr.internal_collection.workflow, wr)

    @override_permission(Scope, "can_create_workspace", DenyAll)
    def test_create_workflow_create_child_workspace_unauthorized(self) -> None:
        workspace = default_workspace()

        with self.assertRaisesRegex(
            PermissionDenied,
            f"{self.scenario.user} cannot create workspaces in"
            f" {workspace.scope}",
        ):
            self.playground.create_workflow(
                task_name="create_child_workspace",
                task_data=ChildWorkspaceData(suffix="test"),
                workspace=workspace,
            )

    def test_create_workflow_create_child_workspace_authorized(self) -> None:
        workspace = default_workspace()

        wr = self.playground.create_workflow(
            task_name="create_child_workspace",
            task_data=ChildWorkspaceData(suffix="test"),
            workspace=workspace,
        )

        self.assertEqual(wr.workspace, workspace)
        self.assertEqual(wr.created_by, self.scenario.user)
        self.assertEqual(wr.task_name, "create_child_workspace")
        self.assertEqual(wr.task_data, {"suffix": "test"})
        self.assertEqual(wr.status, WorkRequest.Statuses.PENDING)

    def test_create_workflow_basic_validation(self) -> None:
        """Workflow parameters are validated against their model."""
        workflow_name = "workrequestmanagerteststestworkflow"

        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name=workflow_name,
            static_parameters={},
            runtime_parameters=RuntimeParameter.ANY,
        )

        with self.assertRaisesRegex(
            ValidationError, "Extra inputs are not permitted"
        ):
            self.playground.create_workflow(template, task_data={"noise": 1})

    def test_create_workflow_thorough_validation(self) -> None:
        """Check validation of parameters on workflow instantiation."""
        workflow_name = "workrequestmanagerteststestworkflow"

        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name=workflow_name,
            static_parameters={},
            runtime_parameters=RuntimeParameter.ANY,
        )

        with self.assertRaisesRegex(ValueError, "Sentinel found"):
            self.playground.create_workflow(
                template, task_data={"dvar": {"sentinel": True}}
            )

    def test_enforce_runtime_parameters(self) -> None:
        task_data = {"foo": "bar"}
        for runtime_parameters in (
            RuntimeParameter.ANY,
            {"foo": RuntimeParameter.ANY},
            {"foo": ["bar", "baz"]},
        ):
            with self.subTest(runtime_parameters=str(runtime_parameters)):
                WorkRequest.objects._enforce_runtime_parameters(
                    task_data, cast(RuntimeParameters, runtime_parameters)
                )

        for runtime_parameters in (
            {"foo": ["quux"]},
            {},
            {"bar": ["quux"]},
        ):
            with (
                self.subTest(runtime_parameters=str(runtime_parameters)),
                self.assertRaisesRegex(
                    ValidationError,
                    r"{'task_data': \[\"runtime_parameters do not permit 'foo'",
                ),
            ):
                WorkRequest.objects._enforce_runtime_parameters(
                    task_data, cast(RuntimeParameters, runtime_parameters)
                )

    def test_create_workflow_permitted_parameter(self) -> None:
        """Check validation of parameters on workflow instantiation."""
        workflow_name = "workrequestmanagerteststestworkflow"

        template = WorkflowTemplate.objects.create(
            name="test",
            workspace=default_workspace(),
            task_name=workflow_name,
            static_parameters={"svar": "foo"},
            runtime_parameters={"svar": ["foo", "bar"]},
        )

        with self.subTest(permitted=True):
            wf = self.playground.create_workflow(
                template, task_data={"svar": "bar"}
            )
            self.assertEqual(wf.task_data["svar"], "bar")

        with (
            self.subTest(permitted=False),
            self.assertRaisesRegex(
                ValidationError,
                (
                    r"{'task_data': \[\"runtime_parameters do not permit "
                    r"'svar': 'baz', only \['foo', 'bar'\]\"\]\}"
                ),
            ),
        ):
            self.playground.create_workflow(template, task_data={"svar": "baz"})

    def test_create_synchronization_point(self) -> None:
        """Synchronization points are created correctly."""
        user = self.playground.get_default_user()

        parent = self.playground.create_workflow("noop")
        parent.priority_base = 5
        parent.priority_adjustment = 1
        parent.save()

        wr = WorkRequest.objects.create_synchronization_point(
            parent=parent, step="test"
        )
        self.assertEqual(wr.workspace, parent.workspace)
        self.assertEqual(wr.parent, parent)
        self.assertEqual(wr.workflow_root, parent)
        self.assertIsNone(wr.workflow_template)
        self.assertIsNotNone(wr.created_at)
        self.assertIsNone(wr.pending_at)
        self.assertIsNone(wr.started_at)
        self.assertIsNone(wr.completed_at)
        self.assertEqual(wr.created_by, user)
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr.result, WorkRequest.Results.NONE)
        self.assertIsNone(wr.worker)
        self.assertEqual(wr.task_type, TaskTypes.INTERNAL)
        self.assertEqual(wr.task_name, "synchronization_point")
        self.assertEqual(wr.task_data, {})
        self.assertEqual(wr.priority_base, 6)
        self.assertIsNone(wr.output_data)
        self.assertQuerySetEqual(wr.dependencies.all(), [])
        self.assertEqual(
            wr.workflow_data,
            WorkRequestWorkflowData(step="test", display_name="test"),
        )
        wr.full_clean()

        workspace = self.playground.create_workspace(name="test", public=True)

        parent = self.playground.create_workflow("noop", workspace=workspace)
        wr = WorkRequest.objects.create_synchronization_point(
            parent=parent, step="test", display_name="Test"
        )
        self.assertEqual(wr.workspace, workspace)
        self.assertEqual(wr.parent, parent)
        self.assertEqual(wr.workflow_root, parent)
        self.assertIsNotNone(wr.created_at)
        self.assertIsNone(wr.pending_at)
        self.assertIsNone(wr.started_at)
        self.assertIsNone(wr.completed_at)
        self.assertEqual(wr.created_by, user)
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr.result, WorkRequest.Results.NONE)
        self.assertIsNone(wr.worker)
        self.assertEqual(wr.task_type, TaskTypes.INTERNAL)
        self.assertEqual(wr.task_name, "synchronization_point")
        self.assertEqual(wr.task_data, {})
        self.assertIsNone(wr.output_data)
        self.assertEqual(
            wr.workflow_data,
            WorkRequestWorkflowData(step="test", display_name="Test"),
        )
        wr.full_clean()

    def test_create_workflow_callback(self) -> None:
        """Workflow callbacks are created correctly."""
        user = self.playground.get_default_user()

        parent = self.playground.create_workflow("noop")

        wr = WorkRequest.objects.create_workflow_callback(
            parent=parent, step="test"
        )
        self.assertEqual(wr.workspace, parent.workspace)
        self.assertEqual(wr.parent, parent)
        self.assertEqual(wr.workflow_root, parent)
        self.assertIsNone(wr.workflow_template)
        self.assertIsNotNone(wr.created_at)
        self.assertIsNone(wr.pending_at)
        self.assertIsNone(wr.started_at)
        self.assertIsNone(wr.completed_at)
        self.assertEqual(wr.created_by, user)
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr.result, WorkRequest.Results.NONE)
        self.assertIsNone(wr.worker)
        self.assertEqual(wr.task_type, TaskTypes.INTERNAL)
        self.assertEqual(wr.task_name, "workflow")
        self.assertEqual(wr.task_data, {})
        self.assertEqual(wr.unblock_strategy, WorkRequest.UnblockStrategy.DEPS)
        self.assertEqual(wr.priority_base, 0)
        self.assertIsNone(wr.output_data)
        self.assertQuerySetEqual(wr.dependencies.all(), [])
        self.assertEqual(
            wr.workflow_data,
            WorkRequestWorkflowData(step="test", display_name="test"),
        )
        wr.full_clean()

        workspace = self.playground.create_workspace(name="test", public=True)

        parent = self.playground.create_workflow("noop", workspace=workspace)
        parent.priority_base = 20
        parent.priority_adjustment = 10
        parent.save()
        wr = WorkRequest.objects.create_workflow_callback(
            parent=parent,
            unblock_strategy=WorkRequest.UnblockStrategy.INCREMENTAL,
            step="test1",
            display_name="Test One",
        )
        self.assertEqual(wr.workspace, workspace)
        self.assertEqual(wr.parent, parent)
        self.assertEqual(wr.workflow_root, parent)
        self.assertIsNotNone(wr.created_at)
        self.assertIsNone(wr.pending_at)
        self.assertIsNone(wr.started_at)
        self.assertIsNone(wr.completed_at)
        self.assertEqual(wr.created_by, user)
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr.result, WorkRequest.Results.NONE)
        self.assertIsNone(wr.worker)
        self.assertEqual(wr.task_type, TaskTypes.INTERNAL)
        self.assertEqual(wr.task_name, "workflow")
        self.assertEqual(wr.task_data, {})
        self.assertEqual(
            wr.unblock_strategy, WorkRequest.UnblockStrategy.INCREMENTAL
        )
        self.assertEqual(wr.priority_base, 30)
        self.assertIsNone(wr.output_data)
        self.assertEqual(
            wr.workflow_data,
            WorkRequestWorkflowData(step="test1", display_name="Test One"),
        )
        wr.full_clean()

    def test_pending(self) -> None:
        """WorkRequestManager.pending() returns pending WorkRequests."""
        work_request_1 = self.playground.create_worker_task(mark_pending=True)

        work_request_2 = self.playground.create_worker_task(mark_pending=True)

        self.playground.create_worker_task(mark_aborted=True)

        self.assertQuerySetEqual(
            WorkRequest.objects.pending(), [work_request_1, work_request_2]
        )

        work_request_1.created_at = timezone.now()
        work_request_1.save()

        self.assertQuerySetEqual(
            WorkRequest.objects.pending(), [work_request_2, work_request_1]
        )

        work_request_1.priority_adjustment = 1
        work_request_1.save()

        self.assertQuerySetEqual(
            WorkRequest.objects.pending(), [work_request_1, work_request_2]
        )

        work_request_2.priority_base = 2
        work_request_2.save()

        self.assertQuerySetEqual(
            WorkRequest.objects.pending(), [work_request_2, work_request_1]
        )

    def test_pending_filter_by_worker(self) -> None:
        """WorkRequestManager.pending() returns WorkRequest for the worker."""
        self.playground.create_worker_task(mark_pending=True, task_name="noop")
        worker = self.playground.create_worker()
        work_request_2 = self.playground.create_worker_task(
            mark_pending=True, worker=worker
        )

        self.assertQuerySetEqual(
            WorkRequest.objects.pending(worker=worker), [work_request_2]
        )

    def test_raise_value_error_exclude_assigned_and_worker(self) -> None:
        """WorkRequestManager.pending() raises ValueError."""
        worker = self.playground.create_worker()

        with self.assertRaisesRegex(
            ValueError, "Cannot exclude_assigned and filter by worker"
        ):
            WorkRequest.objects.pending(exclude_assigned=True, worker=worker)

    def test_pending_exclude_assigned(self) -> None:
        """
        Test WorkRequestManager.pending(exclude_assigned=True).

        It excludes work requests that are assigned to a worker.
        """
        # Pending, not assigned to a worker WorkRequest
        work_request = self.playground.create_worker_task(
            mark_pending=True, task_name="noop"
        )

        # Is returned as expected
        self.assertQuerySetEqual(
            WorkRequest.objects.pending(exclude_assigned=True), [work_request]
        )

        # Creates a worker
        worker = self.playground.create_worker()

        # Assigns the worker to the work_request
        work_request.worker = worker
        work_request.save()

        # pending(exclude_assigned=True) doesn't return it anymore
        self.assertQuerySetEqual(
            WorkRequest.objects.pending(exclude_assigned=True), []
        )

        # without the exclude_assigned it returns it
        self.assertQuerySetEqual(
            WorkRequest.objects.pending(exclude_assigned=False), [work_request]
        )

    def test_running(self) -> None:
        """WorkRequestManager.running() returns running WorkRequests."""
        work_request = self.playground.create_worker_task(mark_running=True)
        self.playground.create_worker_task(mark_aborted=True)

        self.assertQuerySetEqual(WorkRequest.objects.running(), [work_request])

    def test_completed(self) -> None:
        """WorkRequestManager.completed() returns completed WorkRequests."""
        work_request = self.playground.create_worker_task(
            result=WorkRequest.Results.SUCCESS
        )
        self.playground.create_worker_task(mark_running=True)

        self.assertQuerySetEqual(
            WorkRequest.objects.completed(), [work_request]
        )

    def test_reassignable(self) -> None:
        """WorkRequestManager.reassignable()"""
        worker = self.playground.create_worker()
        self.playground.create_worker_task(
            result=WorkRequest.Results.SUCCESS, worker=worker
        )
        work_request_running = self.playground.create_worker_task(
            mark_running=True, worker=worker
        )
        work_request_pending = self.playground.create_worker_task(
            mark_pending=True
        )
        work_request_pending.worker = worker
        work_request_pending.save()

        with self.subTest(with_output=False):
            self.assertQuerySetEqual(
                WorkRequest.objects.reassignable(worker=worker),
                [work_request_running, work_request_pending],
            )

        with self.subTest(with_output=True):
            self.playground.create_build_log_artifact(
                work_request=work_request_running
            )
            self.assertQuerySetEqual(
                WorkRequest.objects.reassignable(worker=worker),
                [work_request_pending],
            )

    def test_aborted(self) -> None:
        """WorkRequestManager.aborted() returns aborted WorkRequests."""
        work_request = self.playground.create_worker_task(mark_aborted=True)
        self.playground.create_worker_task(mark_running=True)

        self.assertQuerySetEqual(WorkRequest.objects.aborted(), [work_request])

    def test_aborted_or_failed(self) -> None:
        """WorkRequestManager.aborted() returns aborted WorkRequests."""
        wr_aborted = self.playground.create_worker_task(mark_aborted=True)
        wr_failed = self.playground.create_worker_task(
            result=WorkRequest.Results.FAILURE,
        )
        self.playground.create_worker_task(mark_running=True)

        self.assertQuerySetEqual(
            WorkRequest.objects.aborted_or_failed(), [wr_aborted, wr_failed]
        )

    def test_expired(self) -> None:
        """Test expired() method."""
        wr = self.playground.create_worker_task(task_name="noop")

        wr.created_at = timezone.now() - dt.timedelta(days=2)
        wr.expiration_delay = dt.timedelta(days=1)
        wr.save()
        self.assertQuerySetEqual(
            WorkRequest.objects.expired(timezone.now()), [wr]
        )

        wr.expiration_delay = dt.timedelta(0)
        wr.save()
        self.assertQuerySetEqual(
            WorkRequest.objects.expired(timezone.now()), []
        )

        wr.created_at = timezone.now()
        wr.expiration_delay = dt.timedelta(days=1)
        wr.save()
        self.assertQuerySetEqual(
            WorkRequest.objects.expired(timezone.now()), []
        )

        workspace = self.playground.create_workspace(
            name="test", default_expiration_delay=dt.timedelta(days=2)
        )
        wr.workspace = workspace
        wr.created_at = timezone.now() - dt.timedelta(days=3)
        wr.expiration_delay = None
        wr.save()
        self.assertQuerySetEqual(
            WorkRequest.objects.expired(timezone.now()), [wr]
        )

        workspace.default_expiration_delay = dt.timedelta(days=5)
        workspace.save()
        self.assertQuerySetEqual(
            WorkRequest.objects.expired(timezone.now()), []
        )

        workspace.default_expiration_delay = dt.timedelta(days=0)
        workspace.save()
        self.assertQuerySetEqual(
            WorkRequest.objects.expired(timezone.now()), []
        )

    def test_in_current_scope(self) -> None:
        """Test the in_current_scope() QuerySet filter."""
        work_request = self.playground.create_worker_task()
        with context.local():
            self.scenario.set_current()
            self.assertQuerySetEqual(
                WorkRequest.objects.in_current_scope(),
                [work_request],
            )

        scope1 = Scope.objects.create(name="Scope1")
        with context.local():
            context.set_permission_context_scoped(
                scope1, PermissionContext.for_ui_anonymous()
            )
            self.assertQuerySetEqual(
                WorkRequest.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"
        ):
            WorkRequest.objects.in_current_scope()

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

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

        workspace1 = self.playground.create_workspace(name="other", public=True)
        with context.local():
            workspace1.set_current()
            self.assertQuerySetEqual(
                WorkRequest.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"
        ):
            WorkRequest.objects.in_current_workspace()

    def test_with_role_annotations_anonymous(self) -> None:
        ws_public = self.scenario.workspace
        ws_private = self.playground.create_workspace(name="private")
        work_requests: dict[str, WorkRequest] = {
            "public": self.playground.create_worker_task(workspace=ws_public),
            "private": self.playground.create_worker_task(workspace=ws_private),
        }

        self.assertRoleAnnotations(
            PermissionContext.for_ui_anonymous(),
            work_requests,
            {
                "public": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: False,
                    WorkRequest.Roles.CONTRIBUTOR: False,
                    WorkRequest.Roles.VIEWER: True,
                },
                "private": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: False,
                    WorkRequest.Roles.CONTRIBUTOR: False,
                    WorkRequest.Roles.VIEWER: False,
                },
            },
        )

    def test_with_role_annotations(self) -> None:
        other = self.playground.create_user("other")

        ws_public = self.scenario.workspace
        ws_private = self.playground.create_workspace(name="private")
        ws_contributed = self.playground.create_workspace(name="contributed")
        self.playground.create_group_role(
            ws_contributed,
            Workspace.Roles.CONTRIBUTOR,
            users=[self.scenario.user],
        )
        ws_owned = self.playground.create_workspace(name="owned")
        self.playground.create_group_role(
            ws_owned, Workspace.Roles.OWNER, users=[self.scenario.user]
        )

        work_requests: dict[str, WorkRequest] = {
            "created_public": self.playground.create_worker_task(
                workspace=ws_public
            ),
            "other_public": self.playground.create_worker_task(
                workspace=ws_public, created_by=other
            ),
            "created_private": self.playground.create_worker_task(
                workspace=ws_private
            ),
            "other_private": self.playground.create_worker_task(
                workspace=ws_private, created_by=other
            ),
            "created_contributed": self.playground.create_worker_task(
                workspace=ws_contributed
            ),
            "other_contributed": self.playground.create_worker_task(
                workspace=ws_contributed, created_by=other
            ),
            "created_owned": self.playground.create_worker_task(
                workspace=ws_owned
            ),
            "other_owned": self.playground.create_worker_task(
                workspace=ws_owned, created_by=other
            ),
        }

        self.assertRoleAnnotations(
            self.scenario.context_ui(),
            work_requests,
            {
                "created_public": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: True,
                    WorkRequest.Roles.CONTRIBUTOR: False,
                    WorkRequest.Roles.VIEWER: True,
                },
                "other_public": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: False,
                    WorkRequest.Roles.CONTRIBUTOR: False,
                    WorkRequest.Roles.VIEWER: True,
                },
                "created_private": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: True,
                    WorkRequest.Roles.CONTRIBUTOR: False,
                    WorkRequest.Roles.VIEWER: False,
                },
                "other_private": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: False,
                    WorkRequest.Roles.CONTRIBUTOR: False,
                    WorkRequest.Roles.VIEWER: False,
                },
                "created_contributed": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: True,
                    WorkRequest.Roles.CONTRIBUTOR: True,
                    WorkRequest.Roles.VIEWER: True,
                },
                "other_contributed": {
                    WorkRequest.Roles.OWNER: False,
                    WorkRequest.Roles.CREATOR: False,
                    WorkRequest.Roles.CONTRIBUTOR: True,
                    WorkRequest.Roles.VIEWER: True,
                },
                "created_owned": {
                    WorkRequest.Roles.OWNER: True,
                    WorkRequest.Roles.CREATOR: True,
                    WorkRequest.Roles.CONTRIBUTOR: True,
                    WorkRequest.Roles.VIEWER: True,
                },
                "other_owned": {
                    WorkRequest.Roles.OWNER: True,
                    WorkRequest.Roles.CREATOR: True,
                    WorkRequest.Roles.CONTRIBUTOR: True,
                    WorkRequest.Roles.VIEWER: True,
                },
            },
        )

    def test_with_role_annotations_extra_groups(self) -> None:
        # extra groups are now tied to work request permissions, and this test
        # as is does not make any sense. The same testing intention needs to be
        # reimplemented based on work request permissions
        workspaces = {
            "public": self.scenario.workspace,
            "private": self.playground.create_workspace(name="private"),
            "contributed": self.playground.create_workspace(name="contributed"),
            "owned": self.playground.create_workspace(name="owned"),
        }

        contributed_contributors = self.playground.create_group_role(
            workspaces["contributed"], Workspace.Roles.CONTRIBUTOR, users=[]
        )
        owned_owners = self.playground.create_group_role(
            workspaces["owned"], Workspace.Roles.OWNER, users=[]
        )

        pcs: dict[str, PermissionContext] = {}
        for name, ws in workspaces.items():
            if not (orig_public := ws.public):
                ws.public = True
                ws.save()
            pcs[name] = self.playground.create_pc_with_extra_groups(
                contributed_contributors, owned_owners, workspace=ws
            )
            ws.public = orig_public
            ws.save()

        expected = {
            "public": {
                WorkRequest.Roles.OWNER: False,
                WorkRequest.Roles.CREATOR: True,
                WorkRequest.Roles.CONTRIBUTOR: False,
                WorkRequest.Roles.VIEWER: True,
            },
            "private": {
                WorkRequest.Roles.OWNER: False,
                WorkRequest.Roles.CREATOR: True,
                WorkRequest.Roles.CONTRIBUTOR: False,
                WorkRequest.Roles.VIEWER: False,
            },
            "contributed": {
                WorkRequest.Roles.OWNER: False,
                WorkRequest.Roles.CREATOR: True,
                WorkRequest.Roles.CONTRIBUTOR: True,
                WorkRequest.Roles.VIEWER: True,
            },
            "owned": {
                WorkRequest.Roles.OWNER: True,
                WorkRequest.Roles.CREATOR: True,
                WorkRequest.Roles.CONTRIBUTOR: True,
                WorkRequest.Roles.VIEWER: True,
            },
        }

        for name, pc in pcs.items():
            assert pc.work_request
            with self.subTest(name=name):
                qs = WorkRequest.objects.with_role_annotations(pc)
                work_request = qs.get(pk=pc.work_request.pk)

                has_roles: dict[str, bool] = {}
                for role in WorkRequest.Roles:
                    has_roles[role] = getattr(work_request, f"is_{role}")

                self.assertEqual(has_roles, expected[name])

    def test_with_role(self) -> None:
        user_contrib = self.playground.create_user("contrib")
        user_viewer = self.playground.create_user("viewer")
        other = self.playground.create_user("other")
        anon = PermissionContext.for_ui_anonymous()
        ws_contributors = self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            users=[user_contrib],
        )
        ws_viewers = self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.VIEWER,
            users=[user_viewer],
        )
        pc_creator = self.scenario.context_ui()
        pc_contrib = PermissionContext.for_ui_user(user_contrib)
        pc_viewer = PermissionContext.for_ui_user(user_viewer)
        pc_other = PermissionContext.for_ui_user(other)
        pc_other_extra_scope_owners = (
            self.playground.create_pc_with_extra_groups(
                self.scenario.scope_owners, user=other
            )
        )
        pc_other_extra_workspace_owners = (
            self.playground.create_pc_with_extra_groups(
                self.scenario.workspace_owners, user=other
            )
        )
        pc_other_extra_ws_contributors = (
            self.playground.create_pc_with_extra_groups(
                ws_contributors, user=other
            )
        )
        pc_other_extra_ws_viewers = self.playground.create_pc_with_extra_groups(
            ws_viewers, user=other
        )
        pc_other_extra_ws_viewers_scope_owners = (
            self.playground.create_pc_with_extra_groups(
                ws_viewers, self.scenario.scope_owners, user=other
            )
        )
        OWNER = WorkRequest.Roles.OWNER
        CREATOR = WorkRequest.Roles.CREATOR
        CONTRIBUTOR = WorkRequest.Roles.CONTRIBUTOR
        VIEWER = WorkRequest.Roles.VIEWER
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        wr = self.playground.create_worker_task()
        for pc, role, expected in (
            (anon, OWNER, False),
            (anon, CREATOR, False),
            (anon, CONTRIBUTOR, False),
            (anon, VIEWER, False),
            (pc_creator, OWNER, False),
            (pc_creator, CREATOR, True),
            (pc_creator, CONTRIBUTOR, False),
            (pc_creator, VIEWER, False),
            (pc_contrib, OWNER, False),
            (pc_contrib, CREATOR, False),
            (pc_contrib, CONTRIBUTOR, True),
            (pc_contrib, VIEWER, True),
            (pc_viewer, OWNER, False),
            (pc_viewer, CREATOR, False),
            (pc_viewer, CONTRIBUTOR, False),
            (pc_viewer, VIEWER, True),
            (pc_other, OWNER, False),
            (pc_other, CREATOR, False),
            (pc_other, CONTRIBUTOR, False),
            (pc_other, VIEWER, False),
            (pc_other_extra_scope_owners, OWNER, True),
            (pc_other_extra_workspace_owners, OWNER, True),
            (pc_other_extra_ws_contributors, CONTRIBUTOR, True),
            (pc_other_extra_ws_viewers, VIEWER, True),
            (pc_other_extra_ws_viewers_scope_owners, OWNER, True),
            (pc_other_extra_ws_viewers_scope_owners, CONTRIBUTOR, True),
        ):
            with self.subTest(pc=str(pc), role=repr(role)):
                results = list(WorkRequest.objects.with_role(pc, role))
                if expected:
                    self.assertIn(wr, results)
                else:
                    self.assertNotIn(wr, results)

    def test_workflows(self) -> None:
        template = self.playground.create_workflow_template(
            name="test", task_name="noop"
        )
        workflow = self.playground.create_workflow(template, task_data={})
        self.playground.create_worker_task()
        self.assertQuerySetEqual(WorkRequest.objects.workflows(), [workflow])

    def test_needs_input(self) -> None:
        template = self.playground.create_workflow_template(
            name="test", task_name="noop"
        )
        workflow1 = self.playground.create_workflow(template, task_data={})
        workflow1.workflow_runtime_status = (
            WorkRequest.RuntimeStatuses.NEEDS_INPUT
        )
        workflow1.save()
        self.playground.create_workflow(template, task_data={})
        self.assertQuerySetEqual(WorkRequest.objects.needs_input(), [workflow1])

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_unsuperseded(self) -> None:
        self.scenario.set_current()
        wr_old, wr_untouched = (
            self.playground.create_worker_task() for _ in range(2)
        )
        wr_old.mark_aborted()
        wr_retried = wr_old.retry()

        self.assertQuerySetEqual(
            WorkRequest.objects.unsuperseded(),
            [wr_untouched, wr_retried],
            ordered=False,
        )

    def test_terminated(self) -> None:
        wr_completed = self.playground.create_worker_task(
            result=WorkRequest.Results.SUCCESS
        )
        wr_aborted = self.playground.create_worker_task(mark_aborted=True)
        self.playground.create_worker_task(mark_pending=True)
        self.playground.create_worker_task(mark_running=True)
        self.playground.create_worker_task()

        self.assertQuerySetEqual(
            WorkRequest.objects.terminated(),
            [wr_completed, wr_aborted],
            ordered=False,
        )

    def test_unterminated(self) -> None:
        self.playground.create_worker_task(result=WorkRequest.Results.SUCCESS)
        self.playground.create_worker_task(mark_aborted=True)
        wr_pending = self.playground.create_worker_task(mark_pending=True)
        wr_running = self.playground.create_worker_task(mark_running=True)
        wr_blocked = self.playground.create_worker_task()

        self.assertQuerySetEqual(
            WorkRequest.objects.unterminated(),
            [wr_pending, wr_running, wr_blocked],
            ordered=False,
        )

    def test_flattened(self) -> None:
        workflows = [self.playground.create_workflow() for _ in range(2)]
        child_1 = workflows[0].create_child_worker(task_name="noop")
        child_2 = workflows[0].create_child_workflow(task_name="noop")
        child_3 = WorkRequest.objects.create_workflow_callback(
            parent=workflows[0], step="invisible", visible=False
        )
        child_4 = child_2.create_child_worker(task_name="noop")
        child_5 = workflows[1].create_child_worker(task_name="noop")

        self.assertCountEqual(
            [
                (wr.id, wr.ancestor_id)
                for wr in WorkRequest.objects.filter(
                    id=workflows[0].id
                ).workflow_flattened()
            ],
            [(child_1.id, workflows[0].id), (child_4.id, workflows[0].id)],
        )
        self.assertCountEqual(
            [
                (wr.id, wr.ancestor_id)
                for wr in WorkRequest.objects.filter(
                    id=workflows[0].id
                ).workflow_flattened(include_workflows=True)
            ],
            [
                (workflows[0].id, workflows[0].id),
                (child_1.id, workflows[0].id),
                (child_2.id, workflows[0].id),
                (child_4.id, workflows[0].id),
            ],
        )
        self.assertCountEqual(
            [
                (wr.id, wr.ancestor_id)
                for wr in WorkRequest.objects.filter(
                    id__in={workflow.id for workflow in workflows}
                ).workflow_flattened(include_workflows=True)
            ],
            [
                (workflows[0].id, workflows[0].id),
                (child_1.id, workflows[0].id),
                (child_2.id, workflows[0].id),
                (child_4.id, workflows[0].id),
                (workflows[1].id, workflows[1].id),
                (child_5.id, workflows[1].id),
            ],
        )
        self.assertCountEqual(
            [
                (wr.id, wr.ancestor_id)
                for wr in WorkRequest.objects.filter(
                    id=workflows[0].id
                ).workflow_flattened(include_invisible=True)
            ],
            [
                (child_1.id, workflows[0].id),
                (child_3.id, workflows[0].id),
                (child_4.id, workflows[0].id),
            ],
        )
        self.assertCountEqual(
            [
                (wr.id, wr.ancestor_id)
                for wr in WorkRequest.objects.filter(
                    id=child_1.id
                ).workflow_flattened()
            ],
            [(child_1.id, child_1.id)],
        )
        self.assertCountEqual(
            [
                (wr.id, wr.ancestor_id)
                for wr in WorkRequest.objects.filter(
                    id__in={child_2.id, child_4.id}
                ).workflow_flattened(include_workflows=True)
            ],
            [
                (child_2.id, child_2.id),
                (child_4.id, child_2.id),
                (child_4.id, child_4.id),
            ],
        )


class WorkRequestTests(TestCase):
    """Tests for the WorkRequest class."""

    scenario = scenarios.DefaultContext()
    template: ClassVar[WorkflowTemplate]
    workflow: ClassVar[WorkRequest]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.template = cls.playground.create_workflow_template(
            name="test",
            task_name="noop",
        )
        cls.workflow = cls.playground.create_workflow(
            cls.template, task_data={}
        )

    def assertHasRole(
        self,
        role: WorkRequest.Roles,
        work_request: WorkRequest,
        *,
        pc: PermissionContext | None = None,
        hits_database: bool = False,
    ) -> None:
        """Ensure the user has the given role on the work request."""
        pc = pc or self.scenario.context_ui()
        expected_queries = 1 if hits_database else 0
        with self.assertNumQueries(expected_queries):
            self.assertTrue(work_request.has_role(pc, role))

    def assertHasRoles(
        self,
        roles: list[WorkRequest.Roles],
        work_request: WorkRequest,
        *,
        pc: PermissionContext | None = None,
    ) -> None:
        """Ensure the user has exactly the given roles on the work request."""
        pc = pc or self.scenario.context_ui()
        found = [
            role
            for role in WorkRequest.Roles
            if work_request.has_role(pc, role)
        ]
        self.assertEqual(
            ", ".join(str(r) for r in found), ", ".join(str(r) for r in roles)
        )

    def assert_data_must_be_dictionary(
        self, wr: WorkRequest, messages: list[str] | None = None
    ) -> None:
        """Check that a WorkRequest's task_data is a dictionary."""
        if messages is None:
            messages = ["task data must be a dictionary"]
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {"task_data": messages},
        )

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

        wr = self.playground.create_worker_task()
        request = RequestFactory().get("/")
        ui = wr.ui(request)
        self.assertIsInstance(ui, WorkRequestUI)

    def test_get_absolute_url(self) -> None:
        wr = self.playground.create_worker_task()
        self.assertEqual(
            wr.get_absolute_url(),
            f"/{wr.workspace.scope.name}/{wr.workspace.name}"
            f"/work-request/{wr.pk}/",
        )

    def test_get_absolute_url_different_scope(self) -> None:
        """``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)
        wr = self.playground.create_worker_task(workspace=other_workspace)
        self.assertEqual(
            wr.get_absolute_url(),
            f"/{other_scope.name}/{other_workspace.name}/work-request/{wr.pk}/",
        )

    def test_get_absolute_url_retry(self) -> None:
        wr = self.playground.create_worker_task()
        self.assertEqual(
            wr.get_absolute_url_retry(),
            f"/{wr.workspace.scope.name}/{wr.workspace.name}"
            f"/work-request/{wr.pk}/retry/",
        )

    def test_get_absolute_url_retry_different_scope(self) -> None:
        """``get_absolute_url_retry`` works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        wr = self.playground.create_worker_task(workspace=other_workspace)
        self.assertEqual(
            wr.get_absolute_url_retry(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/work-request/{wr.pk}/retry/",
        )

    def test_get_absolute_url_abort(self) -> None:
        wr = self.playground.create_worker_task()
        self.assertEqual(
            wr.get_absolute_url_abort(),
            f"/{wr.workspace.scope.name}/{wr.workspace.name}"
            f"/work-request/{wr.pk}/abort/",
        )

    def test_get_absolute_url_abort_different_scope(self) -> None:
        """``get_absolute_url_abort`` works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        wr = self.playground.create_worker_task(workspace=other_workspace)
        self.assertEqual(
            wr.get_absolute_url_abort(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/work-request/{wr.pk}/abort/",
        )

    def test_get_absolute_url_unblock(self) -> None:
        wr = self.playground.create_worker_task()
        self.assertEqual(
            wr.get_absolute_url_unblock(),
            f"/{wr.workspace.scope.name}/{wr.workspace.name}"
            f"/work-request/{wr.pk}/unblock/",
        )

    def test_get_absolute_url_unblock_different_scope(self) -> None:
        """``get_absolute_url_unblock`` works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        wr = self.playground.create_worker_task(workspace=other_workspace)
        self.assertEqual(
            wr.get_absolute_url_unblock(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/work-request/{wr.pk}/unblock/",
        )

    def test_get_absolute_url_confirm(self) -> None:
        wr = self.playground.create_wait_task(
            task_name="confirm",
            workflow_data=WorkRequestWorkflowData(needs_input=True),
        )
        self.assertEqual(
            wr.get_absolute_url_confirm(),
            f"/{wr.workspace.scope.name}/{wr.workspace.name}"
            f"/work-request/{wr.pk}/confirm/",
        )

    def test_get_absolute_url_confirm_different_scope(self) -> None:
        """``get_absolute_url_confirm`` works in another scope."""
        other_scope = self.playground.get_or_create_scope(name="other")
        other_workspace = self.playground.create_workspace(scope=other_scope)
        wr = self.playground.create_wait_task(
            task_name="confirm",
            workflow_data=WorkRequestWorkflowData(needs_input=True),
            workspace=other_workspace,
        )
        self.assertEqual(
            wr.get_absolute_url_confirm(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/work-request/{wr.pk}/confirm/",
        )

    def test_clean_task_data_must_be_dict(self) -> None:
        """Ensure that task_data is a dict."""
        wr = WorkRequest.objects.create_worker(
            workspace=default_workspace(),
            created_by=self.scenario.user,
            task_name="noop",
            task_data={},
        )
        wr.full_clean()

        wr.task_data = None
        self.assert_data_must_be_dictionary(wr)

        wr.task_data = ""
        self.assert_data_must_be_dictionary(wr)

        wr.task_data = 3
        self.assert_data_must_be_dictionary(wr)

        wr.task_data = []
        self.assert_data_must_be_dictionary(wr)

        wr.task_data = object()
        self.assert_data_must_be_dictionary(
            wr,
            messages=[
                'Value must be valid JSON.',
                'task data must be a dictionary',
            ],
        )

    def test_clean_task_name_must_be_valid_worker(self) -> None:
        """Check validation of worker task names."""
        wr = WorkRequest.objects.create_worker(
            workspace=default_workspace(),
            created_by=self.scenario.user,
            task_name="noop",
            task_data={},
        )
        wr.full_clean()

        wr.task_name = "does-not-exist"
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()

        self.assertEqual(
            raised.exception.message_dict,
            {"task_name": ['does-not-exist: invalid Worker task name']},
        )

        wr.task_name = "does_not_exist"
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()

        self.assertEqual(
            raised.exception.message_dict,
            {
                "task_name": [
                    "does_not_exist: invalid Worker task name",
                ]
            },
        )

    def test_clean_task_name_must_be_valid_server(self) -> None:
        """Check validation of server task names."""
        wr = self.playground.create_server_task(task_name="servernoop")
        wr.full_clean()

        wr.task_name = "does-not-exist"
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()

        self.assertEqual(
            raised.exception.message_dict,
            {"task_name": ['does-not-exist: invalid Server task name']},
        )

    def test_clean_task_name_must_be_valid_internal(self) -> None:
        """Check validation of internal task names."""
        wr = self.playground.create_internal_task(
            task_name="synchronization_point"
        )
        wr.full_clean()

        wr.task_name = "workflow"
        wr.full_clean()

        wr.task_name = "does_not_exist"
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {
                "task_name": [
                    "does_not_exist: invalid task name for internal task"
                ]
            },
        )

    def test_clean_task_name_must_be_valid_workflow(self) -> None:
        """Check validation of workflow task names."""
        wr = WorkRequest.objects.create(
            workspace=default_workspace(),
            created_by=self.scenario.user,
            task_type=TaskTypes.WORKFLOW,
            task_name="servernoop",
            task_data={},
        )
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {"task_name": ["servernoop: invalid workflow name"]},
        )

        wr.task_name = "noop"
        wr.full_clean()

    def test_clean_task_name_must_be_valid_signing(self) -> None:
        """Check validation of signing task names."""
        wr = self.playground.create_signing_task(
            task_name="generatekey",
            task_data={"purpose": "uefi", "description": "A UEFI key"},
        )
        wr.full_clean()

        wr.task_name = "does-not-exist"
        with self.assertRaises(ValidationError) as raised:
            wr.full_clean()

        self.assertEqual(
            raised.exception.message_dict,
            {"task_name": ['does-not-exist: invalid Signing task name']},
        )

    def test_clean_invalid_task_type(self) -> None:
        """An invalid work request task type is rejected."""
        wr = WorkRequest.objects.create(
            workspace=default_workspace(),
            created_by=self.scenario.user,
            task_type="invalid",
            task_name="noop",
            task_data={},
        )
        with self.assertRaises(NotImplementedError):
            wr.full_clean()

    def test_clean_data_task_data_raise_validation_error(self) -> None:
        """Check validation of task data."""

        def make_task_data() -> dict[str, Any]:
            return {
                "build_components": ["any", "all"],
                "build_architecture": "amd64",
                "environment": (
                    "debian/match:codename=bookworm:architecture=amd64"
                ),
                "input": {
                    "source_artifact": 5,
                },
            }

        wr = WorkRequest.objects.create_worker(
            created_by=self.scenario.user,
            workspace=default_workspace(),
            task_name="sbuild",
            task_data=make_task_data(),
        )
        wr.full_clean()

        wr.task_data = make_task_data()
        wr.task_data["invalid-key"] = None
        with self.assertRaisesRegex(
            ValidationError, "Extra inputs are not permitted"
        ):
            wr.full_clean()

        wr.task_data = make_task_data()
        del wr.task_data["environment"]
        with self.assertRaisesRegex(
            ValidationError,
            r"environment\\n\s+Field required.*type=missing",
        ):
            wr.full_clean()

    def test_clean_workflow_data_raise_validation_error(self) -> None:
        """Check validation of workflow data."""
        wr = self.playground.create_workflow("noop")
        wr.full_clean()

        wr.task_data["spurious"] = True
        assert wr.configured_task_data is not None
        wr.configured_task_data["spurious"] = True
        with self.assertRaisesRegex(
            ValidationError, "Extra inputs are not permitted"
        ):
            wr.full_clean()

    def test_wait_without_needs_input_raise_validation_error(self) -> None:
        """WAIT tasks must specify needs_input in their workflow_data."""
        wr = self.playground.create_wait_task(
            task_name="delay",
            task_data={"delay_until": timezone.now().isoformat()},
            workflow_data=WorkRequestWorkflowData(needs_input=False),
        )
        wr.full_clean()

        wr.workflow_data_json = {}
        with self.assertRaisesRegex(
            ValidationError, "WAIT tasks must specify needs_input"
        ):
            wr.full_clean()

    def test_clean_workflow_callback_incremental(self) -> None:
        """Workflow callbacks may use the incremental unblock strategy."""
        wr = self.playground.create_internal_task("workflow")
        wr.unblock_strategy = WorkRequest.UnblockStrategy.INCREMENTAL
        wr.full_clean()

    def test_clean_non_workflow_callback_incremental(self) -> None:
        """Other work requests may not use the incremental unblock strategy."""
        for wr in (
            self.playground.create_worker_task(),
            self.playground.create_server_task(),
            self.playground.create_internal_task("synchronization_point"),
            self.playground.create_signing_task(),
            self.playground.create_wait_task(
                task_name="confirm",
                workflow_data=WorkRequestWorkflowData(needs_input=True),
            ),
        ):
            with self.subTest(
                task_type=str(wr.task_type), task_name=wr.task_name
            ):
                wr.unblock_strategy = WorkRequest.UnblockStrategy.INCREMENTAL
                with self.assertRaisesRegex(
                    ValidationError,
                    "Only workflow callbacks may use the incremental unblock"
                    " strategy",
                ):
                    wr.full_clean()

    def test_workflow_data_accessors(self) -> None:
        """Test the custom pydantic accessors for workflow_data."""
        # Empty workflow_data
        wr = self.playground.create_worker_task(task_name="noop")
        self.assertEqual(wr.workflow_data_json, {})
        self.assertEqual(wr.workflow_data, WorkRequestWorkflowData())

        wr = self.playground.create_worker_task(task_name="noop")
        self.assertEqual(wr.workflow_data_json, {})
        self.assertEqual(wr.workflow_data, WorkRequestWorkflowData())

        # Filled workflow data
        wr = self.playground.create_worker_task(
            task_name="noop",
            workflow_data=WorkRequestWorkflowData(
                display_name="Test", step="test"
            ),
        )
        self.assertEqual(
            wr.workflow_data_json,
            {"display_name": "Test", "step": "test"},
        )
        self.assertEqual(
            wr.workflow_data,
            WorkRequestWorkflowData(display_name="Test", step="test"),
        )

        wr.workflow_data = WorkRequestWorkflowData()
        self.assertEqual(wr.workflow_data_json, {})
        self.assertEqual(wr.workflow_data, WorkRequestWorkflowData())

        wr.workflow_data = WorkRequestWorkflowData(
            display_name="Test1", step="test1"
        )
        self.assertEqual(
            wr.workflow_data_json,
            {"display_name": "Test1", "step": "test1"},
        )
        self.assertEqual(
            wr.workflow_data,
            WorkRequestWorkflowData(display_name="Test1", step="test1"),
        )

    def test_has_role(self) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        other_user = self.playground.create_user("other")
        wr = self.playground.create_worker_task()
        OWNER = WorkRequest.Roles.OWNER
        CREATOR = WorkRequest.Roles.CREATOR
        CONTRIBUTOR = WorkRequest.Roles.CONTRIBUTOR
        VIEWER = WorkRequest.Roles.VIEWER
        for scope_roles, workspace_roles, creator, roles, queries in (
            ([], [], False, [], 4),
            ([], [], True, [CREATOR], 3),
            ([], [Workspace.Roles.VIEWER], False, [VIEWER], 4),
            (
                [Scope.Roles.OWNER],
                [],
                False,
                [OWNER, CREATOR, CONTRIBUTOR, VIEWER],
                4,
            ),
            (
                [],
                [Workspace.Roles.OWNER],
                False,
                [OWNER, CREATOR, CONTRIBUTOR, VIEWER],
                4,
            ),
            (
                [],
                [Workspace.Roles.CONTRIBUTOR],
                False,
                [CONTRIBUTOR, VIEWER],
                4,
            ),
        ):
            with self.subTest(
                scope_roles=str(scope_roles),
                workspace_roles=str(workspace_roles),
                creator=creator,
            ):
                Group.objects.all().delete()
                groups: list[Group] = [
                    self.playground.create_group_role(
                        self.scenario.scope, sr, users=[]
                    )
                    for sr in scope_roles
                ] + [
                    self.playground.create_group_role(
                        self.scenario.workspace, wr, users=[]
                    )
                    for wr in workspace_roles
                ]
                for g in groups:
                    g.users.add(self.scenario.user)
                if creator:
                    wr.created_by = self.scenario.user
                else:
                    wr.created_by = other_user
                wr.save()

                with self.subTest(type="query"), self.assertNumQueries(queries):
                    self.assertHasRoles(roles, wr)
                if VIEWER in roles:
                    with self.subTest(type="context"), context.local():
                        self.scenario.set_current()
                        with self.assertNumQueries(0):
                            self.assertHasRoles(roles, wr)

    def test_has_role_extra_groups(self) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        self.workflow.refresh_from_db()
        # These tests prefetch related fields so they don't generate spurious
        # queries in the following tests. From Django 5.1 onwards, we can use
        # from_queryset:
        # self.workflow.refresh_from_db(
        #     from_queryset=WorkRequest.objects.select_related(
        #         "created_by", "workspace","workflow_root__workflow_template"
        #     )
        # )
        self.assertEqual(self.workflow.workspace, self.scenario.workspace)
        self.assertEqual(self.workflow.created_by, self.scenario.user)
        assert self.workflow.workflow_root
        self.assertEqual(
            self.workflow.workflow_root.workflow_template, self.template
        )
        OWNER = WorkRequest.Roles.OWNER
        CREATOR = WorkRequest.Roles.CREATOR
        CONTRIBUTOR = WorkRequest.Roles.CONTRIBUTOR
        VIEWER = WorkRequest.Roles.VIEWER
        for scope_roles, workspace_roles, roles, queries in (
            ([], [], [CREATOR], 3),
            ([], [Workspace.Roles.VIEWER], [CREATOR, VIEWER], 3),
            ([Scope.Roles.OWNER], [], [OWNER, CREATOR, CONTRIBUTOR, VIEWER], 3),
            (
                [],
                [Workspace.Roles.OWNER],
                [OWNER, CREATOR, CONTRIBUTOR, VIEWER],
                3,
            ),
            (
                [],
                [Workspace.Roles.CONTRIBUTOR],
                [CREATOR, CONTRIBUTOR, VIEWER],
                3,
            ),
        ):
            with self.subTest(
                scope_roles=str(scope_roles),
                workspace_roles=str(workspace_roles),
            ):
                Group.objects.all().delete()
                groups: list[Group] = [
                    self.playground.create_group_role(
                        self.scenario.scope, sr, users=[]
                    )
                    for sr in scope_roles
                ] + [
                    self.playground.create_group_role(
                        self.scenario.workspace, wr, users=[]
                    )
                    for wr in workspace_roles
                ]

                if groups:
                    self.template.extra_groups.set(groups)
                else:
                    self.template.extra_groups.clear()
                pc_extra_groups = PermissionContext.for_work_request(
                    work_request=self.workflow
                )
                with (
                    self.subTest(type="extragroup"),
                    self.assertNumQueries(queries),
                ):
                    self.assertHasRoles(
                        roles, self.workflow, pc=pc_extra_groups
                    )
                if VIEWER in roles:
                    with (
                        self.subTest(type="extragroup context"),
                        context.local(),
                    ):
                        context.set_permission_context_scoped(
                            self.scenario.scope, pc_extra_groups
                        )
                        self.scenario.workspace.set_current()
                        with self.assertNumQueries(0):
                            self.assertHasRoles(
                                roles, self.workflow, pc=pc_extra_groups
                            )

    def test_has_role_extra_groups_from_context(self) -> None:
        wr = self.playground.create_worker_task()
        pc = self.playground.create_pc_with_extra_groups(
            self.scenario.workspace_owners
        )
        context.set_permission_context_scoped(scope=self.scenario.scope, pc=pc)
        self.scenario.workspace.set_current()
        self.assertHasRole(
            WorkRequest.Roles.OWNER, wr, pc=pc, hits_database=False
        )

    def test_supersedes(self) -> None:
        """Test behaviour of the supersedes field."""
        wr = self.playground.create_worker_task(task_name="noop")
        self.assertIsNone(wr.supersedes)
        self.assertFalse(hasattr(wr, "superseded"))

        wr1 = self.playground.create_worker_task(task_name="noop")
        wr1.supersedes = wr
        wr1.save()

        wr.refresh_from_db()
        wr1.refresh_from_db()

        self.assertIsNone(wr.supersedes)
        self.assertEqual(wr1.supersedes, wr)
        self.assertEqual(wr.superseded, wr1)
        self.assertFalse(hasattr(wr1, "superseded"))

        self.assertQuerySetEqual(
            WorkRequest.objects.filter(superseded__isnull=False), [wr]
        )
        self.assertQuerySetEqual(
            WorkRequest.objects.filter(supersedes__isnull=False), [wr1]
        )

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_verify_retry_superseded(self) -> None:
        """Superseded work requests cannot be retried."""
        self.scenario.set_current()
        wr = self.workflow.create_child_worker("noop")
        wr.status = WorkRequest.Statuses.ABORTED
        wr.save()
        self.assertTrue(wr.verify_retry())

        wr1 = self.workflow.create_child_worker("noop")
        wr1.supersedes = wr
        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(
            CannotRetry, r"^Cannot retry old superseded tasks$"
        ):
            wr.retry()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_verify_retry_aborted_failed(self) -> None:
        """Only aborted/failed work requests can be retried."""
        self.scenario.set_current()
        wr = self.workflow.create_child_worker("noop")
        wr.status = WorkRequest.Statuses.COMPLETED
        wr.result = WorkRequest.Results.SUCCESS
        wr.save()

        msg = r"^Only aborted or failed tasks can be retried$"

        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

        wr.status = WorkRequest.Statuses.COMPLETED
        wr.result = WorkRequest.Results.ERROR
        wr.save()
        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

        wr.status = WorkRequest.Statuses.PENDING
        wr.result = WorkRequest.Results.NONE
        wr.save()
        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

        wr.status = WorkRequest.Statuses.RUNNING
        wr.result = WorkRequest.Results.NONE
        wr.save()
        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

        wr.status = WorkRequest.Statuses.BLOCKED
        wr.result = WorkRequest.Results.NONE
        wr.save()
        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_verify_retry_task_type(self) -> None:
        """Only some task types can be retried."""
        self.scenario.set_current()
        wr = self.workflow.create_child_worker("noop")
        wr.status = WorkRequest.Statuses.COMPLETED
        wr.result = WorkRequest.Results.SUCCESS
        wr.save()

        msg = (
            r"^Only workflow, worker, internal, or signing tasks"
            " can be retried$"
        )

        wr.task_type = TaskTypes.SERVER
        wr.status = WorkRequest.Statuses.ABORTED
        wr.result = WorkRequest.Results.NONE
        wr.save()
        self.assertFalse(wr.verify_retry())
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_verify_retry_invalid_task_data(self) -> None:
        """Work request with invalid task data cannot be retried."""
        self.scenario.set_current()
        wr = self.workflow.create_child_worker("noop")
        wr.status = WorkRequest.Statuses.ABORTED
        wr.result = WorkRequest.Results.NONE
        wr.task_name = "sbuild"
        wr.save()
        self.assertFalse(wr.verify_retry())
        msg = "^Task dependencies cannot be satisfied$"
        with self.assertRaisesRegex(CannotRetry, msg):
            wr.retry()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_verify_retry_if_lookups_fail(self) -> None:
        """Work requests with unsatisfiable lookups can't retry."""
        self.scenario.set_current()
        wr = self.workflow.create_child_worker("noop")
        wr.status = WorkRequest.Statuses.ABORTED
        wr.result = WorkRequest.Results.NONE
        wr.save()

        wr = self.playground.create_worker_task(
            task_name="noop", mark_aborted=True
        )

        with mock.patch(
            "debusine.db.tasks.DBProxyTask.compute_dynamic_data",
            side_effect=LookupError("expected failure"),
        ):
            self.assertFalse(wr.verify_retry())
            msg = r"^Task dependencies cannot be satisfied$"
            with self.assertRaisesRegex(CannotRetry, msg):
                wr.retry()

        self.assertTrue(wr.verify_retry())
        wr.retry()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_workflow_callback_blocked_deps(self) -> None:
        """A retried workflow callback is BLOCKED if it has unfinished deps."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        dep = self.workflow.create_child_worker("noop")
        wr = WorkRequest.objects.create_workflow_callback(
            parent=self.workflow, step="test"
        )
        wr.add_dependency(dep)
        self.playground.advance_work_request(wr, mark_aborted=True)

        wr_new = wr.retry()

        self.assertEqual(wr_new.pk, wr.pk)
        self.assertEqual(wr_new.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr_new.result, WorkRequest.Results.NONE)
        self.assertEqual(wr_new.unblock_strategy, wr.unblock_strategy)
        self.assertIsNone(wr_new.supersedes)
        self.assertFalse(hasattr(wr_new, "superseded"))

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_workflow_callback_pending_incremental(self) -> None:
        """A retried workflow callback is PENDING if it is incremental."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = WorkRequest.objects.create_workflow_callback(
            parent=self.workflow,
            unblock_strategy=WorkRequest.UnblockStrategy.INCREMENTAL,
            step="test",
        )
        self.playground.advance_work_request(wr, mark_aborted=True)

        wr_new = wr.retry()

        self.assertEqual(wr_new.pk, wr.pk)
        self.assertEqual(wr_new.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(wr_new.result, WorkRequest.Results.NONE)
        self.assertEqual(wr_new.unblock_strategy, wr.unblock_strategy)
        self.assertIsNone(wr_new.supersedes)
        self.assertFalse(hasattr(wr_new, "superseded"))

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede(self) -> None:
        """Test behaviour of retry, superseding work requests."""
        self.scenario.set_current()
        # An aborted workflow, with a failed task and one depending on it
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()

        wr_failed = self.workflow.create_child_worker("noop")
        wr_failed.status = WorkRequest.Statuses.COMPLETED
        wr_failed.result = WorkRequest.Results.FAILURE
        wr_failed.configured_task_data = {"result": WorkRequest.Results.FAILURE}
        wr_failed.save()

        # Two reverse dependencies
        wr_dep1 = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr_dep1, mark_pending=True)
        wr_dep1.dependencies.add(wr_failed)
        wr_dep2 = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(
            wr_dep2, result=WorkRequest.Results.SUCCESS
        )
        wr_dep2.dependencies.add(wr_failed)
        # One forward dependency
        wr_fdep1 = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(
            wr_fdep1, result=WorkRequest.Results.SUCCESS
        )
        wr_failed.dependencies.add(wr_fdep1)

        # wr_failed gets retried
        self.assertTrue(wr_failed.verify_retry())
        wr_new = wr_failed.retry(reason=WorkRequestRetryReason.WORKER_FAILED)

        self.assertEqual(wr_new.workspace, wr_failed.workspace)
        self.assertGreater(wr_new.created_at, wr_failed.created_at)
        self.assertIsNotNone(wr_new.pending_at)
        self.assertIsNone(wr_new.started_at)
        self.assertIsNone(wr_new.completed_at)
        self.assertEqual(wr_new.created_by, wr_failed.created_by)
        self.assertEqual(wr_new.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(wr_new.result, WorkRequest.Results.NONE)
        self.assertIsNone(wr_new.worker)
        self.assertEqual(wr_new.task_type, wr_failed.task_type)
        self.assertEqual(wr_new.task_name, wr_failed.task_name)
        self.assertEqual(wr_new.task_data, wr_failed.task_data)
        self.assertIsNone(wr_new.dynamic_task_data)
        # The new WorkRequest is reconfigured before running
        assert wr_new.configured_task_data is not None
        self.assertNotIn("result", wr_new.configured_task_data)
        self.assertNotIn("FAILED_VERSION", wr_new.configured_task_data)
        self.assertEqual(wr_new.priority_base, wr_failed.priority_base)
        self.assertEqual(
            wr_new.priority_adjustment, wr_failed.priority_adjustment
        )
        self.assertIsNone(wr_new.output_data_json)
        self.assertEqual(wr_new.unblock_strategy, wr_failed.unblock_strategy)
        self.assertEqual(wr_new.parent, self.workflow)
        self.assertEqual(wr_new.workflow_root, self.workflow)
        self.assertEqual(wr_new.workflow_template, wr_failed.workflow_template)
        self.assertEqual(
            wr_new.workflow_data_json,
            {**wr_failed.workflow_data_json, "retry_count": 1},
        )
        self.assertEqual(
            wr_new.event_reactions_json, wr_failed.event_reactions_json
        )
        self.assertIsNone(wr_new.internal_collection)
        self.assertEqual(wr_new.expiration_delay, wr_failed.expiration_delay)
        self.assertEqual(wr_new.supersedes, wr_failed)
        self.assertEqual(wr_failed.superseded, wr_new)

        self.assertQuerySetEqual(wr_new.dependencies.all(), [wr_fdep1])

        self.assertQuerySetEqual(wr_dep1.dependencies.all(), [wr_new])
        wr_dep1.refresh_from_db()
        self.assertEqual(wr_dep1.status, WorkRequest.Statuses.BLOCKED)

        self.assertQuerySetEqual(wr_dep2.dependencies.all(), [wr_new])
        wr_dep2.refresh_from_db()
        self.assertEqual(wr_dep2.status, WorkRequest.Statuses.COMPLETED)

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(self.workflow.result, WorkRequest.Results.NONE)
        self.assertIsNone(self.workflow.completed_at)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede_blocked_by_dep(self) -> None:
        """A retried work request is BLOCKED if it has incomplete deps."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_aborted=True)
        wr_dep = self.workflow.create_child_worker("noop")
        wr.dependencies.add(wr_dep)

        wr_new = wr.retry()

        self.assertEqual(wr_new.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr_new.result, WorkRequest.Results.NONE)
        self.assertEqual(wr_new.unblock_strategy, wr.unblock_strategy)
        self.assertEqual(wr_new.supersedes, wr)
        self.assertEqual(wr.superseded, wr_new)
        self.assertQuerySetEqual(wr_new.dependencies.all(), [wr_dep])

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede_blocked_manual(self) -> None:
        """A retried work request is BLOCKED if it needs manual unblocking."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_aborted=True)
        wr.unblock_strategy = WorkRequest.UnblockStrategy.MANUAL
        wr.save()

        wr_new = wr.retry()

        self.assertEqual(wr_new.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr_new.result, WorkRequest.Results.NONE)
        self.assertEqual(wr_new.unblock_strategy, wr.unblock_strategy)
        self.assertEqual(wr_new.supersedes, wr)
        self.assertEqual(wr.superseded, wr_new)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede_worker_failed_checks_retry_count(self) -> None:
        """Retrying a task (worker failed) checks its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = self.workflow.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                retry_count=MAX_AUTOMATIC_RETRIES
            ),
        )
        self.playground.advance_work_request(wr, mark_aborted=True)

        with self.assertRaisesRegex(
            CannotRetry, "Maximum number of automatic retries exceeded"
        ):
            wr.retry(reason=WorkRequestRetryReason.WORKER_FAILED)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede_worker_failed_increments_retry_count(self) -> None:
        """Retrying a task (worker failure) increments its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = self.workflow.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(retry_count=1),
        )
        self.playground.advance_work_request(wr, mark_aborted=True)

        wr_new = wr.retry(reason=WorkRequestRetryReason.WORKER_FAILED)

        self.assertEqual(wr_new.workflow_data_json, {"retry_count": 2})

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede_delay_increments_retry_count(self) -> None:
        """Retrying a task (delay) increments its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = self.workflow.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                retry_count=MAX_AUTOMATIC_RETRIES
            ),
        )
        self.playground.advance_work_request(wr, mark_aborted=True)

        wr_new = wr.retry(reason=WorkRequestRetryReason.DELAY)

        self.assertEqual(
            wr_new.workflow_data_json,
            {"retry_count": MAX_AUTOMATIC_RETRIES + 1},
        )

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_supersede_manual_resets_retry_count(self) -> None:
        """Retrying a task (manual) resets its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.result = WorkRequest.Results.FAILURE
        self.workflow.completed_at = timezone.now()
        self.workflow.save()
        wr = self.workflow.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                retry_count=MAX_AUTOMATIC_RETRIES
            ),
        )
        self.playground.advance_work_request(wr, mark_aborted=True)

        wr_new = wr.retry()

        self.assertEqual(wr_new.workflow_data_json, {"retry_count": 0})

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_in_running_workflow(self) -> None:
        """Test retrying a task in a running workflow."""
        self.scenario.set_current()
        # An aborted workflow, with a failed task and one depending on it
        now = timezone.now()
        self.workflow.status = WorkRequest.Statuses.RUNNING
        # Bogus value to check that it does not get reset
        self.workflow.completed_at = now
        self.workflow.save()

        wr_failed = self.workflow.create_child_worker("noop")
        wr_failed.status = WorkRequest.Statuses.COMPLETED
        wr_failed.result = WorkRequest.Results.FAILURE
        wr_failed.save()

        # wr_failed gets retried
        wr_failed.retry()

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(self.workflow.completed_at, now)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_in_failed_workflow(self) -> None:
        """Test retrying a task in a failed workflow."""
        self.scenario.set_current()
        # An aborted workflow, with a failed task and one depending on it
        now = timezone.now()
        self.workflow.status = WorkRequest.Statuses.COMPLETED
        self.workflow.result = WorkRequest.Results.FAILURE
        # Bogus value to check that it does not get reset
        self.workflow.completed_at = now
        self.workflow.save()

        wr_failed = self.workflow.create_child_worker("noop")
        wr_failed.status = WorkRequest.Statuses.COMPLETED
        wr_failed.result = WorkRequest.Results.FAILURE
        wr_failed.save()

        # wr_failed gets retried
        wr_failed.retry()

        # The workflow is set to running again
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)
        self.assertIsNone(self.workflow.completed_at)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_force_status_in_running_workflow(self) -> None:
        """Forcibly retry a running task in a running workflow."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.RUNNING
        # Allow dependency failures so that aborting all children of the
        # workflow would normally mark it as completed.
        self.workflow.workflow_data = WorkRequestWorkflowData(
            allow_dependency_failures=True
        )
        self.workflow.save()

        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_running=True)

        wr_new = wr.retry(
            reason=WorkRequestRetryReason.WORKER_FAILED, force_status=True
        )

        self.assertEqual(wr.status, WorkRequest.Statuses.ABORTED)
        self.assertEqual(wr_new.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(wr_new.supersedes, wr)
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_in_completed_workflow(self) -> None:
        """Test retrying a task in a completed workflow."""
        self.scenario.set_current()
        # A completed workflow, with a failed task and one depending on it
        now = timezone.now()
        self.workflow.status = WorkRequest.Statuses.COMPLETED
        self.workflow.result = WorkRequest.Results.SUCCESS
        # Bogus value to check that it does not get reset
        self.workflow.completed_at = now
        self.workflow.save()

        wr_failed = self.workflow.create_child_worker("noop")
        wr_failed.status = WorkRequest.Statuses.COMPLETED
        wr_failed.result = WorkRequest.Results.FAILURE
        wr_failed.save()

        # wr_failed gets retried
        wr_failed.retry()

        # The workflow is set to running again
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)
        self.assertIsNone(self.workflow.completed_at)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_retry_empty_workflow(self) -> None:
        """Test behaviour of retrying for an empty workflow."""
        with transaction.atomic():
            self.scenario.set_current()
            self.workflow.status = WorkRequest.Statuses.ABORTED
            self.workflow.save()

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.workflow.retry()

        # Retrying a workflow does not create a new work request
        self.assertEqual(wr.pk, self.workflow.pk)

        # The scheduler will run the workflow soon.
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(self.workflow.result, WorkRequest.Results.NONE)
        mock_schedule.assert_called_once()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_retry_workflow(self) -> None:
        """Test behaviour of retrying for a non-empty workflow."""
        with transaction.atomic():
            self.scenario.set_current()
            wr_completed = self.workflow.create_child_worker("noop")
            wr_aborted = self.workflow.create_child_worker("noop")

            # Simulate the previous workflow lifecycle.
            self.workflow.mark_running()
            self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)
            self.workflow.unblock_workflow_children()
            wr_completed.refresh_from_db()
            wr_completed.mark_completed(WorkRequest.Results.SUCCESS)
            wr_completed.completed_at = (
                orig_completed := timezone.now() - dt.timedelta(days=7)
            )
            wr_completed.save()
            wr_aborted.refresh_from_db()
            wr_aborted.mark_aborted()

            # The workflow ended up as aborted.
            self.workflow.refresh_from_db()
            self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)

        # Retry it
        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.workflow.retry(
                reason=WorkRequestRetryReason.WORKER_FAILED
            )
        self.assertEqual(wr.pk, self.workflow.pk)

        self.workflow.refresh_from_db()
        wr_completed.refresh_from_db()
        wr_aborted.refresh_from_db()

        # The scheduler will run the workflow soon.
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(self.workflow.result, WorkRequest.Results.NONE)
        self.assertEqual(
            self.workflow.workflow_data_json,
            {"workflow_template_name": "test", "retry_count": 1},
        )
        mock_schedule.assert_called_once()

        self.assertEqual(wr_completed.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(wr_completed.result, WorkRequest.Results.SUCCESS)
        # Successful work requests did not get restarted
        self.assertEqual(wr_completed.completed_at, orig_completed)
        self.assertEqual(wr_aborted.status, WorkRequest.Statuses.ABORTED)
        self.assertEqual(wr_aborted.result, WorkRequest.Results.NONE)
        wr_aborted_new = wr_aborted.superseded
        self.assertEqual(wr_aborted_new.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(wr_aborted_new.result, WorkRequest.Results.NONE)
        self.assertEqual(wr_aborted_new.workflow_data.retry_count, 1)

        # Complete the one running task
        self.playground.advance_work_request(self.workflow, mark_running=True)
        self.playground.advance_work_request(
            wr_aborted_new, result=WorkRequest.Results.SUCCESS
        )
        self.workflow.refresh_from_db()

        # Workflow completed successfully
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(self.workflow.result, WorkRequest.Results.SUCCESS)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_retry_workflow_calls(self) -> None:
        """Test behaviour of retrying for an empty workflow."""
        with transaction.atomic():
            self.scenario.set_current()
            wr_aborted = self.workflow.create_child_worker("noop")

            # Simulate the previous workflow lifecycle.
            self.workflow.mark_running()
            self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)
            self.workflow.unblock_workflow_children()
            wr_aborted.refresh_from_db()
            wr_aborted.mark_aborted()

            # The workflow ended up as aborted.
            self.workflow.refresh_from_db()
            self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            self.workflow.retry()

        mock_schedule.assert_called_once()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_retry_unconfigured_workflow(self) -> None:
        """Test behaviour of retrying a workflow that was aborted very early."""
        with transaction.atomic():
            self.scenario.set_current()
            workflow = self.playground.create_workflow(mark_pending=False)
            workflow.mark_aborted()

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            workflow.retry()

        mock_schedule.assert_called_once()

    @override_permission(WorkRequest, "can_retry", AllowAll)
    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_retry_workflow_child_fails(self) -> None:
        """Test retrying a workflow with a child that cannot retry."""
        with transaction.atomic():
            self.scenario.set_current()
            wr_aborted1 = self.workflow.create_child_worker("noop")
            wr_unretriable = self.workflow.create_child_worker("noop")
            wr_aborted2 = self.workflow.create_child_worker("noop")

            self.workflow.mark_running()
            self.workflow.unblock_workflow_children()
            wr_aborted1.refresh_from_db()
            wr_unretriable.refresh_from_db()
            wr_aborted2.refresh_from_db()
            wr_aborted1.mark_aborted()
            wr_unretriable.mark_aborted()
            wr_aborted2.mark_aborted()

            # The workflow ended up as aborted
            self.workflow.refresh_from_db()
            self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)

        # Make wr_unretriable unretriable
        orig_check_retry = WorkRequest._verify_retry

        def test_check_retry(self: WorkRequest) -> None:
            if self.pk == wr_unretriable.pk:
                raise CannotRetry("expected failure")
            else:
                orig_check_retry(self)

        try:
            with transaction.atomic():
                with (
                    mock.patch("debusine.server.scheduler.schedule"),
                    self.captureOnCommitCallbacks(execute=True),
                    mock.patch(
                        "debusine.db.models.WorkRequest._verify_retry",
                        side_effect=test_check_retry,
                        autospec=True,
                    ),
                    self.assertRaises(CannotRetry) as exc,
                ):
                    self.workflow.retry()

                self.assertEqual(
                    exc.exception.args[0],
                    f"child work request {wr_unretriable.pk} cannot be retried",
                )
                self.assertIsNotNone(exc.exception.__context__)
                assert exc.exception.__context__ is not None
                self.assertEqual(
                    exc.exception.__context__.args[0], "expected failure"
                )

                self.workflow.refresh_from_db()
                wr_aborted1.refresh_from_db()
                wr_aborted1_new = wr_aborted1.superseded
                wr_unretriable.refresh_from_db()
                wr_aborted2.refresh_from_db()
                self.assertFalse(hasattr(wr_aborted2, "superseded"))
                self.assertEqual(
                    self.workflow.status, WorkRequest.Statuses.PENDING
                )
                self.assertEqual(
                    wr_aborted1_new.status, WorkRequest.Statuses.BLOCKED
                )
                self.assertEqual(
                    wr_unretriable.status, WorkRequest.Statuses.ABORTED
                )
                self.assertEqual(
                    wr_aborted2.status, WorkRequest.Statuses.ABORTED
                )
                # Raise to roll back the transaction
                raise exc.exception
        except CannotRetry as exc1:
            self.assertIs(exc1, exc.exception)

        self.workflow.refresh_from_db()
        wr_aborted1.refresh_from_db()
        wr_unretriable.refresh_from_db()
        wr_aborted2.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)
        self.assertEqual(wr_aborted1.status, WorkRequest.Statuses.ABORTED)
        self.assertEqual(wr_unretriable.status, WorkRequest.Statuses.ABORTED)
        self.assertEqual(wr_aborted2.status, WorkRequest.Statuses.ABORTED)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_workflow_worker_failed_checks_retry_count(self) -> None:
        """Automatically retrying a workflow checks its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.workflow_data = WorkRequestWorkflowData(
            retry_count=MAX_AUTOMATIC_RETRIES
        )
        self.workflow.save()

        with self.assertRaisesRegex(
            CannotRetry, "Maximum number of automatic retries exceeded"
        ):
            self.workflow.retry(reason=WorkRequestRetryReason.WORKER_FAILED)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_workflow_worker_failed_increments_retry_count(self) -> None:
        """Retrying a workflow (worker failed) increments its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.workflow_data = WorkRequestWorkflowData(retry_count=1)
        self.workflow.save()
        wr = self.workflow.retry(reason=WorkRequestRetryReason.WORKER_FAILED)
        self.assertEqual(wr.pk, self.workflow.pk)
        self.assertEqual(wr.workflow_data_json, {"retry_count": 2})

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_workflow_delay_increments_retry_count(self) -> None:
        """Retrying a workflow (delay) increments its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.workflow_data = WorkRequestWorkflowData(
            retry_count=MAX_AUTOMATIC_RETRIES
        )
        self.workflow.save()
        wr = self.workflow.retry(reason=WorkRequestRetryReason.DELAY)
        self.assertEqual(wr.pk, self.workflow.pk)
        self.assertEqual(
            wr.workflow_data_json, {"retry_count": MAX_AUTOMATIC_RETRIES + 1}
        )

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_workflow_manual_resets_retry_count(self) -> None:
        """Retrying a workflow (manual) resets its retry count."""
        self.scenario.set_current()
        self.workflow.status = WorkRequest.Statuses.ABORTED
        self.workflow.workflow_data = WorkRequestWorkflowData(
            retry_count=MAX_AUTOMATIC_RETRIES
        )
        self.workflow.save()
        wr = self.workflow.retry()
        self.assertEqual(wr.pk, self.workflow.pk)
        self.assertEqual(wr.workflow_data_json, {"retry_count": 0})

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_update_debusine_promise_work_request_id(self) -> None:
        """Test retrying updates promise of the expected work request id."""
        self.scenario.set_current()
        user = self.playground.get_default_user()

        template = WorkflowTemplate.objects.create(
            name="retry-update-debusine-promise-work-request-id",
            workspace=default_workspace(),
            task_name="noop",
        )
        workflow = self.playground.create_workflow(template, task_data={})

        collection = workflow.internal_collection
        assert collection is not None

        wr_to_retry = self.playground.create_worker_task(
            parent=workflow,
            task_name="noop",
        )
        assert wr_to_retry.parent_id is not None
        wr_aborted = self.playground.create_worker_task(
            parent=workflow,
            task_name="noop",
        )

        wr_to_retry.mark_aborted()
        wr_aborted.mark_aborted()

        debusine_promise_updated = collection.manager.add_bare_data(
            category=BareDataCategory.PROMISE,
            name="test-1",
            data=DebusinePromise(
                promise_work_request_id=wr_to_retry.id,
                promise_workflow_id=wr_to_retry.parent_id,
                promise_category="something",
            ),
            user=user,
        )
        debusine_promise_not_updated = collection.manager.add_bare_data(
            category=BareDataCategory.PROMISE,
            name="test-2",
            data=DebusinePromise(
                promise_work_request_id=wr_aborted.id,
                promise_workflow_id=wr_to_retry.parent_id,
                promise_category="something",
            ),
            user=user,
        )

        wr_supersedes = wr_to_retry.retry()

        debusine_promise_updated.refresh_from_db()
        debusine_promise_not_updated.refresh_from_db()

        # Updated to the new promise_work_request_id
        self.assertEqual(
            debusine_promise_updated.data["promise_work_request_id"],
            wr_supersedes.id,
        )

        # Still the old promise_work_request_id
        self.assertEqual(
            debusine_promise_not_updated.data["promise_work_request_id"],
            wr_aborted.id,
        )

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_reverse_dependencies_workflow_callbacks(self) -> None:
        """Retrying a work request retries dependent workflow callbacks."""
        self.scenario.set_current()
        self.playground.advance_work_request(self.workflow, mark_running=True)
        parent = self.workflow.create_child_workflow("noop")
        self.playground.advance_work_request(parent, mark_running=True)
        wr1 = parent.create_child_workflow("noop")
        self.playground.advance_work_request(wr1, mark_running=True)
        wr1_1 = wr1.create_child_worker("noop")
        wr1_2 = WorkRequest.objects.create_workflow_callback(
            parent=wr1, step="local-analysis"
        )
        wr1_2.add_dependency(wr1_1)
        wr2 = parent.create_child_worker("noop")
        wr2.add_dependency(wr1_1)
        wr3 = WorkRequest.objects.create_workflow_callback(
            parent=parent, step="global-analysis"
        )
        wr3.add_dependency(wr1_2)

        wr1_1.mark_aborted()

        wr1_2.refresh_from_db()
        self.assertEqual(wr1_2.status, WorkRequest.Statuses.ABORTED)
        wr1.refresh_from_db()
        self.assertEqual(wr1.status, WorkRequest.Statuses.ABORTED)
        wr2.refresh_from_db()
        self.assertEqual(wr2.status, WorkRequest.Statuses.ABORTED)
        wr3.refresh_from_db()
        self.assertEqual(wr3.status, WorkRequest.Statuses.ABORTED)
        parent.refresh_from_db()
        self.assertEqual(parent.status, WorkRequest.Statuses.ABORTED)
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)

        wr1_1_new = wr1_1.retry()

        self.assertEqual(wr1_1_new.status, WorkRequest.Statuses.PENDING)
        # Dependent workflow callbacks are recursively retried.
        wr1_2.refresh_from_db()
        self.assertEqual(wr1_2.status, WorkRequest.Statuses.BLOCKED)
        wr3.refresh_from_db()
        self.assertEqual(wr3.status, WorkRequest.Statuses.BLOCKED)
        # Other dependent work requests are left alone.
        wr2.refresh_from_db()
        self.assertEqual(wr2.status, WorkRequest.Statuses.ABORTED)
        # Containing workflows are retried.
        wr1.refresh_from_db()
        self.assertEqual(wr1.status, WorkRequest.Statuses.RUNNING)
        parent.refresh_from_db()
        self.assertEqual(parent.status, WorkRequest.Statuses.RUNNING)
        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)

    @override_permission(WorkRequest, "can_abort", DenyAll)
    def test_verify_abort_no_permission(self) -> None:
        """Aborting a work request without permission fails."""
        wr = self.workflow.create_child_worker("noop")
        self.scenario.set_current()

        self.assertFalse(wr.verify_abort())
        with self.assertRaisesRegex(
            PermissionDenied, rf"{self.scenario.user} cannot abort {wr}"
        ):
            wr.abort()

    @override_permission(WorkRequest, "can_abort", AllowAll)
    def test_verify_abort_superseded(self) -> None:
        """Superseded work requests cannot be aborted."""
        wr = self.workflow.create_child_worker("noop")
        self.scenario.set_current()

        self.assertTrue(wr.verify_abort())

        wr1 = self.workflow.create_child_worker("noop")
        wr1.supersedes = wr
        self.assertFalse(wr.verify_abort())
        with self.assertRaisesRegex(
            CannotAbort, r"^Cannot abort old superseded tasks$"
        ):
            wr.abort()

    @override_permission(WorkRequest, "can_abort", AllowAll)
    def test_verify_abort_pending_blocked_running(self) -> None:
        """Only pending/blocked/running work requests can be aborted."""
        wr = self.workflow.create_child_worker("noop")
        self.scenario.set_current()
        msg = r"^Only pending, blocked, or running tasks can be aborted$"

        wr.status = WorkRequest.Statuses.COMPLETED
        wr.result = WorkRequest.Results.SUCCESS
        wr.save()
        self.assertFalse(wr.verify_abort())
        with self.assertRaisesRegex(CannotAbort, msg):
            wr.abort()

        wr.status = WorkRequest.Statuses.ABORTED
        wr.result = WorkRequest.Results.NONE
        wr.save()
        self.assertFalse(wr.verify_abort())
        with self.assertRaisesRegex(CannotAbort, msg):
            wr.abort()

    @override_permission(WorkRequest, "can_abort", AllowAll)
    def test_abort_non_workflow(self) -> None:
        """A non-workflow work request can be aborted."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, assign_new_worker=True)
        self.scenario.set_current()

        self.assertTrue(wr.verify_abort())
        wr.abort()

        wr.refresh_from_db()
        self.assertEqual(wr.aborted_by, self.scenario.user)
        self.assertEqual(wr.status, WorkRequest.Statuses.ABORTED)
        self.assertIsNone(wr.worker)

    @override_permission(WorkRequest, "can_abort", AllowAll)
    def test_abort_workflow(self) -> None:
        """Aborting a workflow aborts all its child work requests as well."""
        child1 = self.workflow.create_child_workflow("noop")
        grandchild1 = child1.create_child_worker("noop")
        grandchild2 = child1.create_child_worker("noop")
        grandchild3 = child1.create_child_worker("noop")
        self.playground.advance_work_request(grandchild3, mark_running=True)
        grandchild3.mark_completed(result=WorkRequest.Results.SUCCESS)
        child2 = self.workflow.create_child_worker("noop")
        child3 = WorkRequest.objects.create_workflow_callback(
            parent=self.workflow, step="invisible", visible=False
        )
        self.playground.advance_work_request(
            grandchild1, assign_new_worker=True
        )
        self.scenario.set_current()

        self.assertTrue(self.workflow.verify_abort())
        self.workflow.abort()

        for work_request in (
            self.workflow,
            child1,
            grandchild1,
            grandchild2,
            child2,
            child3,
        ):
            work_request.refresh_from_db()
            self.assertEqual(work_request.aborted_by, self.scenario.user)
            self.assertEqual(work_request.status, WorkRequest.Statuses.ABORTED)
            self.assertIsNone(work_request.worker)

        grandchild3.refresh_from_db()
        self.assertIsNone(grandchild3.aborted_by)
        self.assertEqual(grandchild3.status, WorkRequest.Statuses.COMPLETED)

    def test_is_workflow_task_type_workflow(self) -> None:
        """A WORKFLOW work request is a workflow."""
        self.assertTrue(self.workflow.is_workflow)

    def test_is_workflow_child(self) -> None:
        """A WORKER work request is not a workflow, even if it's part of one."""
        wr = self.playground.create_worker_task(parent=self.workflow)

        self.assertFalse(wr.is_workflow)

    def test_is_part_of_workflow_standalone(self) -> None:
        """A standalone work request is not part of a workflow."""
        wr = self.playground.create_worker_task()

        self.assertFalse(wr.is_part_of_workflow)

    def test_is_part_of_workflow_root(self) -> None:
        """A WORKFLOW work request with no parent is not part of a workflow."""
        self.assertFalse(self.workflow.is_part_of_workflow)

    def test_is_part_of_workflow_child(self) -> None:
        """A work request with a parent is part of a workflow."""
        wr = self.playground.create_worker_task(parent=self.workflow)

        self.assertTrue(wr.is_part_of_workflow)

    def test_get_workflow_root_standalone(self) -> None:
        """A standalone work request has no workflow root."""
        wr = self.playground.create_worker_task()

        self.assertIsNone(wr.workflow_root)

    def test_get_workflow_root_root(self) -> None:
        """A top-level work request in a workflow is its own root."""
        self.assertEqual(self.workflow.workflow_root, self.workflow)

    def test_get_workflow_root_child(self) -> None:
        """A child work request in a workflow has a root."""
        wr = self.playground.create_worker_task(parent=self.workflow)

        self.assertEqual(wr.workflow_root, self.workflow)

    def test_get_workflow_root_grandchild(self) -> None:
        """A grandchild work request in a workflow has a root."""
        root = self.workflow
        child_template = WorkflowTemplate.objects.create(
            name="child",
            workspace=default_workspace(),
            task_name="noop",
            static_parameters={},
        )
        child = self.playground.create_workflow(
            child_template, task_data={}, parent=root
        )
        wr = self.playground.create_worker_task(parent=child)

        self.assertEqual(wr.workflow_root, root)

    def test_is_workflow_root(self) -> None:
        """Tests for is_workflow_root()."""
        # Standalone work request is not a workflow
        self.assertFalse(self.playground.create_worker_task().is_workflow_root)

        root_workflow = self.playground.create_workflow()
        self.assertTrue(root_workflow.is_workflow_root)

        sub_workflow = self.playground.create_workflow(parent=root_workflow)

        # A workflow in a workflow is not a root workflow
        self.assertFalse(sub_workflow.is_workflow_root)

    def test_workflow_display_name_explicit(self) -> None:
        """`workflow_display_name` returns an explicitly-set display name."""
        wr = self.playground.create_worker_task(
            workflow_data=WorkRequestWorkflowData(
                display_name="Foo", step="foo"
            )
        )

        self.assertEqual(wr.workflow_display_name, "Foo")

    def test_workflow_display_name_task_name(self) -> None:
        """`workflow_display_name` defaults to the task name."""
        wr = self.playground.create_worker_task(task_name="noop")

        self.assertEqual(wr.workflow_display_name, "noop")

    def test_requires_signature_running_external_debsign(self) -> None:
        """A running `Wait/ExternalDebsign` task requires a signature."""
        package = self.playground.create_upload_artifacts().upload
        wr = self.playground.create_wait_task(
            task_name="externaldebsign",
            task_data=ExternalDebsignData(unsigned=package.pk),
            workflow_data=WorkRequestWorkflowData(needs_input=True),
            mark_running=True,
        )

        self.assertTrue(wr.requires_signature)

    def test_requires_signature_completed_external_debsign(self) -> None:
        """A completed `Wait/ExternalDebsign` task requires no signature."""
        package = self.playground.create_upload_artifacts().upload
        wr = self.playground.create_wait_task(
            task_name="externaldebsign",
            task_data=ExternalDebsignData(unsigned=package.pk),
            workflow_data=WorkRequestWorkflowData(needs_input=True),
            mark_running=True,
            result=WorkRequest.Results.SUCCESS,
        )

        self.assertFalse(wr.requires_signature)

    def test_requires_signature_other_task(self) -> None:
        """A non-`Wait/ExternalDebsign` task requires no signature."""
        wr = self.playground.create_worker_task(task_name="noop")

        self.assertFalse(wr.requires_signature)

    def test_effective_expiration_delay(self) -> None:
        """Test WorkRequest.effective_expiration_delay() method."""
        wr = self.playground.create_worker_task(task_name="noop")

        wr.expiration_delay = dt.timedelta(days=1)
        self.assertEqual(wr.effective_expiration_delay(), dt.timedelta(days=1))

        wr.expiration_delay = dt.timedelta(days=0)
        self.assertEqual(wr.effective_expiration_delay(), dt.timedelta(days=0))

        wr.workspace = self.playground.create_workspace(
            name="test", default_expiration_delay=dt.timedelta(days=2)
        )

        wr.expiration_delay = dt.timedelta(days=1)
        self.assertEqual(wr.effective_expiration_delay(), dt.timedelta(days=1))
        self.assertEqual(wr.expire_at, wr.created_at + dt.timedelta(days=1))

        wr.expiration_delay = dt.timedelta(days=0)
        self.assertEqual(wr.effective_expiration_delay(), dt.timedelta(days=0))

        wr.expiration_delay = None
        self.assertEqual(wr.effective_expiration_delay(), dt.timedelta(days=2))

    def test_get_task(self) -> None:
        """Test get_task with various task types."""
        for factory, expected_class in (
            (self.playground.create_worker_task, Noop),
            (self.playground.create_server_task, ServerNoop),
            (self.playground.create_signing_task, SigningNoop),
            (self.playground.create_workflow, NoopWorkflow),
            (
                partial(
                    self.playground.create_wait_task,
                    task_name="delay",
                    task_data=DelayData(delay_until=timezone.now()),
                    workflow_data=WorkRequestWorkflowData(needs_input=False),
                ),
                Delay,
            ),
        ):
            with self.subTest(factory=repr(factory)):
                wr = factory()

                task = wr.get_task()
                self.assertEqual(task.work_request, wr)
                self.assertEqual(task.TASK_TYPE, expected_class.TASK_TYPE)
                self.assertEqual(task.name, expected_class.name)

    def test_get_task_internal_workflow_callback(self) -> None:
        """A workflow callback is instantiated as its workflow."""
        parent = self.playground.create_workflow(task_name="noop")
        wr = WorkRequest.objects.create_workflow_callback(
            parent=parent, step="test"
        )
        task = wr.get_task()
        self.assertIsInstance(task, NoopWorkflow)

    def test_get_task_internal_workflow_callback_outside_workflow(self) -> None:
        """A workflow callback outside a workflow cannot be instantiated."""
        wr = self.playground.create_internal_task(task_name="workflow")
        with self.assertRaisesRegex(
            InternalTaskError,
            "Workflow callback is not contained in a workflow",
        ):
            wr.get_task()

    def test_get_task_internal_not_workflow_callback(self) -> None:
        """A non-workflow-callback internal task cannot be instantiated."""
        wr = self.playground.create_internal_task(
            task_name="synchronization_point"
        )
        with self.assertRaisesRegex(
            InternalTaskError,
            "Internal tasks other than workflow callbacks cannot be "
            "instantiated",
        ):
            wr.get_task()

    def test_get_roles(self) -> None:
        other = self.playground.create_user("other")

        workspaces: dict[str, Workspace] = {
            "public": self.scenario.workspace,
            "private": self.playground.create_workspace(name="private"),
            "contributed": self.playground.create_workspace(name="contributed"),
            "owned": self.playground.create_workspace(name="owned"),
        }
        self.playground.create_group_role(
            workspaces["contributed"],
            Workspace.Roles.CONTRIBUTOR,
            users=[self.scenario.user],
        )
        self.playground.create_group_role(
            workspaces["owned"],
            Workspace.Roles.OWNER,
            users=[self.scenario.user],
        )

        work_requests: dict[str, WorkRequest] = {}
        for ws_name, ws in workspaces.items():
            work_requests[f"created_{ws_name}"] = (
                self.playground.create_worker_task(
                    workspace=ws, created_by=self.scenario.user
                )
            )
            work_requests[f"other_{ws_name}"] = (
                self.playground.create_worker_task(
                    workspace=ws, created_by=other
                )
            )

        expected: dict[str, list[WorkRequestRoles]] = {
            "created_public": [
                WorkRequest.Roles.CREATOR,
                WorkRequest.Roles.VIEWER,
            ],
            "other_public": [WorkRequest.Roles.VIEWER],
            "created_private": [WorkRequest.Roles.CREATOR],
            "other_private": [],
            "created_contributed": [
                WorkRequest.Roles.CREATOR,
                WorkRequest.Roles.CONTRIBUTOR,
            ],
            "other_contributed": [
                WorkRequest.Roles.CONTRIBUTOR,
            ],
            "created_owned": [WorkRequest.Roles.OWNER],
            "other_owned": [WorkRequest.Roles.OWNER],
        }

        base_queryset = WorkRequest.objects.filter(
            pk__in=[ws.pk for ws in work_requests.values()]
        ).select_related("workspace", "created_by")
        names_by_pk = {wr.pk: name for name, wr in work_requests.items()}
        for name, pc, queryset, num_queries in (
            (
                "plain",
                self.scenario.context_ui(),
                base_queryset,
                {
                    "created_public": 2,
                    "other_public": 3,
                    "created_private": 3,
                    "other_private": 4,
                    "created_contributed": 3,
                    "other_contributed": 4,
                    "created_owned": 3,
                    "other_owned": 4,
                },
            ),
            (
                "annotated",
                self.scenario.context_ui(),
                base_queryset.with_role_annotations(self.scenario.context_ui()),
                {
                    "created_public": 0,
                    "other_public": 0,
                    "created_private": 0,
                    "other_private": 0,
                    "created_contributed": 0,
                    "other_contributed": 0,
                    "created_owned": 0,
                    "other_owned": 0,
                },
            ),
            (
                "annotated_anonymous",
                self.scenario.context_ui(),
                base_queryset.with_role_annotations(
                    PermissionContext.for_ui_anonymous()
                ),
                {
                    "created_public": 2,
                    "other_public": 3,
                    "created_private": 3,
                    "other_private": 4,
                    "created_contributed": 3,
                    "other_contributed": 4,
                    "created_owned": 3,
                    "other_owned": 4,
                },
            ),
        ):
            with self.subTest(queryset=name):
                actual: dict[str, frozenset[WorkRequestRoles]] = {}
                for wr in queryset:
                    wr_name = names_by_pk[wr.pk]
                    with (
                        self.subTest(work_request=wr_name),
                        self.assertNumQueries(num_queries[wr_name]),
                    ):
                        actual[wr_name] = wr.get_roles(pc)

                self.assertEqual(
                    actual, {k: frozenset(v) for k, v in expected.items()}
                )

    def test_get_roles_extra_groups(self) -> None:
        workspaces: dict[str, Workspace] = {
            "public": self.scenario.workspace,
            "private": self.playground.create_workspace(name="private"),
            "contributed": self.playground.create_workspace(name="contributed"),
            "owned": self.playground.create_workspace(name="owned"),
        }
        groups = [
            self.playground.create_group_role(
                workspaces["contributed"],
                Workspace.Roles.CONTRIBUTOR,
                users=[],
            ),
            self.playground.create_group_role(
                workspaces["owned"],
                Workspace.Roles.OWNER,
                users=[],
            ),
        ]

        pcs: dict[str, PermissionContext] = {}
        for ws_name, ws in workspaces.items():
            if not (orig_public := ws.public):
                ws.public = True
                ws.save()
            pcs[ws_name] = self.playground.create_pc_with_extra_groups(
                *groups, workspace=ws, user=self.scenario.user
            )
            ws.public = orig_public
            ws.save()

        expected: dict[str, list[WorkRequestRoles]] = {
            "public": [
                WorkRequest.Roles.CREATOR,
                WorkRequest.Roles.VIEWER,
            ],
            "private": [WorkRequest.Roles.CREATOR],
            "contributed": [
                WorkRequest.Roles.CREATOR,
                WorkRequest.Roles.CONTRIBUTOR,
            ],
            "owned": [WorkRequest.Roles.OWNER],
        }

        base_queryset = WorkRequest.objects.filter(
            pk__in=[
                pc.work_request.pk  # type: ignore[union-attr]
                for pc in pcs.values()
            ]
        ).select_related("workspace", "created_by")
        for name, with_annotations, num_queries in (
            (
                "extra_groups",
                False,
                {"public": 2, "private": 3, "contributed": 3, "owned": 3},
            ),
            (
                "annotated_extra_groups",
                True,
                {"public": 0, "private": 0, "contributed": 0, "owned": 0},
            ),
        ):
            with self.subTest(queryset=name):
                actual: dict[str, frozenset[WorkRequestRoles]] = {}
                for wr_name, pc in pcs.items():
                    wr = pc.work_request
                    assert wr
                    if with_annotations:
                        wr = base_queryset.with_role_annotations(pc).get(
                            pk=wr.pk
                        )
                    with (
                        self.subTest(work_request=wr_name),
                        self.assertNumQueries(num_queries[wr_name]),
                    ):
                        actual[wr_name] = wr.get_roles(pc)

                self.assertEqual(
                    actual, {k: frozenset(v) for k, v in expected.items()}
                )

    def test_can_display_public(self) -> None:
        """Test the can_display predicate delegating to containing workspace."""
        wr = self.playground.create_worker_task()
        predicate = wr.can_display
        self.assertPermissionPredicate(
            predicate, PermissionContext.for_ui_anonymous(), True
        )
        self.assertPermissionPredicate(
            predicate, self.scenario.context_ui(), True
        )
        with self.scenario.assign_role(
            self.scenario.workspace, Workspace.Roles.CONTRIBUTOR
        ):
            self.assertPermissionPredicate(
                predicate, self.scenario.context_ui(), True
            )
        with self.scenario.assign_role(
            self.scenario.workspace, Workspace.Roles.OWNER
        ):
            self.assertPermissionPredicate(
                predicate, self.scenario.context_ui(), True
            )

    def test_can_display_private(self) -> None:
        """Test the can_display predicate delegating to containing workspace."""
        other = self.playground.create_user("other")
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        wr = self.playground.create_worker_task()
        predicate = wr.can_display
        self.assertPermissionPredicate(
            predicate, PermissionContext.for_ui_anonymous(), False
        )
        for creator, roles, expected in (
            (True, [], False),
            (True, [Workspace.Roles.CONTRIBUTOR], True),
            (True, [Workspace.Roles.OWNER], True),
            (False, [], False),
            (False, [Workspace.Roles.CONTRIBUTOR], True),
            (False, [Workspace.Roles.OWNER], True),
        ):
            with (
                self.subTest(creator=creator, roles=repr(roles)),
                self.scenario.assign_role(self.scenario.workspace, *roles),
            ):
                if creator:
                    wr.created_by = self.scenario.user
                else:
                    wr.created_by = other
                wr.save()
                self.assertPermissionPredicate(
                    predicate,
                    self.scenario.context_ui(),
                    expected,
                )

    def test_can_display_other_work_request(self) -> None:
        wr1 = self.playground.create_worker_task()
        wr2 = self.playground.create_worker_task()

        pc = self.scenario.context_ui()
        with self.subTest(name="check user"):
            self.assertTrue(wr1.can_display(pc))
            self.assertTrue(wr2.can_display(pc))
        with self.subTest(name="filter user"):
            self.assertQuerySetEqual(
                WorkRequest.objects.can_display(pc),
                [wr1, wr2, self.workflow],
                ordered=False,
            )

        pc_wr1 = PermissionContext.for_work_request(work_request=wr1)
        with self.subTest(name="check wr"):
            self.assertTrue(wr1.can_display(pc_wr1))
            self.assertFalse(wr2.can_display(pc_wr1))

        with self.subTest(name="filter wr"):
            self.assertQuerySetEqual(
                WorkRequest.objects.can_display(pc_wr1), [wr1]
            )

        pc_wr1_pcd = pc_wr1.with_permission_checks_disabled()
        with self.subTest(name="pcd check wr"):
            self.assertTrue(wr1.can_display(pc_wr1_pcd))
            self.assertTrue(wr2.can_display(pc_wr1_pcd))

        with self.subTest(name="pcd filter wr"):
            self.assertQuerySetEqual(
                WorkRequest.objects.can_display(pc_wr1_pcd),
                [wr1, wr2, self.workflow],
                ordered=False,
            )

    def test_can_retry(self) -> None:
        """Test the can_retry predicate delegating to containing workspace."""
        user = self.scenario.user
        wr = self.playground.create_worker_task()
        for u, roles, expected in (
            (AnonymousUser(), [], False),
            (user, [], False),
            (user, [Workspace.Roles.VIEWER], False),
            (user, [Workspace.Roles.CONTRIBUTOR], True),
            (user, [Workspace.Roles.OWNER], True),
        ):
            with (
                self.subTest(user=repr(u), roles=repr(roles)),
                self.scenario.assign_role(self.scenario.workspace, *roles),
            ):
                if u.is_authenticated:
                    pc = PermissionContext.for_ui_user(user)
                else:
                    pc = PermissionContext.for_ui_anonymous()
                self.assertPermissionPredicate(wr.can_retry, pc, expected)

    def test_can_abort(self) -> None:
        """Test the can_abort predicate."""
        scope_owner = self.playground.create_user("scope_owner")
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[scope_owner]
        )

        workspace_owner = self.playground.create_user("workspace_owner")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.OWNER,
            users=[workspace_owner],
        )

        # Anonymous and unrelated users are not allowed.
        self.assertPermission(
            "can_abort",
            users=(AnonymousUser(), self.playground.create_user("unrelated")),
            denied=self.workflow,
        )
        # Workers are not allowed.
        self.assertPermission(
            "can_abort",
            users=None,
            token=self.playground.create_worker_token(),
            denied=self.workflow,
        )
        # Work request creator has access.
        self.assertPermission(
            "can_abort",
            users=self.workflow.created_by,
            allowed=[self.workflow],
        )
        # Scope owner has access.
        self.assertPermission(
            "can_abort",
            users=scope_owner,
            allowed=[self.workflow],
        )
        # Workspace owner has access.
        self.assertPermission(
            "can_abort",
            users=workspace_owner,
            allowed=[self.workflow],
        )

    def test_can_unblock(self) -> None:
        """Test the can_unblock predicate."""
        scope_owner = self.playground.create_user("scope_owner")
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[scope_owner]
        )

        workspace_owner = self.playground.create_user("workspace_owner")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.OWNER,
            users=[workspace_owner],
        )

        # Anonymous and not-owners not allowed
        self.assertPermission(
            "can_unblock",
            users=(AnonymousUser(), self.scenario.user),
            denied=self.workflow,
        )
        # Workers not allowed
        self.assertPermission(
            "can_unblock",
            users=None,
            token=self.playground.create_worker_token(),
            denied=self.workflow,
        )
        # Scope owner has access
        self.assertPermission(
            "can_unblock",
            users=scope_owner,
            allowed=[self.workflow],
        )
        # Workspace owner has access
        self.assertPermission(
            "can_unblock",
            users=workspace_owner,
            allowed=[self.workflow],
        )

    def test_can_confirm(self) -> None:
        """Test the can_confirm predicate."""
        scope_owner = self.playground.create_user("scope_owner")
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[scope_owner]
        )

        workspace_owner = self.playground.create_user("workspace_owner")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.OWNER,
            users=[workspace_owner],
        )

        # Anonymous and unrelated users are not allowed.
        self.assertPermission(
            "can_confirm",
            users=(AnonymousUser(), self.playground.create_user("unrelated")),
            denied=self.workflow,
        )
        # Workers are not allowed.
        self.assertPermission(
            "can_confirm",
            users=None,
            token=self.playground.create_worker_token(),
            denied=self.workflow,
        )
        # Work request creator has access.
        self.assertPermission(
            "can_confirm",
            users=self.workflow.created_by,
            allowed=[self.workflow],
        )
        # Scope owner has access.
        self.assertPermission(
            "can_confirm",
            users=scope_owner,
            allowed=[self.workflow],
        )
        # Workspace owner has access.
        self.assertPermission(
            "can_confirm",
            users=workspace_owner,
            allowed=[self.workflow],
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_aborted_propagates_last_activity_at(self) -> None:
        """Test mark_aborted() propagates last_activity_at to workflow."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.workflow.create_child_worker("noop")

            wr.mark_aborted()

        self.workflow.refresh_from_db()

        self.assertEqual(
            self.workflow.workflow_last_activity_at, wr.completed_at
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_pending_propagates_last_activity_at(self) -> None:
        """Test mark_pending() propagates last_activity_at to workflow."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                parent=self.workflow, task_name="noop"
            )

            wr.mark_pending()

        self.workflow.refresh_from_db()

        self.assertEqual(self.workflow.workflow_last_activity_at, wr.pending_at)

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_running_propagates_last_activity_at(self) -> None:
        """Test mark_running() propagates last_activity_at to workflow."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                assign_new_worker=True,
                parent=self.workflow,
                task_name="noop",
                mark_pending=True,
            )

            wr.mark_running()

        self.workflow.refresh_from_db()

        self.assertEqual(self.workflow.workflow_last_activity_at, wr.started_at)

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_completed_propagates_last_activity_at(self) -> None:
        """Test mark_completed() propagates last_activity_at to workflow."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                mark_running=True,
                assign_new_worker=True,
                parent=self.workflow,
                task_name="noop",
            )

            wr.mark_completed(WorkRequest.Results.SUCCESS)

        self.workflow.refresh_from_db()

        self.assertEqual(
            self.workflow.workflow_last_activity_at, wr.completed_at
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_aborted_updates_runtime_status(self) -> None:
        """Test mark_aborted() updates workflow's runtime status."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                mark_running=True,
                assign_new_worker=True,
                parent=self.workflow,
                task_name="noop",
            )

            wr.mark_aborted()

        self.workflow.refresh_from_db()

        self.assertEqual(
            self.workflow.workflow_runtime_status,
            WorkRequest.RuntimeStatuses.ABORTED,
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_completed_updates_runtime_status(self) -> None:
        """Test mark_completed() updates workflow's runtime status."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                mark_running=True,
                assign_new_worker=True,
                parent=self.workflow,
                task_name="noop",
            )

            wr.mark_completed(WorkRequest.Results.SUCCESS)

        self.workflow.refresh_from_db()

        self.assertEqual(
            self.workflow.workflow_runtime_status,
            WorkRequest.RuntimeStatuses.COMPLETED,
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_pending_updates_runtime_status(self) -> None:
        """Test mark_pending() updates workflow's runtime status."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                mark_running=True,
                assign_new_worker=True,
                parent=self.workflow,
                task_name="noop",
            )
            wr.status = WorkRequest.Statuses.BLOCKED
            wr.save()

            wr.mark_pending()

        self.workflow.refresh_from_db()

        self.assertEqual(
            self.workflow.workflow_runtime_status,
            WorkRequest.RuntimeStatuses.PENDING,
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True, PGTRANSACTION_RETRY=0)
    def test_mark_running_updates_runtime_status(self) -> None:
        """Test mark_running() updates workflow's runtime status."""
        with (
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr = self.playground.create_worker_task(
                assign_new_worker=True,
                parent=self.workflow,
                task_name="noop",
                mark_pending=True,
            )

            wr.mark_running()

        self.workflow.refresh_from_db()

        self.assertEqual(
            self.workflow.workflow_runtime_status,
            WorkRequest.RuntimeStatuses.RUNNING,
        )

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_mark_pending_runs_scheduler(self) -> None:
        """mark_pending() asks the scheduler to run."""
        with transaction.atomic():
            wr = self.playground.create_worker_task()

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr.mark_pending()

        mock_schedule.assert_called_once()

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_assign_worker_does_not_run_scheduler(self) -> None:
        """assign_worker() does not ask the scheduler to run."""
        with transaction.atomic():
            wr = self.playground.create_worker_task(mark_pending=True)
            worker = self.playground.create_worker()

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr.assign_worker(worker)

        mock_schedule.assert_not_called()

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_mark_running_does_not_run_scheduler(self) -> None:
        """mark_running() does not ask the scheduler to run."""
        with transaction.atomic():
            wr = self.playground.create_worker_task(assign_new_worker=True)

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr.mark_running()

        mock_schedule.assert_not_called()

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_mark_completed_runs_scheduler(self) -> None:
        """mark_completed() asks the scheduler to run."""
        with transaction.atomic():
            wr = self.playground.create_worker_task(mark_pending=True)

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr.mark_completed(WorkRequest.Results.SUCCESS)

        mock_schedule.assert_called_once()

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True)
    def test_mark_aborted_runs_scheduler(self) -> None:
        """mark_aborted() asks the scheduler to run."""
        with transaction.atomic():
            wr = self.playground.create_worker_task()

        with (
            mock.patch("debusine.server.scheduler.schedule") as mock_schedule,
            self.captureOnCommitCallbacks(execute=True),
        ):
            wr.mark_aborted()

        mock_schedule.assert_called_once()

    def test_mark_completed_deletes_token(self) -> None:
        """Test mark_completed() deletes the workflow's temporary token."""
        wr = self.playground.create_worker_task(
            mark_running=True,
            assign_new_worker=True,
            parent=self.workflow,
            task_name="noop",
        )
        self.playground.create_work_request_token(work_request=wr)

        wr.mark_completed(WorkRequest.Results.SUCCESS)

        self.assertFalse(Token.objects.filter(work_request=wr).exists())

    def test_mark_aborted_deletes_token(self) -> None:
        """Test mark_aborted() deletes the workflow's temporary token."""
        wr = self.playground.create_worker_task(
            mark_running=True,
            assign_new_worker=True,
            parent=self.workflow,
            task_name="noop",
        )
        self.playground.create_work_request_token(work_request=wr)

        wr.mark_aborted()

        self.assertFalse(Token.objects.filter(work_request=wr).exists())

    @preserve_db_task_registry()
    def test_compute_workflow_runtime_status_needs_input(self) -> None:
        """Test compute_workflow_runtime_status returns NEEDS_INPUT."""

        class WaitNoop(
            SampleBaseWaitTask[BaseTaskData, BaseDynamicTaskData],
        ):
            TASK_VERSION = 1

            @override
            def _execute(self) -> WorkRequest.Results:
                raise NotImplementedError()

        wr = self.workflow.create_child_wait(
            "waitnoop",
            workflow_data=WorkRequestWorkflowData(needs_input=True),
        )
        self.playground.advance_work_request(wr, mark_running=True)

        self.workflow.create_child_worker("noop").mark_aborted()

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.NEEDS_INPUT,
        )

    def test_compute_workflow_runtime_status_running(self) -> None:
        """Test compute_workflow_runtime_status returns RUNNING."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_running=True)

        self.workflow.create_child_worker("noop").mark_aborted()

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.RUNNING,
        )

    @preserve_db_task_registry()
    def test_compute_workflow_runtime_status_waiting(self) -> None:
        """Test compute_workflow_runtime_status returns WAITING."""

        class WaitNoop(
            SampleBaseWaitTask[BaseTaskData, BaseDynamicTaskData],
        ):
            TASK_VERSION = 1

            @override
            def _execute(self) -> WorkRequest.Results:
                raise NotImplementedError()

        wr = self.workflow.create_child_wait(
            "waitnoop",
            workflow_data=WorkRequestWorkflowData(needs_input=False),
        )
        self.playground.advance_work_request(wr, mark_running=True)

        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_aborted=True)

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.WAITING,
        )

    def test_compute_workflow_runtime_status_pending(self) -> None:
        """Test compute_workflow_runtime_status returns PENDING."""
        wr = self.workflow.create_child_worker("noop")
        wr.mark_pending()

        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_running=True)
        wr.mark_aborted()

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.PENDING,
        )

    def test_compute_workflow_runtime_status_aborted(self) -> None:
        """Test compute_workflow_runtime_status returns ABORTED."""
        self.workflow.create_child_worker("noop").mark_aborted()

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.ABORTED,
        )

    def test_compute_workflow_runtime_status_completed(self) -> None:
        """Test compute_workflow_runtime_status returns COMPLETED."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.COMPLETED,
        )

    def test_compute_workflow_runtime_status_blocked(self) -> None:
        """Test compute_workflow_runtime_status returns BLOCKED."""
        self.workflow.create_child_worker("noop")
        self.workflow.create_child_worker("noop").mark_aborted()

        self.assertEqual(
            compute_workflow_runtime_status(self.workflow),
            WorkRequest.RuntimeStatuses.BLOCKED,
        )

    def test_workflow_display_name_parameters_template_name(self) -> None:
        """Test workflow_display_name_parameters returns template name."""
        self.assertEqual(
            self.workflow.workflow_display_name_parameters,
            self.template.name,
        )

    def test_workflow_display_name_parameters_name_parameters(self) -> None:
        """Test workflow_display_name_parameters returns name and parameters."""
        self.workflow.dynamic_task_data = {"parameter_summary": "a,b"}
        self.assertEqual(
            self.workflow.workflow_display_name_parameters,
            f"{self.template.name}(a,b)",
        )

    def test_workflow_display_name_parameters_none(self) -> None:
        """Test workflow_display_name_parameters returns None."""
        self.workflow.workflow_data = WorkRequestWorkflowData()

        self.assertIsNone(self.workflow.workflow_display_name_parameters)

    def test_set_current(self) -> None:
        """Test WorkRequest.set_current."""
        scope = self.playground.get_or_create_scope(name="test")
        user = self.playground.create_user(username="testuser")
        workspace = self.playground.create_workspace(
            scope=scope, name="test", public=True
        )
        work_request = self.playground.create_worker_task(
            workspace=workspace, created_by=user
        )
        work_request.set_current()
        self.assertEqual(context.scope, scope)
        self.assertEqual(context.user, user)
        self.assertEqual(context.workspace, workspace)

    def test_compute_extra_groups_no_workflow(self) -> None:
        wr = self.playground.create_worker_task()
        self.assertQuerySetEqual(wr.get_extra_groups(), [])

    def test_compute_extra_groups_no_workflow_template(self) -> None:
        wr = self.playground.create_workflow()
        wr.workflow_template = None
        wr.save()
        self.assertQuerySetEqual(wr.get_extra_groups(), [])

    def test_compute_extra_groups_no_extra_groups(self) -> None:
        wt = self.playground.create_workflow_template("noop", "noop")
        wr = self.playground.create_workflow(wt)
        self.assertIsNotNone(wr.workflow_template)
        self.assertQuerySetEqual(wr.get_extra_groups(), [])

    def test_compute_extra_groups_two_extra_groups(self) -> None:
        group1 = self.playground.create_group("group1")
        group2 = self.playground.create_group("group2")
        wt = self.playground.create_workflow_template("noop", "noop")
        wr = self.playground.create_workflow(wt)
        assert wr.workflow_template is not None
        wt.extra_groups.set([group1, group2])
        self.assertQuerySetEqual(
            wr.get_extra_groups(), [group1, group2], ordered=False
        )

    def test_compute_extra_groups_nonrecursive(self) -> None:
        group1 = self.playground.create_group("group1")
        group2 = self.playground.create_group("group2")
        group3 = self.playground.create_group("group3")

        tpl_root = self.playground.create_workflow_template("root", "noop")
        tpl_root.extra_groups.set([group1, group2])
        tpl_nested = self.playground.create_workflow_template("nested", "noop")
        tpl_nested.extra_groups.set([group2, group3])

        # A work request
        wr = self.playground.create_worker_task()
        # The root workflow
        wf_root = self.playground.create_workflow(tpl_root)
        # A nested workflow
        wf_nested = self.playground.create_workflow(tpl_nested)

        wf_nested.parent = wf_root
        wf_nested.workflow_root = wf_root
        wf_nested.save()

        # Its containing workflow
        wr.parent = wf_nested
        wr.workflow_root = wf_root
        wr.save()

        self.assertQuerySetEqual(
            wf_root.get_extra_groups(), [group1, group2], ordered=False
        )
        self.assertQuerySetEqual(
            wf_nested.get_extra_groups(), [group1, group2], ordered=False
        )
        self.assertQuerySetEqual(
            wr.get_extra_groups(), [group1, group2], ordered=False
        )

    def test_compute_dynamic_data(self) -> None:
        work_request = self.playground.create_worker_task(
            task_name="noop", mark_pending=True
        )
        self.assertIsNone(work_request.dynamic_task_data)
        with mock.patch(
            "debusine.db.tasks.DBProxyTask.compute_dynamic_data",
            return_value=BaseDynamicTaskData(subject="test"),
        ):
            work_request.compute_dynamic_data()
        assert work_request.dynamic_task_data is not None
        self.assertEqual(work_request.dynamic_task_data["subject"], "test")

    def test_configuring_task_introduces_errors(self) -> None:
        work_request = self.playground.create_worker_task(
            task_name="noop", mark_pending=True
        )
        self.assertIsNone(work_request.dynamic_task_data)
        work_request.configured_task_data = {"invalid": True}
        work_request.save()

        with self.assertRaisesRegex(
            TaskConfigError, "Configured task data is invalid"
        ):
            work_request.compute_dynamic_data()

    def test_mark_pending_applies_task_configuration(self) -> None:
        work_request = self.playground.create_worker_task(task_name="noop")
        with mock.patch(
            "debusine.db.tasks.DBProxyTask.apply_task_configuration"
        ) as apply_task_configuration:
            work_request.mark_pending()
        apply_task_configuration.assert_called_once_with()

    def test_used_task_data(self) -> None:
        """Ensure used_task_data picks the right data."""
        work_request = self.playground.create_worker_task(
            task_name="noop", mark_pending=True
        )

        work_request.task_data = {}
        work_request.configured_task_data = None
        self.assertEqual(work_request.used_task_data, {})

        work_request.task_data = {"test": 1}
        work_request.configured_task_data = None
        self.assertEqual(work_request.used_task_data, {"test": 1})

        work_request.task_data = {"test": 1}
        work_request.configured_task_data = {}
        self.assertEqual(work_request.used_task_data, {})

        work_request.task_data = {"test": 1}
        work_request.configured_task_data = {"test": 2}
        self.assertEqual(work_request.used_task_data, {"test": 2})

    @override_settings(DEBUSINE_FALLBACK_ARCHITECTURE="risc64")
    def test_get_fallback_architecture(self) -> None:
        work_request = self.playground.create_worker_task(
            task_name="noop", mark_pending=True
        )
        self.assertEqual(work_request.get_fallback_architecture(), "risc64")

    @override_settings(DEBUSINE_FALLBACK_ARCHITECTURE="risc64")
    def test_get_fallback_architecture_defaults_amd64(self) -> None:
        work_request = self.playground.create_worker_task(
            task_name="noop", mark_pending=True
        )
        del settings.DEBUSINE_FALLBACK_ARCHITECTURE
        self.assertEqual(work_request.get_fallback_architecture(), "amd64")


class WorkRequestTransactionTests(TransactionTestCase):
    """Test transactional behaviour of the WorkRequest class."""

    scenario = scenarios.DefaultContext(set_current=True)

    def test_supersedes_expired(self) -> None:
        """Test expiring a superseded work request."""
        wr = self.playground.create_worker_task(task_name="noop")
        wr.created_at = timezone.now() - dt.timedelta(days=365)
        wr.expiration_delay = dt.timedelta(days=7)
        wr.save()
        wr1 = self.playground.create_worker_task(task_name="noop")
        wr1.supersedes = wr
        wr1.save()

        # Expire the superseded work request
        with (
            io.StringIO() as stdout,
            io.StringIO() as stderr,
            DeleteOperation(out=stdout, err=stderr, dry_run=False) as operation,
            mock.patch("pgtransaction.Atomic.execute_set_transaction_modes"),
        ):
            delete_expired = DeleteExpiredWorkRequests(operation)
            delete_expired.run()

        self.assertQuerySetEqual(WorkRequest.objects.filter(pk=wr.pk), [])
        wr1.refresh_from_db()

        # Supersedes is set to None
        self.assertIsNone(wr1.supersedes)

    @override_permission(WorkRequest, "can_retry", AllowAll)
    def test_retry_internal(self) -> None:
        """Internal work requests can be retried."""
        workflow = self.playground.create_workflow()

        wr_aborted = WorkRequest.objects.create_synchronization_point(
            parent=workflow, step="test"
        )
        wr_aborted.status = WorkRequest.Statuses.ABORTED
        wr_aborted.save()

        self.assertTrue(wr_aborted.verify_retry())
        wr_aborted.retry()

        update_workflows()

        workflow.refresh_from_db()
        self.assertIsNotNone(workflow.workflow_runtime_status)

    def test_schedule_workflow_update(self) -> None:
        """Test schedule_workflow_update()."""
        workflow = self.playground.create_workflow()
        sub_workflow = self.playground.create_workflow(
            parent=workflow, task_name="noop", mark_pending=False
        )
        # FIXME: the test fails if we call mark_pending
        sub_workflow.status = WorkRequest.Statuses.PENDING
        sub_workflow.save()

        self.assertIsNone(sub_workflow.workflow_runtime_status)
        self.assertIsNone(workflow.workflow_runtime_status)

        sub_workflow.schedule_workflow_update()
        self.assertTrue(sub_workflow.workflows_need_update)

        update_workflows()

        sub_workflow.refresh_from_db()
        workflow.refresh_from_db()

        self.assertEqual(
            sub_workflow.workflow_runtime_status,
            WorkRequest.RuntimeStatuses.COMPLETED,
        )
        self.assertEqual(
            workflow.workflow_runtime_status,
            WorkRequest.RuntimeStatuses.COMPLETED,
        )
        self.assertFalse(sub_workflow.workflows_need_update)


class WorkRequestWorkerTests(ChannelsHelpersMixin, TestCase):
    """Tests for the WorkRequest class scheduling."""

    worker: ClassVar[Worker]
    work_request: ClassVar[WorkRequest]
    workflow: ClassVar[WorkRequest]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()

        cls.worker = cls.playground.create_worker()
        cls.work_request = cls.playground.create_worker_task(
            task_name="noop", mark_pending=True
        )
        cls.workflow = cls.playground.create_workflow()
        cls.playground.advance_work_request(cls.workflow, mark_running=True)

    def test_str(self) -> None:
        """Test WorkerRequest.__str__ return WorkRequest.id."""
        self.assertEqual(self.work_request.__str__(), str(self.work_request.id))

    def test_mark_pending_from_aborted(self) -> None:
        """Test WorkRequest.mark_pending() doesn't change (was aborted)."""
        self.work_request.assign_worker(self.worker)
        self.work_request.status = WorkRequest.Statuses.ABORTED
        self.work_request.save()

        with self.assertRaisesRegex(
            StatusChangeError,
            f"Cannot mark WorkRequest {self.work_request.pk} as pending: "
            f"current status is aborted",
        ):
            self.work_request.mark_pending()

        self.work_request.refresh_from_db()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.ABORTED)

    def test_mark_pending(self) -> None:
        work_request = self.playground.create_worker_task()
        self.assertIsNone(work_request.pending_at)
        self.assertIsNone(work_request.configured_task_data)

        work_request.mark_pending()

        work_request.refresh_from_db()
        self.assertEqual(work_request.status, WorkRequest.Statuses.PENDING)
        assert work_request.pending_at is not None
        self.assertLess(work_request.pending_at, timezone.now())
        self.assertIsNotNone(work_request.configured_task_data)

        with self.assertRaisesRegex(
            StatusChangeError,
            f"Cannot mark WorkRequest {work_request.pk} as pending:"
            f" current status is pending",
        ):
            work_request.mark_pending()

    def test_mark_running_from_aborted(self) -> None:
        """Test WorkRequest.mark_running() doesn't change (was aborted)."""
        self.work_request.assign_worker(self.worker)
        self.work_request.status = WorkRequest.Statuses.ABORTED
        self.work_request.save()

        with self.assertRaisesRegex(
            StatusChangeError,
            r"Cannot mark WorkRequest \d+ as running:"
            " current status is aborted",
        ):
            self.work_request.mark_running()

        self.work_request.refresh_from_db()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.ABORTED)

    def test_mark_running(self) -> None:
        """Test WorkRequest.mark_running() change status to running."""
        self.work_request.status = WorkRequest.Statuses.PENDING
        self.work_request.save()
        self.assertIsNone(self.work_request.started_at)

        self.work_request.assign_worker(self.worker)
        self.work_request.mark_running()

        self.work_request.refresh_from_db()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.RUNNING)
        assert self.work_request.started_at is not None
        self.assertLess(self.work_request.started_at, timezone.now())

        # Marking as running again (a running WorkRequest) is a no-op
        started_at = self.work_request.started_at
        self.work_request.mark_running()

        self.work_request.refresh_from_db()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(self.work_request.started_at, started_at)

    def test_mark_running_emits_metric(self) -> None:
        registry = save_registry()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.PENDING)
        self.work_request.pending_at = timezone.now() - dt.timedelta(hours=1)
        self.work_request.save()

        self.work_request.assign_worker(self.worker)
        self.work_request.mark_running()

        self.work_request.refresh_from_db()
        labels = {
            "task_type": self.work_request.task_type,
            "task_name": self.work_request.task_name,
            "priority": self.work_request.priority_for_metrics,
        }
        assert_metric_diff(
            registry,
            0,
            "debusine_work_request_start_latency_bucket",
            **labels,
            le="3600.0",
        )
        assert_metric_diff(
            registry,
            1,
            "debusine_work_request_start_latency_bucket",
            **labels,
            le="10800.0",
        )
        assert_metric_diff(
            registry,
            1,
            "debusine_work_request_start_latency_bucket",
            **labels,
            le="+Inf",
        )

    def test_mark_running_no_pending_at_emits_no_metric(self) -> None:
        """If ``pending_at`` is unset, ``mark_running`` emits no metric."""
        start_latency.labels(
            task_type=str(self.work_request.task_type),
            task_name=self.work_request.task_name,
            priority=self.work_request.priority_for_metrics,
        )
        registry = save_registry()

        self.assertEqual(self.work_request.status, WorkRequest.Statuses.PENDING)
        self.work_request.pending_at = None
        self.work_request.save()

        self.work_request.assign_worker(self.worker)
        self.work_request.mark_running()

        self.work_request.refresh_from_db()
        labels = {
            "task_type": self.work_request.task_type,
            "task_name": self.work_request.task_name,
            "priority": self.work_request.priority_for_metrics,
        }
        assert_metric_diff(
            registry,
            0,
            "debusine_work_request_start_latency_bucket",
            **labels,
            le="+Inf",
        )

    def test_mark_running_fails_worker_already_running(self) -> None:
        """WorkRequest.mark_running() return False: worker already running."""
        self.work_request.assign_worker(self.worker)
        self.work_request.status = WorkRequest.Statuses.PENDING
        self.work_request.save()

        other_work_request = self.playground.create_worker_task(
            mark_running=True, worker=self.worker
        )

        with self.assertRaisesRegex(
            StatusChangeError,
            rf"Cannot mark WorkRequest {self.work_request.pk} as running: the"
            rf" assigned worker {re.escape(str(self.work_request.worker))}"
            r" is running too many WorkRequests:"
            rf" \[{re.escape(repr(other_work_request))}\]",
        ):
            self.work_request.mark_running()

    def test_mark_running_fails_no_assigned_worker(self) -> None:
        """WorkRequest.mark_running() return False: no worker assigned."""
        self.work_request.status = WorkRequest.Statuses.PENDING
        self.work_request.worker = None
        self.work_request.save()
        with self.assertRaisesRegex(
            StatusChangeError,
            r"Cannot mark WorkRequest \d+ as running:"
            " it does not have an assigned worker",
        ):
            self.work_request.mark_running()

    def test_mark_running_fails_unconfigured_with_worker(self) -> None:
        self.work_request.configured_task_data = None
        self.work_request.worker = self.worker
        with self.assertRaisesRegex(
            StatusChangeError, r"Unconfigured Worker task marked running"
        ):
            self.work_request.mark_running()

    def test_mark_completed_from_aborted(self) -> None:
        """Test WorkRequest.mark_completed() doesn't change (was aborted)."""
        self.work_request.assign_worker(self.worker)
        self.work_request.status = WorkRequest.Statuses.ABORTED
        self.work_request.save()

        self.work_request.refresh_from_db()

        with self.assertRaisesRegex(
            StatusChangeError,
            r"Cannot mark WorkRequest \d+ as completed:"
            r" current status is aborted",
        ):
            self.work_request.mark_completed(WorkRequest.Results.SUCCESS)

        self.work_request.refresh_from_db()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.ABORTED)

    def test_mark_completed(self) -> None:
        """Test WorkRequest.mark_completed() changes status to completed."""
        self.work_request.assign_worker(self.worker)
        self.work_request.status = WorkRequest.Statuses.RUNNING
        self.work_request.save()

        self.assertIsNone(self.work_request.completed_at)
        self.assertEqual(self.work_request.result, WorkRequest.Results.NONE)

        self.work_request.mark_completed(WorkRequest.Results.SUCCESS)

        self.work_request.refresh_from_db()
        self.assertEqual(
            self.work_request.status, WorkRequest.Statuses.COMPLETED
        )
        self.assertEqual(self.work_request.result, WorkRequest.Results.SUCCESS)
        assert self.work_request.completed_at is not None
        self.assertLess(self.work_request.completed_at, timezone.now())

    def test_mark_completed_no_result_none(self) -> None:
        self.playground.advance_work_request(
            self.work_request, mark_running=True
        )
        with self.assertRaisesRegex(
            StatusChangeError, r"result cannot be NONE"
        ):
            self.work_request.mark_completed(WorkRequest.Results.NONE)

    def test_mark_completed_records_worker_pool_statistics(self) -> None:
        """mark_completed() records worker pool task execution statistics."""
        worker_pool = self.playground.create_worker_pool()
        worker = self.playground.create_worker(worker_pool=worker_pool)
        self.work_request.assign_worker(worker)
        self.work_request.mark_running()
        output_data = OutputData(
            runtime_statistics=RuntimeStatistics(duration=10)
        )

        self.work_request.mark_completed(
            WorkRequest.Results.SUCCESS, output_data=output_data
        )

        self.assertQuerySetEqual(
            WorkerPoolTaskExecutionStatistics.objects.filter(
                worker_pool=worker_pool
            ).values_list("worker_pool", "worker", "runtime"),
            [(worker_pool.id, worker.id, 10)],
        )

    def test_mark_completed_merges_output_data(self) -> None:
        """mark_completed() updates any existing output data."""
        # This particular set of data wouldn't happen (we won't record
        # errors and then also record runtime statistics), but it's enough
        # to test that we merge multiple output data structures.
        self.work_request.output_data = OutputData(
            runtime_statistics=RuntimeStatistics(duration=1),
            errors=[OutputDataError(message="message", code="code")],
        )

        self.work_request.mark_completed(
            WorkRequest.Results.FAILURE,
            output_data=OutputData(
                runtime_statistics=RuntimeStatistics(duration=2, cpu_time=1)
            ),
        )

        self.assertEqual(
            self.work_request.output_data,
            OutputData(
                runtime_statistics=RuntimeStatistics(duration=2, cpu_time=1),
                errors=[OutputDataError(message="message", code="code")],
            ),
        )

    def test_mark_aborted(self) -> None:
        """Test WorkRequest.mark_aborted() changes status to aborted."""
        self.assertIsNone(self.work_request.completed_at)

        with self.assertLogsContains(
            f"Marked WorkRequest {self.work_request.pk} as aborted"
            f" (from status pending)",
            logger="debusine.db.models.work_requests",
            level=logging.DEBUG,
        ):
            self.work_request.mark_aborted()

        self.work_request.refresh_from_db()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.ABORTED)
        assert self.work_request.completed_at is not None
        self.assertLess(self.work_request.completed_at, timezone.now())

    def test_mark_aborted_sets_output_data(self) -> None:
        """mark_aborted() can set output data if there is none."""
        self.work_request.mark_aborted(
            output_data=OutputData(confirmation=Confirmation(confirmed=False)),
        )

        self.assertEqual(
            self.work_request.output_data,
            OutputData(confirmation=Confirmation(confirmed=False)),
        )

    def test_mark_aborted_merges_output_data(self) -> None:
        """mark_aborted() updates any existing output data."""
        self.work_request.output_data = OutputData(
            errors=[OutputDataError(message="message", code="code")],
        )

        self.work_request.mark_aborted(
            output_data=OutputData(confirmation=Confirmation(confirmed=False)),
        )

        self.assertEqual(
            self.work_request.output_data,
            OutputData(
                errors=[OutputDataError(message="message", code="code")],
                confirmation=Confirmation(confirmed=False),
            ),
        )

    def test_maybe_finish_workflow_orchestrator_running(self) -> None:
        """Skip ``maybe_finish_workflow`` while running the orchestrator."""
        wr = self.workflow.create_child_worker("noop")
        self.workflow.mark_running()
        with mock.patch(
            "debusine.db.models.work_requests.WorkRequest.maybe_finish_workflow"
        ):
            self.playground.advance_work_request(
                wr, result=WorkRequest.Results.SUCCESS
            )

        with self.workflow.running_workflow_orchestrator():
            self.assertFalse(self.workflow.maybe_finish_workflow())

    def test_maybe_finish_workflow_not_workflow(self) -> None:
        """`maybe_finish_workflow` does nothing on non-workflows."""
        self.assertFalse(self.work_request.maybe_finish_workflow())

    def test_maybe_finish_workflow_already_aborted(self) -> None:
        """`maybe_finish_workflow` does nothing if it was already aborted."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )
        self.workflow.mark_aborted()

        self.assertFalse(self.workflow.maybe_finish_workflow())

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)

    def test_maybe_finish_workflow_has_children_in_progress(self) -> None:
        """`maybe_finish_workflow` does nothing if a child is in progress."""
        children = [self.workflow.create_child_worker("noop") for _ in range(2)]
        self.workflow.mark_running()
        self.workflow.unblock_workflow_children()
        children[0].refresh_from_db()
        children[0].mark_completed(WorkRequest.Results.SUCCESS)

        self.assertFalse(self.workflow.maybe_finish_workflow())

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.RUNNING)

    def test_maybe_finish_workflow_all_succeeded(self) -> None:
        """All children succeeded: COMPLETED/SUCCESS or COMPLETED/SKIPPED."""
        children = [self.workflow.create_child_worker("noop") for _ in range(3)]
        self.workflow.mark_running()
        self.workflow.unblock_workflow_children()

        children[0].refresh_from_db()
        children[0].mark_completed(WorkRequest.Results.SUCCESS)
        children[1].refresh_from_db()
        children[1].mark_completed(WorkRequest.Results.SUCCESS)
        children[2].refresh_from_db()
        children[2].mark_completed(WorkRequest.Results.SKIPPED)

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(self.workflow.result, WorkRequest.Results.SUCCESS)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())

    def test_maybe_finish_workflow_allow_failure(self) -> None:
        """Some children failed/aborted but are allowed to fail."""
        children = [
            self.workflow.create_child_worker(
                "noop",
                workflow_data=WorkRequestWorkflowData(
                    allow_failure=True, display_name="test", step="test"
                ),
            )
            for _ in range(4)
        ]
        self.workflow.mark_running()
        self.workflow.unblock_workflow_children()

        children[0].refresh_from_db()
        children[0].mark_completed(WorkRequest.Results.FAILURE)
        children[1].refresh_from_db()
        children[1].mark_completed(WorkRequest.Results.SUCCESS)
        children[2].refresh_from_db()
        children[2].mark_completed(WorkRequest.Results.SKIPPED)
        children[3].refresh_from_db()
        children[3].mark_aborted()

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(self.workflow.result, WorkRequest.Results.SUCCESS)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())

    def test_maybe_finish_workflow_allow_dependency_failures(self) -> None:
        """Some children failed, but the parent allows that."""
        self.workflow.workflow_data = WorkRequestWorkflowData(
            allow_dependency_failures=True
        )
        self.workflow.save()
        children = [self.workflow.create_child_worker("noop") for _ in range(4)]
        self.workflow.mark_running()
        self.workflow.unblock_workflow_children()

        children[0].refresh_from_db()
        children[0].mark_completed(WorkRequest.Results.FAILURE)
        children[1].refresh_from_db()
        children[1].mark_completed(WorkRequest.Results.SUCCESS)
        children[2].refresh_from_db()
        children[2].mark_completed(WorkRequest.Results.SKIPPED)
        children[3].refresh_from_db()
        children[3].mark_aborted()

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(self.workflow.result, WorkRequest.Results.SUCCESS)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())

    def test_maybe_finish_workflow_some_failed(self) -> None:
        """Some children failed: COMPLETED/FAILURE."""
        children = [self.workflow.create_child_worker("noop") for _ in range(2)]
        self.workflow.mark_running()
        self.workflow.unblock_workflow_children()

        children[0].refresh_from_db()
        children[0].mark_completed(WorkRequest.Results.FAILURE)
        children[1].refresh_from_db()
        children[1].mark_completed(WorkRequest.Results.SUCCESS)

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(self.workflow.result, WorkRequest.Results.FAILURE)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())

    def test_maybe_finish_workflow_some_aborted(self) -> None:
        """Some children were aborted: `maybe_finish_workflow` sets ABORTED."""
        children = [self.workflow.create_child_worker("noop") for _ in range(2)]
        self.workflow.mark_running()
        self.workflow.unblock_workflow_children()

        children[0].refresh_from_db()
        children[0].mark_aborted()
        children[1].refresh_from_db()
        children[1].mark_completed(WorkRequest.Results.SUCCESS)

        self.workflow.refresh_from_db()
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())

    def test_maybe_finish_workflow_superseded_aborted(self) -> None:
        """Some aborted children were superseded: ignores them."""
        children = [self.workflow.create_child_worker("noop") for _ in range(2)]
        self.workflow.started_at = timezone.now()
        self.workflow.status = WorkRequest.Statuses.RUNNING
        self.workflow.save()

        children[0].completed_at = timezone.now()
        children[0].status = WorkRequest.Statuses.ABORTED
        children[0].save()
        children[1].completed_at = timezone.now()
        children[1].status = WorkRequest.Statuses.COMPLETED
        children[1].result = WorkRequest.Results.SUCCESS
        children[1].save()

        self.assertTrue(self.workflow.maybe_finish_workflow())
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.ABORTED)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())
        prev_completed_at = self.workflow.completed_at

        new_wr = self.workflow.create_child_worker("noop")
        new_wr.supersedes = children[0]
        new_wr.status = WorkRequest.Statuses.COMPLETED
        new_wr.result = WorkRequest.Results.SUCCESS
        new_wr.save()
        self.workflow.status = WorkRequest.Statuses.RUNNING
        self.workflow.save()

        self.assertTrue(self.workflow.maybe_finish_workflow())
        self.assertEqual(self.workflow.status, WorkRequest.Statuses.COMPLETED)
        assert self.workflow.completed_at is not None
        self.assertLess(self.workflow.completed_at, timezone.now())
        self.assertGreater(self.workflow.completed_at, prev_completed_at)

    async def test_assign_worker(self) -> None:
        """Assign Worker to WorkRequest."""

        def setup_work_request() -> None:
            # Initial status (no worker)
            self.assertIsNone(self.work_request.worker)
            self.assertEqual(
                self.work_request.status, WorkRequest.Statuses.PENDING
            )

        await sync_to_async(setup_work_request)()
        assert self.worker.token is not None

        channel = await self.create_channel(self.worker.token.hash)

        def assign() -> None:
            # Assign the worker to the WorkRequest
            with self.captureOnCommitCallbacks(execute=True):
                self.work_request.assign_worker(self.worker)

        await sync_to_async(assign)()

        await self.assert_channel_received(
            channel, {"type": "work_request.assigned"}
        )

        def assert_final_status() -> None:
            self.work_request.refresh_from_db()

            # Assert final status
            self.assertEqual(self.work_request.worker, self.worker)
            self.assertEqual(
                self.work_request.status, WorkRequest.Statuses.PENDING
            )

        await sync_to_async(assert_final_status)()

    def test_de_assign_worker_from_running(self) -> None:
        """Worker is de-assigned from a work request (was running)."""
        worker = self.playground.create_worker()
        self.work_request.assign_worker(worker)

        self.work_request.mark_running()

        self.assertEqual(self.work_request.worker, worker)
        self.assertIsInstance(self.work_request.started_at, dt.datetime)
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.RUNNING)

        self.assertTrue(self.work_request.de_assign_worker())

        self.work_request.refresh_from_db()

        self.assertIsNone(self.work_request.worker)
        self.assertIsNone(self.work_request.started_at)
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.PENDING)

    def test_de_assign_worker_from_pending(self) -> None:
        """Worker is de-assigned from a work request (was pending)."""
        worker = self.playground.create_worker()
        self.work_request.assign_worker(worker)

        self.assertEqual(self.work_request.worker, worker)
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.PENDING)

        self.assertTrue(self.work_request.de_assign_worker())

        self.work_request.refresh_from_db()

        self.assertIsNone(self.work_request.worker)
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.PENDING)

    def test_de_assign_worker_returns_false(self) -> None:
        """Worker cannot be de-assigned: was not in pending or running."""
        self.work_request.mark_completed(WorkRequest.Results.SUCCESS)

        log_msg = (
            f"WorkRequest {self.work_request.pk} cannot be de-assigned: "
            f"current status: {self.work_request.status}"
        )

        with self.assertLogsContains(
            log_msg, logger="debusine.db.models", level=logging.WARNING
        ):
            self.assertFalse(self.work_request.de_assign_worker())

    def test_duration(self) -> None:
        """duration() returns the correct duration."""
        self.work_request.started_at = dt.datetime(
            2022, 3, 7, 10, 51, tzinfo=dt.UTC
        )
        self.work_request.completed_at = dt.datetime(
            2022, 3, 9, 12, 53, tzinfo=dt.UTC
        )
        duration = self.work_request.completed_at - self.work_request.started_at

        self.assertEqual(
            self.work_request.duration,
            duration.total_seconds(),
        )

    def test_duration_is_none_completed_at_is_none(self) -> None:
        """duration() returns None because completed_at is None."""
        self.work_request.started_at = dt.datetime(
            2022, 3, 7, 10, 51, tzinfo=dt.UTC
        )
        self.work_request.completed_at = None

        self.assertIsNone(self.work_request.duration)

    def test_duration_is_none_started_at_is_none(self) -> None:
        """duration() returns None because started_at is None."""
        self.work_request.started_at = None
        self.work_request.completed_at = dt.datetime(
            2022, 3, 7, 10, 51, tzinfo=dt.UTC
        )

        self.assertIsNone(self.work_request.duration)

    def test_priority_effective(self) -> None:
        """priority_effective() returns the effective priority."""
        self.work_request.priority_base = 10
        self.work_request.priority_adjustment = 1

        self.assertEqual(self.work_request.priority_effective, 11)

    def test_priority_for_metrics(self) -> None:
        self.work_request.priority_adjustment = 1
        for priority, label in (
            (1, "normal"),
            (0, "normal"),
            (-1, "background"),
            (-100, "background"),
            (-101, "indefinite"),
        ):
            with self.subTest(priority=priority):
                self.work_request.priority_base = (
                    priority - self.work_request.priority_adjustment
                )

                self.assertEqual(self.work_request.priority_for_metrics, label)

    def test_invalid_pending(self) -> None:
        """Reject invalid pending state change."""
        self.work_request.status = WorkRequest.Statuses.RUNNING
        with self.assertRaisesRegex(
            StatusChangeError,
            r"Cannot mark WorkRequest \d+ as pending:"
            " current status is running",
        ):
            self.work_request.mark_pending()
        self.assertEqual(self.work_request.status, WorkRequest.Statuses.RUNNING)

    def test_add_dependency_within_workflow(self) -> None:
        """Dependencies within a workflow are allowed."""
        children = [self.workflow.create_child_worker("noop") for _ in range(2)]

        children[0].add_dependency(children[1])

        self.assertQuerySetEqual(children[0].dependencies.all(), [children[1]])

    def test_add_dependency_outside_workflow(self) -> None:
        """Dependencies from outside a workflow are forbidden."""
        with self.assertRaisesRegex(
            ValueError, "Only work requests in a workflow may have dependencies"
        ):
            self.work_request.add_dependency(
                self.playground.create_worker_task()
            )

    def test_add_dependency_target_outside_workflow(self) -> None:
        """Dependencies from within a workflow to outside are forbidden."""
        child = self.workflow.create_child_worker("noop")
        outside = self.playground.create_worker_task()

        with self.assertRaisesRegex(
            ValueError,
            "Work requests in a workflow may not depend on other work "
            "requests outside that workflow",
        ):
            child.add_dependency(outside)

    def test_add_dependency_auto_block(self) -> None:
        """`add_dependency` automatically blocks PENDING/DEPS work requests."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        self.assertEqual(rdep.status, WorkRequest.Statuses.BLOCKED)

    def test_add_dependency_no_auto_block_not_pending(self) -> None:
        """`add_dependency` doesn't automatically block non-PENDING requests."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(rdep, mark_aborted=True)
        rdep.add_dependency(wr)
        self.assertEqual(rdep.status, WorkRequest.Statuses.ABORTED)

    def test_add_dependency_no_auto_block_manual(self) -> None:
        """`add_dependency` doesn't automatically block MANUAL requests."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(rdep, mark_pending=True)
        rdep.unblock_strategy = WorkRequest.UnblockStrategy.MANUAL
        rdep.save()
        rdep.add_dependency(wr)
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_add_dependency_no_auto_block_incremental(self) -> None:
        """`add_dependency` doesn't automatically block INCREMENTAL requests."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_internal("workflow")
        self.playground.advance_work_request(rdep, mark_pending=True)
        rdep.unblock_strategy = WorkRequest.UnblockStrategy.INCREMENTAL
        rdep.save()
        rdep.add_dependency(wr)
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_add_dependency_no_auto_block_already_completed(self) -> None:
        """`add_dependency` doesn't auto-block on completed dependencies."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )
        rdep = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(rdep, mark_pending=True)
        rdep.add_dependency(wr)
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_add_dependency_no_auto_block_already_aborted(self) -> None:
        """`add_dependency` doesn't auto-block on aborted dependencies."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_aborted=True)
        rdep = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(rdep, mark_pending=True)
        rdep.add_dependency(wr)
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_failure(self) -> None:
        """Failure result gets rdep aborted."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.FAILURE
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.ABORTED)

    def test_dependencies_aborted(self) -> None:
        """Aborting a work request aborts its rdeps."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        self.playground.advance_work_request(wr, mark_aborted=True)
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.ABORTED)

    def test_dependencies_success_only_dep(self) -> None:
        """Success result, with nothing else to do, gets rdep pending."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_skipped_only_dep(self) -> None:
        """Skipped result, with nothing else to do, gets rdep pending."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SKIPPED
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_failure_allow_failure(self) -> None:
        """Failure result, with allow_failure, gets rdep pending."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        wr.workflow_data = WorkRequestWorkflowData(
            allow_failure=True, display_name="test", step="test"
        )
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.FAILURE
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_aborted_allow_failure(self) -> None:
        """Aborting a work request, with allow_failure, gets rdep pending."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        wr.workflow_data = WorkRequestWorkflowData(
            allow_failure=True, display_name="test", step="test"
        )
        self.playground.advance_work_request(wr, mark_aborted=True)
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_failure_allow_dependency_failures(self) -> None:
        """Rdeps with allow_dependency_failures ignore failures."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                allow_dependency_failures=True
            ),
        )
        rdep.add_dependency(wr)
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.FAILURE
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_aborted_allow_dependency_failures(self) -> None:
        """Rdeps with allow_dependency_failures ignore aborts."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker(
            "noop",
            workflow_data=WorkRequestWorkflowData(
                allow_dependency_failures=True
            ),
        )
        rdep.add_dependency(wr)
        self.playground.advance_work_request(wr, mark_aborted=True)
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_success_manual(self) -> None:
        """Success result with manual unblock keeps rdep blocked."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        rdep.unblock_strategy = WorkRequest.UnblockStrategy.MANUAL
        rdep.save()
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.BLOCKED)

    def test_dependencies_success_incremental(self) -> None:
        """Success result with incremental unblock marks rdep pending."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_internal("workflow")
        rdep.add_dependency(wr)
        rdep.unblock_strategy = WorkRequest.UnblockStrategy.INCREMENTAL
        rdep.save()
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_dependencies_success_more_deps(self) -> None:
        """Success result with something else to do keeps rdep blocked."""
        wr = self.workflow.create_child_worker("noop")
        rdep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(wr)
        wr2 = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr2, mark_running=True)
        rdep.add_dependency(wr2)
        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.BLOCKED)

    def test_dependencies_in_pending_workflow(self) -> None:
        """Parent workflow that isn't running yet keeps rdep blocked."""
        # Don't use self.workflow, since it's running.
        workflow = self.playground.create_workflow()
        rdep = workflow.create_child_worker("noop")
        dep = workflow.create_child_worker("noop")
        rdep.add_dependency(dep)
        self.playground.advance_work_request(
            dep, result=WorkRequest.Results.SUCCESS
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.BLOCKED)

    def test_dependencies_in_running_workflow(self) -> None:
        """Parent workflow in running status gets rdep pending."""
        rdep = self.workflow.create_child_worker("noop")
        dep = self.workflow.create_child_worker("noop")
        rdep.add_dependency(dep)
        self.workflow.mark_running()
        self.playground.advance_work_request(
            dep, result=WorkRequest.Results.SUCCESS
        )
        rdep.refresh_from_db()
        self.assertEqual(rdep.status, WorkRequest.Statuses.PENDING)

    def test_add_event_reaction(self) -> None:
        """`add_event_reaction` is idempotent."""
        update_collection_with_artifacts = ActionUpdateCollectionWithArtifacts(
            collection="internal@collections",
            name_template="test",
            artifact_filters={"category": ArtifactCategory.TEST},
        )
        self.work_request.add_event_reaction(
            "on_success", update_collection_with_artifacts
        )
        assert self.work_request.event_reactions == EventReactions(
            on_success=[update_collection_with_artifacts]
        )

        self.work_request.add_event_reaction(
            "on_success", update_collection_with_artifacts
        )
        self.assertEqual(
            self.work_request.event_reactions,
            EventReactions(on_success=[update_collection_with_artifacts]),
        )

        send_notification = ActionSendNotification(channel="test")
        self.work_request.add_event_reaction("on_success", send_notification)
        self.work_request.add_event_reaction(
            "on_failure", update_collection_with_artifacts
        )
        self.assertEqual(
            self.work_request.event_reactions,
            EventReactions(
                on_success=[
                    update_collection_with_artifacts,
                    send_notification,
                ],
                on_failure=[update_collection_with_artifacts],
            ),
        )

    def test_create_child_non_workflow(self) -> None:
        """Non-WORKFLOW work requests may not have children."""
        expected_message = r"Only workflows may have child work requests\."
        for method in ("worker", "server", "internal", "workflow", "signing"):
            with (
                self.subTest(method=method),
                self.assertRaisesRegex(ValueError, expected_message),
            ):
                factory_method = getattr(
                    self.work_request, f"create_child_{method}"
                )
                factory_method("noop")
        with (
            self.subTest(method="wait"),
            self.assertRaisesRegex(ValueError, expected_message),
        ):
            self.work_request.create_child_wait(
                "noop",
                workflow_data=WorkRequestWorkflowData(needs_input=False),
            )

    def test_create_child_validates_work_request(self) -> None:
        """Creating a child validates the resulting work request."""
        with self.assertRaises(ValidationError) as raised:
            self.workflow.create_child_worker(task_name="does-not-exist")

        self.assertEqual(
            raised.exception.message_dict,
            {"task_name": ["does-not-exist: invalid Worker task name"]},
        )

    def test_create_child_inherits_priority(self) -> None:
        """A child inherits the effective priority of its parent as its base."""
        self.workflow.priority_base = 5
        self.workflow.priority_adjustment = 1
        self.workflow.save()

        wr = self.workflow.create_child_worker(task_name="noop")
        self.assertEqual(wr.priority_base, 6)

    def test_create_child_relative_priority(self) -> None:
        """A child can be given a priority relative to its parent's priority."""
        self.workflow.priority_base = 5
        self.workflow.priority_adjustment = 1
        self.workflow.save()

        wr = self.workflow.create_child_worker("noop", relative_priority=-2)
        self.assertEqual(wr.priority_base, 4)

    def test_create_child_inherits_task_configuration(self) -> None:
        """A child inherits the task_configuration of its parent by_default."""
        self.workflow.task_data["task_configuration"] = "test"
        self.workflow.save()

        wr = self.workflow.create_child_worker(task_name="noop")
        self.assertEqual(wr.task_data["task_configuration"], "test")

    def test_create_child_inherits_task_configuration_default(self) -> None:
        """A child inherits the task_configuration of its parent by_default."""
        wr = self.workflow.create_child_worker("noop")
        self.assertNotIn("task_configuration", wr.task_data)

    def test_create_child_can_override_task_configuration(self) -> None:
        self.workflow.task_data["task_configuration"] = "test"
        self.workflow.save()

        wr = self.workflow.create_child_worker(
            task_name="noop", task_data=BaseTaskData(task_configuration="other")
        )
        self.assertEqual(wr.task_data["task_configuration"], "other")

    def test_create_child_can_disable_task_configuration(self) -> None:
        self.workflow.task_data["task_configuration"] = "test"
        self.workflow.save()

        wr = self.workflow.create_child_worker(
            task_name="noop", task_data=BaseTaskData(task_configuration=None)
        )
        self.assertEqual(wr.task_data["task_configuration"], None)

    def test_lookup_single(self) -> None:
        lookup = "test"

        with mock.patch(
            "debusine.server.collections.lookup.lookup_single"
        ) as mock_lookup_single:
            self.work_request.lookup_single(
                lookup, default_category=CollectionCategory.TEST
            )

        mock_lookup_single.assert_called_once_with(
            lookup,
            workspace=self.work_request.workspace,
            pc=PermissionContext.for_work_request(
                work_request=self.work_request
            ),
            default_category=CollectionCategory.TEST,
            workflow_root=self.work_request.workflow_root,
            expect_type=LookupChildType.ANY,
            precache_result=None,
        )

    def test_lookup_multiple(self) -> None:
        lookup = LookupMultiple.model_validate(
            {"collection": "internal@collections", "child_type": "artifact"}
        )

        with mock.patch(
            "debusine.server.collections.lookup.lookup_multiple"
        ) as mock_lookup_multiple:
            self.work_request.lookup_multiple(
                lookup, default_category=CollectionCategory.TEST
            )

        mock_lookup_multiple.assert_called_once_with(
            lookup,
            workspace=self.work_request.workspace,
            pc=PermissionContext.for_work_request(
                work_request=self.work_request
            ),
            default_category=CollectionCategory.TEST,
            workflow_root=self.work_request.workflow_root,
            expect_type=LookupChildType.ANY,
        )

    @preserve_db_task_registry()
    def test_get_triggered_actions(self) -> None:
        """`get_triggered_actions` combines actions from various sources."""

        class EventReactionsTask(BaseTestWorkerTask[NoopData]):
            """Task that provides more event reactions."""

            @override
            def get_event_reactions(
                self,
                event_name: Literal[
                    "on_creation",
                    "on_unblock",
                    "on_assignment",
                    "on_success",
                    "on_failure",
                ],
            ) -> list[EventReaction]:
                event_reactions = list(super().get_event_reactions(event_name))
                if event_name == "on_failure":
                    event_reactions.append(ActionRetryWithDelays(delays=["1m"]))
                return event_reactions

        base_actions: dict[str, list[EventReaction]] = {
            action: [] for action in ActionTypes
        }
        base_actions[ActionTypes.RECORD_IN_TASK_HISTORY].append(
            ActionRecordInTaskHistory()
        )

        wr = self.playground.create_worker_task(task_name="eventreactionstask")
        self.assertEqual(wr.get_triggered_actions("on_success"), base_actions)
        self.assertEqual(
            wr.get_triggered_actions("on_failure"),
            {
                **base_actions,
                ActionTypes.RETRY_WITH_DELAYS: [
                    ActionRetryWithDelays(delays=["1m"])
                ],
            },
        )

        wr.event_reactions = EventReactions(
            on_failure=[ActionSendNotification(channel="admin-team")]
        )
        self.assertEqual(wr.get_triggered_actions("on_success"), base_actions)
        self.assertEqual(
            wr.get_triggered_actions("on_failure"),
            {
                **base_actions,
                ActionTypes.RETRY_WITH_DELAYS: [
                    ActionRetryWithDelays(delays=["1m"])
                ],
                ActionTypes.SEND_NOTIFICATION: [
                    ActionSendNotification(channel="admin-team"),
                ],
            },
        )

    def test_skip_if_lookup_result_changed_with_no_change(self) -> None:
        """Don't skip work request if lookup result did not change."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_pending=True)
        self.playground.create_collection("test", CollectionCategory.TEST)
        lookup = f"test@{CollectionCategory.TEST}/name:foo"
        action = ActionSkipIfLookupResultChanged(
            lookup=lookup, collection_item_id=None
        )
        wr.event_reactions = EventReactions(on_assignment=[action])
        wr.save()
        worker = self.playground.create_worker()

        # No exception is raised.
        wr.assign_worker(worker)

    def test_skip_if_lookup_result_changed_with_change(self) -> None:
        """Skip work request if lookup result changed."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_pending=True)
        collection = self.playground.create_collection(
            "test", CollectionCategory.TEST
        )
        lookup = f"test@{CollectionCategory.TEST}/name:foo"
        action = ActionSkipIfLookupResultChanged(
            lookup=lookup, collection_item_id=None
        )
        wr.event_reactions = EventReactions(on_assignment=[action])
        wr.save()
        CollectionItem.objects.create_from_artifact(
            self.playground.create_artifact()[0],
            parent_collection=collection,
            name="foo",
            data={},
            created_by_user=self.playground.get_default_user(),
        )
        worker = self.playground.create_worker()

        with self.assertRaises(SkipWorkRequest):
            wr.assign_worker(worker)

        wr.refresh_from_db()
        self.assertEqual(wr.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(wr.result, WorkRequest.Results.SKIPPED)
        self.assertEqual(
            wr.output_data,
            OutputData(skip_reason=f"Result of lookup {lookup!r} changed"),
        )

    def test_skip_if_lookup_result_changed_invalid_lookup(self) -> None:
        """Log an error and don't skip work request if the lookup is invalid."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_pending=True)
        self.playground.create_collection("test", CollectionCategory.TEST)
        action = ActionSkipIfLookupResultChanged(
            lookup="", collection_item_id=None
        )
        wr.event_reactions = EventReactions(on_assignment=[action])
        wr.save()
        worker = self.playground.create_worker()

        # No exception is raised.
        with self.assertLogsContains(
            f"Empty lookup in WorkRequest {wr}",
            logger="debusine.db.models.work_requests",
            level=logging.ERROR,
        ):
            wr.assign_worker(worker)

    def test_skip_if_lookup_result_changed_resolves_promise_artifact(
        self,
    ) -> None:
        """Skip if lookup result changed: resolve promise with artifact."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_pending=True)
        collection = self.playground.create_collection(
            "test", CollectionCategory.TEST
        )
        artifact_1, _ = self.playground.create_artifact()
        item_1 = CollectionItem.objects.create_from_artifact(
            artifact_1,
            parent_collection=collection,
            name="foo",
            data={},
            created_by_user=self.playground.get_default_user(),
        )
        lookup = f"test@{CollectionCategory.TEST}/name:foo"
        action = ActionSkipIfLookupResultChanged(
            lookup=lookup, collection_item_id=item_1.id, promise_name="expected"
        )
        wr.event_reactions = EventReactions(on_assignment=[action])
        wr.save()
        artifact_2, _ = self.playground.create_artifact()
        item_1.removed_at = timezone.now()
        item_1.save()
        CollectionItem.objects.create_from_artifact(
            artifact_2,
            parent_collection=collection,
            name="foo",
            data={},
            created_by_user=self.playground.get_default_user(),
        )
        worker = self.playground.create_worker()

        with self.assertRaises(SkipWorkRequest):
            wr.assign_worker(worker)

        wr.refresh_from_db()
        self.assertEqual(wr.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(wr.result, WorkRequest.Results.SKIPPED)
        self.assertEqual(
            wr.output_data,
            OutputData(skip_reason=f"Result of lookup {lookup!r} changed"),
        )
        self.assertEqual(
            wr.lookup_single(
                "internal@collections/name:expected",
                expect_type=LookupChildType.ARTIFACT,
            ).artifact,
            artifact_2,
        )

    def test_skip_if_lookup_result_changed_resolves_promise_bare_data(
        self,
    ) -> None:
        """Skip if lookup result changed: resolve promise with bare data."""
        wr = self.workflow.create_child_worker("noop")
        self.playground.advance_work_request(wr, mark_pending=True)
        collection = self.playground.create_collection(
            "test", CollectionCategory.TEST
        )
        item_1 = self.playground.create_bare_data_item(collection, "foo")
        lookup = f"test@{CollectionCategory.TEST}/name:foo"
        action = ActionSkipIfLookupResultChanged(
            lookup=lookup, collection_item_id=item_1.id, promise_name="expected"
        )
        wr.event_reactions = EventReactions(on_assignment=[action])
        wr.save()
        item_1.removed_at = timezone.now()
        item_1.save()
        item_2 = self.playground.create_bare_data_item(
            collection, "foo", data={"foo": "bar"}
        )
        worker = self.playground.create_worker()

        with self.assertRaises(SkipWorkRequest):
            wr.assign_worker(worker)

        wr.refresh_from_db()
        self.assertEqual(wr.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(wr.result, WorkRequest.Results.SKIPPED)
        self.assertEqual(
            wr.output_data,
            OutputData(skip_reason=f"Result of lookup {lookup!r} changed"),
        )
        self.assertEqual(
            wr.lookup_single(
                "internal@collections/name:expected",
                expect_type=LookupChildType.BARE,
            ).collection_item.data,
            item_2.data,
        )

    def test_update_collection_with_artifacts(self) -> None:
        """Update collection with artifacts on task completion."""
        c = self.workflow.internal_collection
        assert c is not None
        cm = CollectionManagerInterface.get_manager_for(c)
        a_replaced, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST,
            data={
                "deb_fields": {
                    "Package": "actiontest",
                    "Version": "1.0-1",
                },
                "type": "rebuild",
            },
        )
        cm.add_artifact(
            a_replaced, user=self.workflow.created_by, name="actiontest_rebuild"
        )
        item = cm.lookup("name:actiontest_rebuild")
        assert item is not None
        self.assertEqual(item.artifact, a_replaced)

        a_ignored, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST,
            data={
                "deb_fields": {
                    "Package": "ignored",
                    "Version": "1.0-1",
                },
                "type": "rebuild",
            },
        )
        cm.add_artifact(
            a_ignored, user=self.workflow.created_by, name="ignored_rebuild"
        )
        item = cm.lookup("name:ignored_rebuild")
        assert item is not None
        self.assertEqual(item.artifact, a_ignored)

        a_replacement, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST,
            data={
                'deb_fields': {
                    "Package": "actiontest",
                    "Version": "1.0-1+b1",
                },
                "type": "rebuild",
            },
            work_request=self.workflow,
        )

        a_added, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST,
            data={
                "deb_fields": {
                    "Package": "actiontest2",
                    "Version": "2.0-1+b1",
                },
                "type": "rebuild",
            },
            work_request=self.workflow,
        )

        action = ActionUpdateCollectionWithArtifacts(
            artifact_filters={
                "category": ArtifactCategory.TEST,
                "data__deb_fields__Package__startswith": "actiontest",
            },
            collection="internal@collections",
            name_template="{name}_{type}",
            variables={
                "$name": "deb_fields.Package",
                "$type": "type",
            },
        )
        self.workflow.event_reactions = EventReactions(
            on_success=[action],
            on_failure=[action],
        )
        self.workflow.mark_completed(WorkRequest.Results.SUCCESS)
        self.assertIsNotNone(
            CollectionItem.objects.get(
                parent_collection=c, artifact=a_replaced
            ).removed_at
        )
        item = cm.lookup("name:ignored_rebuild")
        assert item is not None
        self.assertEqual(item.artifact, a_ignored)
        item = cm.lookup("name:actiontest_rebuild")
        assert item is not None
        self.assertEqual(item.artifact, a_replacement)
        self.assertEqual(item.data, {"name": "actiontest", "type": "rebuild"})
        item = cm.lookup("name:actiontest2_rebuild")
        assert item is not None
        self.assertEqual(item.artifact, a_added)
        self.assertEqual(item.data, {"name": "actiontest2", "type": "rebuild"})

    def test_update_collection_with_artifacts_no_name_template(self) -> None:
        """Without name_template, variables are passed through."""
        c = Collection.objects.create(
            name="debian",
            category=CollectionCategory.ENVIRONMENTS,
            workspace=default_workspace(),
        )
        tarball, _ = self.playground.create_artifact(
            category=ArtifactCategory.SYSTEM_TARBALL,
            data={"architecture": "amd64"},
            work_request=self.workflow,
        )
        action = ActionUpdateCollectionWithArtifacts(
            artifact_filters={"category": ArtifactCategory.SYSTEM_TARBALL},
            collection="debian@debian:environments",
            variables={"codename": "bookworm", "variant": "buildd"},
        )
        self.workflow.event_reactions = EventReactions(on_success=[action])

        self.workflow.mark_completed(WorkRequest.Results.SUCCESS)

        item = c.manager.lookup(
            "match:format=tarball:codename=bookworm:architecture=amd64:"
            "variant=buildd"
        )
        assert item is not None
        self.assertIsNotNone(item.artifact, tarball)

    def test_update_collection_with_artifacts_created_at(self) -> None:
        sid = self.playground.create_collection("sid", CollectionCategory.SUITE)
        yesterday = timezone.now() - dt.timedelta(days=1)
        wr = self.playground.create_worker_task(
            event_reactions=EventReactions(
                on_success=[
                    ActionUpdateCollectionWithArtifacts(
                        artifact_filters={
                            "category": ArtifactCategory.BINARY_PACKAGE
                        },
                        collection=sid.id,
                        variables={
                            "component": "main",
                            "section": "devel",
                            "priority": "optional",
                        },
                        created_at=yesterday,
                    )
                ]
            )
        )
        binary = self.playground.create_minimal_binary_package_artifact()
        binary.created_by_work_request = wr
        binary.save()
        self.assertFalse(sid.child_items.exists())

        self.playground.advance_work_request(
            wr, result=WorkRequest.Results.SUCCESS
        )

        self.assertEqual(sid.child_items.get().created_at, yesterday)

    def test_update_collection_with_artifacts_constraint_violation(
        self,
    ) -> None:
        """Violating collection constraints is an error."""
        self.playground.create_artifact(
            category=ArtifactCategory.TEST, work_request=self.workflow
        )
        # This action does not specify an item name.
        action = ActionUpdateCollectionWithArtifacts(
            artifact_filters={"category": ArtifactCategory.TEST},
            collection="internal@collections",
        )
        self.workflow.event_reactions = EventReactions(on_success=[action])
        self.workflow.status = WorkRequest.Statuses.RUNNING
        with self.assertLogsContains(
            "Cannot replace or add artifact",
            logger="debusine.db.models",
            level=logging.ERROR,
        ):
            self.workflow.mark_completed(WorkRequest.Results.SUCCESS)

    def test_update_collection_with_artifacts_invalid_artifact_filters(
        self,
    ) -> None:
        """A failure to filter artifacts is an error."""
        action = ActionUpdateCollectionWithArtifacts(
            artifact_filters={"xxx": "yyy"},
            collection="internal@collections",
        )
        self.workflow.event_reactions = EventReactions(on_success=[action])
        with self.assertLogsContains(
            "Invalid update-collection-with-artifacts artifact_filters",
            logger="debusine.db.models",
            level=logging.ERROR,
        ):
            self.workflow.mark_completed(WorkRequest.Results.SUCCESS)

    def test_update_collection_with_artifacts_invalid_variables(self) -> None:
        """A failure to expand variables is an error."""
        self.playground.create_artifact(
            category=ArtifactCategory.TEST, work_request=self.workflow
        )
        action = ActionUpdateCollectionWithArtifacts(
            artifact_filters={"category": ArtifactCategory.TEST},
            collection="internal@collections",
            name_template="{name}",
            variables={"$name": "zzz"},
        )
        self.workflow.event_reactions = EventReactions(on_success=[action])
        with self.assertLogsContains(
            "Invalid update-collection-with-artifacts variables",
            logger="debusine.db.models",
            level=logging.ERROR,
        ):
            self.workflow.mark_completed(WorkRequest.Results.SUCCESS)

    def test_update_collection_with_artifacts_different_workspace(self) -> None:
        """The updated collection must be in the same workspace."""
        different_workspace = self.playground.create_workspace(
            name="test", public=True
        )
        Collection.objects.create(
            name="test_differentworkspace",
            category=CollectionCategory.WORKFLOW_INTERNAL,
            workspace=different_workspace,
        )
        action = ActionUpdateCollectionWithArtifacts(
            collection="test_differentworkspace@debusine:workflow-internal",
            artifact_filters={"category": ArtifactCategory.TEST},
        )
        self.workflow.event_reactions = EventReactions(on_success=[action])
        with self.assertLogsContains(
            "'test_differentworkspace@debusine:workflow-internal' does not "
            "exist or is hidden",
            logger="debusine.db.models",
            level=logging.ERROR,
        ):
            self.workflow.mark_completed(WorkRequest.Results.SUCCESS)

    def test_update_collection_with_data_on_creation(self) -> None:
        """
        Update collection with data on work request creation.

        This can also be done on completion, but
        `update-collection-with-artifacts` is usually more useful in that
        case.
        """
        assert self.workflow.internal_collection is not None
        cm = CollectionManagerInterface.get_manager_for(
            self.workflow.internal_collection
        )

        action = ActionUpdateCollectionWithData(
            collection="internal@collections",
            category=BareDataCategory.TEST,
            name_template="{package}_{version}",
            data={"package": "hello", "version": "1.0-1"},
        )
        self.workflow.create_child_worker(
            "noop", event_reactions=EventReactions(on_creation=[action])
        )
        item = cm.lookup("name:hello_1.0-1")
        assert item is not None
        self.assertEqual(item.data, {"package": "hello", "version": "1.0-1"})

    def test_update_collection_with_data_on_unblock(self) -> None:
        """Update collection with data when a work request is unblocked."""
        assert self.workflow.internal_collection is not None
        cm = CollectionManagerInterface.get_manager_for(
            self.workflow.internal_collection
        )

        action = ActionUpdateCollectionWithData(
            collection="internal@collections",
            category=BareDataCategory.TEST,
            name_template="{package}_{version}",
            data={"package": "hello", "version": "1.0-1"},
        )
        wr = self.workflow.create_child_worker(
            "noop", event_reactions=EventReactions(on_unblock=[action])
        )
        wr.unblock_strategy = WorkRequest.UnblockStrategy.MANUAL
        wr.save()

        self.assertIsNone(cm.lookup("name:hello_1.0-1"))

        wr.mark_pending()

        item = cm.lookup("name:hello_1.0-1")
        assert item is not None
        self.assertEqual(item.data, {"package": "hello", "version": "1.0-1"})

    def test_update_collection_with_data_created_at(self) -> None:
        collection = self.playground.create_singleton_collection(
            CollectionCategory.PACKAGE_BUILD_LOGS
        )
        yesterday = timezone.now() - dt.timedelta(days=1)
        wr = self.playground.create_worker_task(mark_pending=True)
        wr.event_reactions = EventReactions(
            on_success=[
                ActionUpdateCollectionWithData(
                    collection=collection.id,
                    category=BareDataCategory.PACKAGE_BUILD_LOG,
                    data={
                        "work_request_id": wr.id,
                        "vendor": "debian",
                        "codename": "sid",
                        "architecture": "all",
                        "srcpkg_name": "hello",
                        "srcpkg_version": "1.0",
                    },
                    created_at=yesterday,
                )
            ]
        )
        self.assertFalse(collection.child_items.exists())

        wr.mark_completed(WorkRequest.Results.SUCCESS)

        self.assertEqual(collection.child_items.get().created_at, yesterday)

    def test_update_collection_with_data_no_name_template(self) -> None:
        """Without name_template, name=None is passed."""
        action = ActionUpdateCollectionWithData(
            collection="internal@collections",
            category=BareDataCategory.TEST,
            data={"package": "hello", "version": "1.0-1"},
        )
        # We don't currently have a collection that accepts creating bare
        # data items without a name, but that's OK because the log message
        # is reasonably distinctive.
        with self.assertLogsContains(
            f"Cannot replace or add bare data of category "
            f"{BareDataCategory.TEST} to collection"
            f" {self.workflow.internal_collection}",
            logger="debusine.db.models",
            level=logging.ERROR,
        ):
            self.workflow.create_child_worker(
                "noop", event_reactions=EventReactions(on_creation=[action])
            )

    def test_update_collection_with_data_different_workspace(self) -> None:
        """The updated collection must be in the same workspace."""
        different_workspace = self.playground.create_workspace(
            name="test", public=True
        )
        Collection.objects.create(
            name="test_differentworkspace",
            category=CollectionCategory.WORKFLOW_INTERNAL,
            workspace=different_workspace,
        )
        action = ActionUpdateCollectionWithData(
            collection="test_differentworkspace@debusine:workflow-internal",
            category=BareDataCategory.TEST,
        )
        with self.assertLogsContains(
            "'test_differentworkspace@debusine:workflow-internal' does not "
            "exist or is hidden",
            logger="debusine.db.models",
            level=logging.ERROR,
        ):
            self.workflow.create_child_worker(
                "noop", event_reactions=EventReactions(on_creation=[action])
            )

    def test_retry_with_delays(self) -> None:
        """Automatically retry a work request on failure, with a delay."""
        action = ActionRetryWithDelays(
            delays=["5m", "30m", "1h", "2h", "2d", "1w"]
        )
        wr = self.workflow.create_child_worker(
            "noop",
            event_reactions=EventReactions(on_failure=[action]),
        )
        self.playground.advance_work_request(wr, mark_pending=True)

        for retry_count, delay in (
            (1, dt.timedelta(minutes=5)),
            (2, dt.timedelta(minutes=30)),
            (3, dt.timedelta(hours=1)),
            (4, dt.timedelta(hours=2)),
            (5, dt.timedelta(days=2)),
            (6, dt.timedelta(weeks=1)),
        ):
            with self.freeze_time() as now:
                wr.mark_completed(WorkRequest.Results.FAILURE)

            self.assertEqual(wr.status, WorkRequest.Statuses.COMPLETED)
            self.assertEqual(wr.result, WorkRequest.Results.FAILURE)
            wr_new = wr.superseded
            assert wr_new.workflow_data is not None
            self.assertEqual(wr_new.workflow_data.retry_count, retry_count)
            self.assertEqual(wr_new.dependencies.count(), retry_count)
            wr_delay = wr_new.dependencies.latest("created_at")
            self.assertEqual(wr_delay.status, WorkRequest.Statuses.PENDING)
            self.assertEqual(
                wr_delay.task_data["delay_until"],
                pydantic_core.to_jsonable_python(now + delay),
            )

            # Pretend that the delay task has expired.
            wr_delay.mark_completed(WorkRequest.Results.SUCCESS)
            wr_new.refresh_from_db()

            wr = wr_new

        # Failing once more does not retry again, since the delays list has
        # been exhausted.
        wr.mark_completed(WorkRequest.Results.FAILURE)

        self.assertEqual(wr.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(wr.result, WorkRequest.Results.FAILURE)
        self.assertFalse(hasattr(wr, "superseded"))

    def test_retry_with_delays_outside_workflow(self) -> None:
        """The retry-with-delays action may only be used in a workflow."""
        wr = self.playground.create_worker_task(
            event_reactions=EventReactions(
                on_failure=[ActionRetryWithDelays(delays=["1m"])]
            )
        )

        with self.assertRaisesRegex(
            CannotRetry,
            "retry-with-delays action may only be used in a workflow",
        ):
            self.playground.advance_work_request(
                wr, result=WorkRequest.Results.FAILURE
            )

    def test_record_in_task_history(self) -> None:
        """Record a task run in a task-history collection."""
        wr = self.playground.create_worker_task(
            assign_new_worker=True, mark_running=True
        )
        assert wr.started_at is not None
        output_data = OutputData(
            runtime_statistics=RuntimeStatistics(duration=10)
        )
        wr.mark_completed(WorkRequest.Results.SUCCESS, output_data=output_data)

        task_history = wr.workspace.get_singleton_collection(
            pc=PermissionContext.for_work_request(work_request=wr),
            category=CollectionCategory.TASK_HISTORY,
        )
        task_run = task_history.manager.lookup(
            f"last-success:{wr.task_type}:{wr.task_name}::"
        )
        assert task_run is not None
        self.assertEqual(
            task_run.name, f"{wr.task_type}:{wr.task_name}:::{wr.id}"
        )
        self.assertEqual(task_run.child_type, CollectionItem.Types.BARE)
        self.assertEqual(
            task_run.category, BareDataCategory.HISTORICAL_TASK_RUN
        )
        self.assertEqual(
            task_run.data,
            {
                "task_type": wr.task_type,
                "task_name": wr.task_name,
                "subject": None,
                "context": None,
                "timestamp": int(wr.started_at.timestamp()),
                "work_request_id": wr.id,
                "result": WorkRequest.Results.SUCCESS,
                "runtime_statistics": {"duration": 10},
            },
        )

    def test_record_in_task_history_no_collection(self) -> None:
        """We don't record a task run if there's no task-history collection."""
        wr = self.playground.create_worker_task(
            assign_new_worker=True, mark_running=True
        )
        wr.workspace.get_singleton_collection(
            pc=PermissionContext.for_work_request(work_request=wr),
            category=CollectionCategory.TASK_HISTORY,
        ).delete()
        output_data = OutputData(
            runtime_statistics=RuntimeStatistics(duration=10)
        )
        wr.mark_completed(WorkRequest.Results.SUCCESS, output_data=output_data)

        self.assertFalse(
            CollectionItem.objects.active()
            .filter(
                child_type=CollectionItem.Types.BARE,
                category=BareDataCategory.HISTORICAL_TASK_RUN,
            )
            .exists()
        )

    def test_record_in_task_history_bad_task_data(self) -> None:
        """We record a task run even if the task data is bad."""
        wr = self.playground.create_worker_task(
            assign_new_worker=True, mark_running=True
        )
        assert wr.started_at is not None
        wr.task_data["nonsense"] = 0
        assert isinstance(wr.configured_task_data, dict)
        wr.configured_task_data["nonsense"] = 0
        # Covering this case requires a manually-added event reaction; it
        # isn't added automatically if we can't construct the task.
        wr.event_reactions = EventReactions(
            on_failure=[ActionRecordInTaskHistory()]
        )
        wr.save()
        output_data = OutputData(
            runtime_statistics=RuntimeStatistics(duration=10)
        )
        wr.mark_completed(WorkRequest.Results.FAILURE, output_data=output_data)

        task_history = wr.workspace.get_singleton_collection(
            pc=PermissionContext.for_work_request(work_request=wr),
            category=CollectionCategory.TASK_HISTORY,
        )
        task_run = task_history.manager.lookup(
            f"last-failure:{wr.task_type}:{wr.task_name}::"
        )
        assert task_run is not None
        self.assertEqual(
            task_run.name, f"{wr.task_type}:{wr.task_name}:::{wr.id}"
        )
        self.assertEqual(task_run.child_type, CollectionItem.Types.BARE)
        self.assertEqual(
            task_run.category, BareDataCategory.HISTORICAL_TASK_RUN
        )
        self.assertEqual(
            task_run.data,
            {
                "task_type": wr.task_type,
                "task_name": wr.task_name,
                "subject": None,
                "context": None,
                "timestamp": int(wr.started_at.timestamp()),
                "work_request_id": wr.id,
                "result": WorkRequest.Results.FAILURE,
                "runtime_statistics": {"duration": 10},
            },
        )

    def test_record_in_task_history_with_subject_and_context(self) -> None:
        """We record subject and context for the task run if available."""
        source_artifact = self.playground.create_source_artifact(name="hello")
        environment = self.playground.create_debian_environment()
        assert environment.artifact_id is not None
        wr = self.playground.create_worker_task(
            assign_new_worker=True,
            mark_running=True,
            task_name="sbuild",
            task_data=SbuildData(
                input=SbuildInput(source_artifact=source_artifact.id),
                build_architecture="amd64",
                environment=environment.artifact_id,
            ),
        )
        assert wr.started_at is not None
        output_data = OutputData(
            runtime_statistics=RuntimeStatistics(duration=10)
        )
        wr.mark_completed(WorkRequest.Results.SUCCESS, output_data=output_data)

        task_history = wr.workspace.get_singleton_collection(
            pc=PermissionContext.for_work_request(work_request=wr),
            category=CollectionCategory.TASK_HISTORY,
        )
        task_run = task_history.manager.lookup(
            f"last-success:{wr.task_type}:{wr.task_name}:"
            f"hello:any%3Aamd64%3Aamd64"
        )
        assert task_run is not None
        self.assertEqual(
            task_run.name,
            f"{wr.task_type}:{wr.task_name}:hello:any%3Aamd64%3Aamd64:{wr.id}",
        )
        self.assertEqual(task_run.child_type, CollectionItem.Types.BARE)
        self.assertEqual(
            task_run.category, BareDataCategory.HISTORICAL_TASK_RUN
        )
        self.assertEqual(
            task_run.data,
            {
                "task_type": wr.task_type,
                "task_name": wr.task_name,
                "subject": "hello",
                "context": "any:amd64:amd64",
                "timestamp": int(wr.started_at.timestamp()),
                "work_request_id": wr.id,
                "result": WorkRequest.Results.SUCCESS,
                "runtime_statistics": {"duration": 10},
            },
        )


class WorkRequestWorkerTransactionTests(
    ChannelsHelpersMixin, TransactionTestCase
):
    """Test transactional behaviour of WorkRequest class scheduling."""

    def test_mark_completed_updates_workflow_last_activity_at(self) -> None:
        """
        Test mark_completed() updates its workflow's activity.

        Only when the work request is part of a workflow.
        """
        template = self.playground.create_workflow_template("test", "noop")
        workflow = self.playground.create_workflow(template, task_data={})
        sub_workflow = workflow.create_child_workflow(task_name="noop")
        child = sub_workflow.create_child_worker(task_name="noop")
        child.worker = self.playground.create_worker()
        child.configured_task_data = child.task_data
        child.status = WorkRequest.Statuses.RUNNING
        child.save()

        child.mark_completed(WorkRequest.Results.SUCCESS)
        self.assertTrue(child.workflows_need_update)

        update_workflows()

        workflow.refresh_from_db()
        sub_workflow.refresh_from_db()
        child.refresh_from_db()

        self.assertIsNotNone(sub_workflow.workflow_last_activity_at)
        self.assertIsNotNone(workflow.workflow_last_activity_at)
        self.assertEqual(workflow.workflow_last_activity_at, child.completed_at)
        self.assertEqual(
            sub_workflow.workflow_last_activity_at, child.completed_at
        )
        self.assertFalse(child.workflows_need_update)

    def test_mark_running_updates_workflow_last_activity_at(self) -> None:
        """
        Test mark_running() updates its workflow's activity.

        Only when the work request is part of a workflow.
        """
        template = self.playground.create_workflow_template("test", "noop")
        workflow = self.playground.create_workflow(template, task_data={})
        sub_workflow = workflow.create_child_workflow(task_name="noop")
        child = sub_workflow.create_child_worker(task_name="noop")
        child.worker = self.playground.create_worker()
        child.status = WorkRequest.Statuses.PENDING
        child.configured_task_data = child.task_data
        child.save()

        child.mark_running()
        self.assertTrue(child.workflows_need_update)

        update_workflows()

        workflow.refresh_from_db()
        sub_workflow.refresh_from_db()
        child.refresh_from_db()

        self.assertIsNotNone(sub_workflow.workflow_last_activity_at)
        self.assertIsNotNone(workflow.workflow_last_activity_at)
        self.assertEqual(workflow.workflow_last_activity_at, child.started_at)
        self.assertEqual(
            sub_workflow.workflow_last_activity_at, child.started_at
        )
        self.assertFalse(child.workflows_need_update)


class WorkflowTemplateManagerTests(TestCase):
    """Tests for WorkflowTemplateManager."""

    scenario = scenarios.DefaultContext()

    def test_in_current_scope(self) -> None:
        """Test the in_current_scope() QuerySet filter."""
        template = self.playground.create_workflow_template(
            name="test", task_name="noop"
        )
        with context.local():
            self.scenario.set_current()
            self.assertQuerySetEqual(
                WorkflowTemplate.objects.in_current_scope(), [template]
            )

        scope1 = Scope.objects.create(name="Scope1")
        with context.local():
            context.set_permission_context_scoped(
                scope1, PermissionContext.for_ui_anonymous()
            )
            self.assertQuerySetEqual(
                WorkflowTemplate.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"
        ):
            WorkflowTemplate.objects.in_current_scope()

    def test_in_current_workspace(self) -> None:
        """Test the in_current_workspace() QuerySet filter."""
        template = self.playground.create_workflow_template(
            name="test", task_name="noop"
        )
        context.set_permission_context_scoped(
            self.scenario.scope, self.scenario.context_ui()
        )

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

        workspace1 = self.playground.create_workspace(name="other", public=True)
        with context.local():
            workspace1.set_current()
            self.assertQuerySetEqual(
                WorkflowTemplate.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"
        ):
            WorkflowTemplate.objects.in_current_workspace()


class WorkflowTemplateTests(TestCase):
    """Unit tests for the WorkflowTemplate class."""

    scenario = scenarios.DefaultContext()

    def assert_static_parameters_must_be_dictionary(
        self, wt: WorkflowTemplate, messages: list[str] | None = None
    ) -> None:
        """Check that a WorkflowTemplate's static_parameters is a dictionary."""
        if messages is None:
            messages = ["static parameters must be a dictionary"]
        with self.assertRaises(ValidationError) as raised:
            wt.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {"static_parameters": messages},
        )

    def test_str(self) -> None:
        wt = self.playground.create_workflow_template(
            "sample", task_name="noop"
        )
        self.assertEqual(str(wt), "sample")

    def test_update_suites_with_well_known_name(self) -> None:
        """``update_suites`` can be created with the expected name."""
        self.playground.create_workflow_template(
            name="update-suites", task_name="update_suites"
        )

    def test_update_suites_without_well_known_name(self) -> None:
        """``update_suites`` cannot be created with an unexpected name."""
        with self.assertRaisesRegex(
            IntegrityError, "db_workflowtemplate_update_suites"
        ):
            self.playground.create_workflow_template(
                name="wrong", task_name="update_suites"
            )

    def test_clean_static_parameters_must_be_dict(self) -> None:
        """Ensure that static_parameters is a dict."""
        wt = WorkflowTemplate.objects.create(
            name="test",
            workspace=self.scenario.workspace,
            task_name="noop",
            static_parameters={},
        )
        wt.full_clean()

        wt.static_parameters = None
        self.assert_static_parameters_must_be_dictionary(wt)

        wt.static_parameters = ""
        self.assert_static_parameters_must_be_dictionary(wt)

        wt.static_parameters = 3
        self.assert_static_parameters_must_be_dictionary(wt)

        wt.static_parameters = []
        self.assert_static_parameters_must_be_dictionary(wt)

        wt.static_parameters = object()
        self.assert_static_parameters_must_be_dictionary(
            wt,
            messages=[
                'Value must be valid JSON.',
                'static parameters must be a dictionary',
            ],
        )

    @preserve_db_task_registry()
    def test_validate_static_parameters(self) -> None:
        """Test validation of static_parameters."""

        class ValidatingWorkflow(
            SampleWorkflow[BaseWorkflowData, BaseDynamicTaskData]
        ):
            """Workflow used to test validation of template data."""

            @override
            @classmethod
            def validate_static_parameters(cls, data: dict[str, Any]) -> None:
                """data-controlled validation."""
                if msg := data.get("error"):
                    raise ValueError(msg)

            @override
            def populate(self) -> None:
                raise NotImplementedError()

        wt = WorkflowTemplate.objects.create(
            name="test",
            workspace=self.scenario.workspace,
            task_name="validatingworkflow",
            static_parameters={},
        )
        wt.full_clean()

        wt.static_parameters = {"error": "example message"}
        with self.assertRaises(ValidationError) as raised:
            wt.full_clean()
        self.assertEqual(
            raised.exception.message_dict,
            {"static_parameters": ["example message"]},
        )

    def test_predicate_deny_from_context(self) -> None:
        """Test predicates propagating DENY from context_has_role."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template("test", "noop")
        with mock.patch(
            "debusine.db.models.workspaces.Workspace.has_role",
            return_value=False,
        ):
            self.assertFalse(template.can_run(self.scenario.context_ui()))
            self.assertFalse(template.can_configure(self.scenario.context_ui()))

    def test_can_display_public_workspace(self) -> None:
        """Test the can_retry predicate delegating to containing workspace."""
        user = self.scenario.user
        wr = self.playground.create_worker_task()
        for u, roles, expected in (
            (AnonymousUser(), [], True),
            (user, [], True),
            (user, [Workspace.Roles.VIEWER], True),
            (user, [Workspace.Roles.CONTRIBUTOR], True),
            (user, [Workspace.Roles.OWNER], True),
        ):
            with (
                self.subTest(user=repr(u), roles=repr(roles)),
                self.playground.assign_role(self.scenario.workspace, u, *roles),
            ):
                if u.is_authenticated:
                    pc = PermissionContext.for_ui_user(user)
                else:
                    pc = PermissionContext.for_ui_anonymous()
                self.assertEqual(wr.can_display(pc), expected)
                expected_qs = [wr] if expected else []
                self.assertQuerySetEqual(
                    WorkRequest.objects.filter(pk=wr.pk).can_display(pc),
                    expected_qs,
                )

    def test_can_display_private_workspace(self) -> None:
        """Test the can_retry predicate delegating to containing workspace."""
        other = self.playground.create_user("other")
        user = self.scenario.user
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        wr = self.playground.create_worker_task()
        for u, creator, roles, expected in (
            (AnonymousUser(), False, [], False),
            (user, False, [], False),
            (user, False, [Workspace.Roles.VIEWER], True),
            (user, False, [Workspace.Roles.CONTRIBUTOR], True),
            (user, False, [Workspace.Roles.OWNER], True),
            (user, True, [], False),
            (user, True, [Workspace.Roles.VIEWER], True),
            (user, True, [Workspace.Roles.CONTRIBUTOR], True),
            (user, True, [Workspace.Roles.OWNER], True),
        ):
            with (
                self.subTest(user=repr(u), creator=creator, roles=repr(roles)),
                self.playground.assign_role(self.scenario.workspace, u, *roles),
            ):
                if u.is_authenticated:
                    if creator:
                        wr.created_by = u
                    else:
                        wr.created_by = other
                    wr.save()
                    pc = PermissionContext.for_ui_user(user)
                else:
                    pc = PermissionContext.for_ui_anonymous()

                self.assertEqual(wr.can_display(pc), expected)
                expected_qs = [wr] if expected else []
                self.assertQuerySetEqual(
                    WorkRequest.objects.filter(pk=wr.pk).can_display(pc),
                    expected_qs,
                )

    def test_can_run_public_workspace(self) -> None:
        """Test can_run on public workspaces."""
        self.scenario.workspace.public = True
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template("test", "noop")

        self.assertPermission(
            "can_run",
            users=(AnonymousUser(), self.scenario.user),
            denied=[template],
        )

    def test_can_run_current_owner(self) -> None:
        """Test can_run on owned workspaces."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        user1 = self.playground.create_user("test1")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.OWNER,
            users=[self.scenario.user, user1],
        )
        templates = [
            self.playground.create_workflow_template("test", "noop"),
            self.playground.create_workflow_template(
                "test-restricted", "noop", restricted=True
            ),
        ]

        self.scenario.set_current()

        for template in templates:
            with self.subTest(
                restricted=template.restricted, user=self.scenario.user.username
            ):
                # Test shortcut code path
                with self.assertNumQueries(0):
                    self.assertTrue(
                        template.can_run(self.scenario.context_ui())
                    )

                with self.assertNumQueries(1):
                    self.assertTrue(
                        template.can_run(PermissionContext.for_ui_user(user1))
                    )

    def test_can_run_current_contributor(self) -> None:
        """Test can_run on a workspace where the user is a contributor."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        user1 = self.playground.create_user("test1")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            users=[self.scenario.user, user1],
        )
        templates = [
            self.playground.create_workflow_template("test", "noop"),
            self.playground.create_workflow_template(
                "test-restricted", "noop", restricted=True
            ),
        ]

        self.scenario.set_current()

        for template in templates:
            with self.subTest(
                restricted=template.restricted, user=self.scenario.user.username
            ):
                # Test shortcut code path
                with self.assertNumQueries(0):
                    pc = self.scenario.context_ui()
                    if template.restricted:
                        self.assertFalse(template.can_run(pc))
                    else:
                        self.assertTrue(template.can_run(pc))

                with self.assertNumQueries(1):
                    pc = PermissionContext.for_ui_user(user1)
                    if template.restricted:
                        self.assertFalse(template.can_run(pc))
                    else:
                        self.assertTrue(template.can_run(pc))

    def test_can_run_by_roles_unrestricted(self) -> None:
        """Test can_run behaviour with roles on an unrestricted template."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template("test", "noop")

        self.assertPermissionWhenRole(
            template.can_run,
            self.scenario.context_ui(),
            workspace_roles=(
                Workspace.Roles.OWNER,
                Workspace.Roles.CONTRIBUTOR,
            ),
            scope_roles=Scope.Roles.OWNER,
        )

    def test_can_run_by_roles_restricted(self) -> None:
        """Test can_run behaviour with roles on a restricted template."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template(
            "test", "noop", restricted=True
        )

        self.assertPermissionWhenRole(
            template.can_run,
            self.scenario.context_ui(),
            workspace_roles=Workspace.Roles.OWNER,
            scope_roles=Scope.Roles.OWNER,
        )
        self.assertNoPermissionWhenRole(
            template.can_run,
            self.scenario.context_ui(),
            workspace_roles=Workspace.Roles.CONTRIBUTOR,
        )

    def test_can_run_with_token(self) -> None:
        """Test can_run with a worker token."""
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[self.scenario.user]
        )
        template = self.playground.create_workflow_template("test", "noop")

        # Anonymous cannot
        self.assertPermission(
            "can_run",
            users=AnonymousUser(),
            denied=[template],
        )

        # User can
        self.assertPermission(
            "can_run",
            users=self.scenario.user,
            allowed=[template],
        )

        # A worker token cannot in any case
        self.assertPermission(
            "can_run",
            users=[None],
            denied=[template],
            token=self.playground.create_worker_token(),
        )

    def test_can_configure_public_workspace(self) -> None:
        """Test can_configure on public workspaces."""
        self.scenario.workspace.public = True
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template("test", "noop")

        self.assertPermission(
            "can_configure",
            users=(AnonymousUser(), self.scenario.user),
            denied=[template],
        )

    def test_can_configure_current_owner(self) -> None:
        """Test can_configure on owned workspaces."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        user1 = self.playground.create_user("test1")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.OWNER,
            users=[self.scenario.user, user1],
        )
        template = self.playground.create_workflow_template("test", "noop")

        self.scenario.set_current()

        # Test shortcut code path
        with self.assertNumQueries(1):
            self.assertTrue(template.can_configure(self.scenario.context_ui()))

        with self.assertNumQueries(2):
            self.assertTrue(
                template.can_configure(PermissionContext.for_ui_user(user1))
            )

    def test_can_configure_current_contributor(self) -> None:
        """Test can_configure on a workspace where the user is a contributor."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template("test", "noop")
        user1 = self.playground.create_user("test1")
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.CONTRIBUTOR,
            users=[self.scenario.user, user1],
        )

        self.scenario.set_current()

        # Test shortcut code path
        with self.assertNumQueries(0):
            self.assertFalse(template.can_configure(self.scenario.context_ui()))

        with self.assertNumQueries(1):
            self.assertFalse(
                template.can_configure(PermissionContext.for_ui_user(user1))
            )

    def test_can_configure_by_roles(self) -> None:
        """Test can_configure behaviour with roles."""
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        template = self.playground.create_workflow_template("test", "noop")

        self.assertPermissionWhenRole(
            template.can_configure,
            self.scenario.context_ui(),
            workspace_roles=Workspace.Roles.OWNER,
            scope_roles=Scope.Roles.OWNER,
        )

    def test_can_configure_with_token(self) -> None:
        """Test can_configure with a worker token."""
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[self.scenario.user]
        )
        template = self.playground.create_workflow_template("test", "noop")

        # Anonymous cannot
        self.assertPermission(
            "can_configure",
            users=AnonymousUser(),
            denied=[template],
        )

        # User can
        self.assertPermission(
            "can_configure",
            users=self.scenario.user,
            allowed=[template],
        )

        # A worker token cannot in any case
        self.assertPermission(
            "can_configure",
            users=[None],
            denied=[template],
            token=self.playground.create_worker_token(),
        )

    def test_can_configure_can_grant_extra_groups(self) -> None:
        template = self.playground.create_workflow_template("test", "noop")
        template.extra_groups.set([self.scenario.scope_owners])
        self.scenario.scope_owners.users.add(self.scenario.user)
        self.scenario.workspace_owners.users.add(self.scenario.user)
        self.assertPermission(
            "can_configure",
            users=self.scenario.user,
            allowed=[template],
        )

    def test_can_configure_cannot_grant_extra_groups(self) -> None:
        template = self.playground.create_workflow_template("test", "noop")
        template.extra_groups.set([self.scenario.scope_owners])
        self.scenario.workspace_owners.users.add(self.scenario.user)
        self.assertPermission(
            "can_configure",
            users=self.scenario.user,
            denied=[template],
        )

    def test_completed_workflows(self) -> None:
        """Test completed_workflows()."""
        template = self.playground.create_workflow_template(
            name="noop-template", task_name="noop"
        )
        self.playground.create_workflow(
            task_name=template, result=WorkRequest.Results.SUCCESS
        )

        context._workspace.set(template.workspace)

        self.assertEqual(template.completed_workflows(), 1)

    def test_running_workflows(self) -> None:
        """Test running_workflows()."""
        template = self.playground.create_workflow_template(
            name="noop-template", task_name="noop"
        )
        self.playground.create_workflow(mark_running=True)

        context._workspace.set(template.workspace)

        self.assertEqual(template.running_workflows(), 1)

    def test_needs_input_workflows(self) -> None:
        """Test needs_input_workflows()."""
        template = self.playground.create_workflow_template(
            name="noop-template", task_name="noop"
        )
        workflow = self.playground.create_workflow(mark_running=True)

        workflow.workflow_runtime_status = (
            WorkRequest.RuntimeStatuses.NEEDS_INPUT
        )
        workflow.save()

        assert workflow.workspace

        context._workspace.set(workflow.workspace)

        self.assertEqual(template.needs_input_workflows(), 1)

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

        template = self.playground.create_workflow_template("test", "noop")
        request = RequestFactory().get("/")
        ui = template.ui(request)
        self.assertIsInstance(ui, WorkflowTemplateUI)

    def test_get_absolute_url(self) -> None:
        """Test for get_absolute_url."""
        template = self.playground.create_workflow_template("test", "noop")
        self.assertEqual(
            template.get_absolute_url(),
            f"/{self.scenario.scope.name}/{self.scenario.workspace.name}"
            f"/workflow-template/{template.name}/",
        )

    def test_get_absolute_url_different_scope(self) -> None:
        """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)
        template = self.playground.create_workflow_template(
            "test", "noop", workspace=other_workspace
        )
        self.assertEqual(
            template.get_absolute_url(),
            f"/{other_scope.name}/{other_workspace.name}"
            f"/workflow-template/{template.name}/",
        )


class DeleteUnusedTests(TestCase):
    """Test DeleteUnused."""

    def test_one(self) -> None:
        """One work request to delete."""
        wr1 = self.playground.create_worker_task()
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, {wr1})
        self.assertEqual(to_delete.collections, set())
        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [])

    def test_multiple(self) -> None:
        """Multiple work requests to delete."""
        wr1 = self.playground.create_worker_task()
        wr2 = self.playground.create_workflow()
        c2 = wr2.internal_collection
        wr3 = self.playground.create_worker_task(parent=wr2)
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, {wr1, wr2, wr3})
        self.assertEqual(to_delete.collections, {c2})
        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [])

    def test_referenced_by_collection_item(self) -> None:
        """
        Work requests are not unused if referenced by a CollectionItem.

        Keep work requests that:
        - CollectionItem.created_by_workflow -> FK WorkRequest
        - CollectionItem.removed_by_workflow -> FK WorkRequest
        - CollectionItem.id != WorkRequest.internal_collection.id
        """
        work_request_keep = self.playground.create_workflow()
        work_request_delete = self.playground.create_workflow()
        collection_1 = work_request_delete.internal_collection
        assert collection_1 is not None

        # Ensure all work requests are up for deletion (they are not
        # referenced by any CollectionItem yet)
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(
            to_delete.work_requests,
            {work_request_delete, work_request_keep},
        )

        # Make work_request_keep to be kept: referenced by a
        # CollectionItem (via created_by_workflow) and
        # WorkRequest.internal_collection != collection_1
        item_1 = collection_1.manager.add_bare_data(
            name="Item-1",
            user=self.playground.get_default_user(),
            category=BareDataCategory.TEST,
            workflow=work_request_keep,
        )

        # work_request_delete is in a collection but deleted: referenced
        # by a CollectionItem, but
        # WorkRequest.internal-collection == collection_2
        collection_1.manager.add_bare_data(
            name="Item-2",
            user=self.playground.get_default_user(),
            category=BareDataCategory.TEST,
            workflow=work_request_delete,
        )

        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, {work_request_delete})

        # work_request_keep still kept: now is referenced by a
        # CollectionItem.removed_by_workflow
        collection_1.manager.remove_item(
            item=item_1,
            user=self.playground.get_default_user(),
            workflow=work_request_keep,
        )

        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, {work_request_delete})

        # Check again if no CollectionItem referenced the work requests:
        # they are deleted
        CollectionItem.objects.all().delete()
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(
            to_delete.work_requests,
            {work_request_keep, work_request_delete},
        )

        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [])

    def test_valid_parent(self) -> None:
        """Keep work requests with non-unused parent."""
        wr1 = self.playground.create_workflow()
        wr2 = self.playground.create_worker_task(parent=wr1)
        wr3 = self.playground.create_worker_task()
        to_delete = DeleteUnused(
            WorkRequest.objects.filter(pk__in=[wr2.pk, wr3.pk])
        )
        self.assertEqual(to_delete.work_requests, {wr3})
        self.assertEqual(to_delete.collections, set())
        to_delete.perform_deletions()
        self.assertQuerySetEqual(
            WorkRequest.objects.all(), [wr1, wr2], ordered=False
        )

    def test_valid_child(self) -> None:
        """Keep work requests with non-unused child."""
        wr1 = self.playground.create_workflow()
        wr2 = self.playground.create_worker_task(parent=wr1)
        to_delete = DeleteUnused(WorkRequest.objects.filter(pk__in=[wr1.pk]))
        self.assertEqual(to_delete.work_requests, set())
        self.assertEqual(to_delete.collections, set())
        to_delete.perform_deletions()
        self.assertQuerySetEqual(
            WorkRequest.objects.all(), [wr1, wr2], ordered=False
        )

    def test_unused_child_with_build_log(self) -> None:
        """Keep unused work requests unused child which has a build log."""
        user = self.playground.get_default_user()
        wr1 = self.playground.create_workflow()
        wr2 = self.playground.create_worker_task(parent=wr1)
        build_log = self.playground.create_build_log_artifact()
        build_logs = wr2.workspace.get_singleton_collection(
            pc=PermissionContext.for_ui_user(user),
            category=CollectionCategory.PACKAGE_BUILD_LOGS,
        )
        build_logs.manager.add_artifact(
            build_log,
            user=user,
            workflow=wr2,
            variables={
                "work_request_id": wr2.id,
                "vendor": "debian",
                "codename": "bookworm",
                "architecture": "amd64",
                "srcpkg_name": "hello-2",
                "srcpkg_version": "1.0-2",
            },
        )

        to_delete = DeleteUnused(
            WorkRequest.objects.filter(pk__in=[wr1.pk, wr2.pk])
        )
        self.assertEqual(to_delete.work_requests, set())
        self.assertEqual(to_delete.collections, set())
        to_delete.perform_deletions()
        self.assertQuerySetEqual(
            WorkRequest.objects.all(), [wr1, wr2], ordered=False
        )

    def test_with_collection(self) -> None:
        """Also delete internal collections."""
        wr1 = self.playground.create_workflow()
        c1 = wr1.internal_collection
        assert c1 is not None
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, {wr1})
        self.assertEqual(to_delete.collections, {c1})
        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [])
        self.assertQuerySetEqual(Collection.objects.filter(pk=c1.pk), [])

    def test_delete_with_internal_collection(self) -> None:
        """Delete a work request referenced by an internal collection item."""
        work_request = self.playground.create_workflow()
        collection = work_request.internal_collection
        assert collection is not None

        collection.manager.add_bare_data(
            name="Item",
            user=self.playground.get_default_user(),
            category=BareDataCategory.TEST,
            workflow=work_request,
        )

        # Work request is now expired but referenced by an item in an internal
        # collection: it should be deleted
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, {work_request})

        # Check that it is actually deleted
        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [])
        self.assertQuerySetEqual(
            Collection.objects.filter(pk=collection.pk), []
        )

    def test_delete_workflow_with_build_logs(self) -> None:
        """Work requests referenced by build logs are kept."""
        user = self.playground.get_default_user()
        wr1 = self.playground.create_workflow()
        build_log = self.playground.create_build_log_artifact()
        build_logs = wr1.workspace.get_singleton_collection(
            pc=PermissionContext.for_ui_user(user),
            category=CollectionCategory.PACKAGE_BUILD_LOGS,
        )
        build_logs.manager.add_artifact(
            build_log,
            user=user,
            workflow=wr1,
            variables={
                "work_request_id": wr1.id,
                "vendor": "debian",
                "codename": "bookworm",
                "architecture": "amd64",
                "srcpkg_name": "hello-2",
                "srcpkg_version": "1.0-2",
            },
        )
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, set())
        self.assertEqual(to_delete.collections, set())
        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [wr1])

    def test_delete_work_request_with_build_logs(self) -> None:
        """Work requests referenced by build logs are kept."""
        user = self.playground.get_default_user()
        wr1 = self.playground.create_worker_task()
        build_log = self.playground.create_build_log_artifact()
        build_logs = wr1.workspace.get_singleton_collection(
            pc=PermissionContext.for_ui_user(user),
            category=CollectionCategory.PACKAGE_BUILD_LOGS,
        )
        build_logs.manager.add_artifact(
            build_log,
            user=user,
            workflow=wr1,
            variables={
                "work_request_id": wr1.id,
                "vendor": "debian",
                "codename": "bookworm",
                "architecture": "amd64",
                "srcpkg_name": "hello-2",
                "srcpkg_version": "1.0-2",
            },
        )
        to_delete = DeleteUnused(WorkRequest.objects.all())
        self.assertEqual(to_delete.work_requests, set())
        self.assertEqual(to_delete.collections, set())
        to_delete.perform_deletions()
        self.assertQuerySetEqual(WorkRequest.objects.all(), [wr1])


class TaskConfigurationTests(TestCase):
    """test :py:class:`WorkRequest` task configuration handling."""

    scenario = scenarios.DefaultContext()

    def test_worker_unblock_configures(self) -> None:
        wr = self.playground.create_worker_task()
        self.assertIsNone(wr.configured_task_data)
        wr.mark_pending()
        self.assertIsNotNone(wr.configured_task_data)

    def test_worker_mark_running_wants_configure(self) -> None:
        wr = self.playground.create_worker_task(assign_new_worker=True)
        # Simulate that the task was accydentally not configured
        wr.configured_task_data = None
        wr.save()
        with self.assertRaisesRegex(
            StatusChangeError, r"Unconfigured Worker task marked running"
        ):
            wr.mark_running()

    def test_worker_create_child(self) -> None:
        workflow = self.playground.create_workflow()
        wr = workflow.create_child_worker("noop")
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)

    def test_server_unblock_configures(self) -> None:
        wr = self.playground.create_server_task()
        self.assertIsNone(wr.configured_task_data)
        wr.mark_pending()
        self.assertIsNotNone(wr.configured_task_data)

    def test_server_mark_running_wants_configure(
        self,
    ) -> None:
        wr = self.playground.create_server_task(mark_pending=True)
        # Set it back to unconfigured
        wr.configured_task_data = None
        # Assign a worker without configuring the task for it
        wr.worker = self.playground.create_worker()
        wr.save()
        with self.assertRaisesRegex(
            StatusChangeError, r"Unconfigured Server task marked running"
        ):
            wr.mark_running()

    def test_server_create_child(self) -> None:
        workflow = self.playground.create_workflow()
        wr = workflow.create_child_server("servernoop")
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)

    def test_internal_unblock_leaves_unconfigured(self) -> None:
        """Internal tasks are not configured on unblock."""
        wr = self.playground.create_internal_task(task_name="workflow")
        self.assertIsNone(wr.configured_task_data)
        wr.mark_pending()
        self.assertIsNone(wr.configured_task_data)

    def test_internal_mark_running_ok_unconfigured(
        self,
    ) -> None:
        wr = self.playground.create_internal_task(
            task_name="workflow", mark_pending=True
        )
        wr.mark_running()
        self.assertEqual(wr.status, WorkRequest.Statuses.RUNNING)
        self.assertIsNone(wr.configured_task_data)

    def test_internal_create_child(self) -> None:
        workflow = self.playground.create_workflow()
        wr = workflow.create_child_internal(task_name="workflow")
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)

    def test_workflow_create_child(self) -> None:
        workflow = self.playground.create_workflow()
        wr = workflow.create_child_workflow("noop")
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)

    def test_workflow_unblock_configures(self) -> None:
        workflow = self.playground.create_workflow()
        wr = workflow.create_child_workflow("noop")
        self.assertIsNone(wr.configured_task_data)
        wr.mark_pending()
        self.assertIsNotNone(wr.configured_task_data)

    def test_workflow_mark_running_wants_configure(self) -> None:
        workflow = self.playground.create_workflow()
        # Set it back to unconfigured
        workflow.configured_task_data = None
        workflow.save()
        with self.assertRaisesRegex(
            StatusChangeError, r"Unconfigured Workflow task marked running"
        ):
            workflow.mark_running()

    @override_permission(WorkflowTemplate, "can_run", AllowAll)
    def test_workflow_create_workflow(self) -> None:
        self.scenario.set_current()
        template = self.playground.create_workflow_template("test", "noop")
        wr = WorkRequest.objects.create_workflow(template=template, data={})
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)

    def test_signing_unblock_configures(self) -> None:
        wr = self.playground.create_signing_task()
        self.assertIsNone(wr.configured_task_data)
        wr.mark_pending()
        self.assertIsNotNone(wr.configured_task_data)

    def test_signing_mark_running_wants_configure(
        self,
    ) -> None:
        wr = self.playground.create_signing_task(mark_pending=True)
        # Set it back to unconfigured
        wr.configured_task_data = None
        # Assign a worker without configuring the task for it
        wr.worker = self.playground.create_worker()
        wr.save()
        with self.assertRaisesRegex(
            StatusChangeError, r"Unconfigured Signing task marked running"
        ):
            wr.mark_running()

    def test_signing_new_child(self) -> None:
        workflow = self.playground.create_workflow()
        wr = workflow.create_child_signing("noop")
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)

    def test_wait_unblock_configures(self) -> None:
        package = self.playground.create_upload_artifacts().upload
        wr = self.playground.create_wait_task(
            task_name="externaldebsign",
            task_data=ExternalDebsignData(unsigned=package.pk),
            workflow_data=WorkRequestWorkflowData(needs_input=True),
        )
        self.assertIsNone(wr.configured_task_data)
        wr.mark_pending()
        self.assertIsNotNone(wr.configured_task_data)

    def test_wait_mark_running_wants_configure(
        self,
    ) -> None:
        package = self.playground.create_upload_artifacts().upload
        wr = self.playground.create_wait_task(
            task_name="externaldebsign",
            task_data=ExternalDebsignData(unsigned=package.pk),
            workflow_data=WorkRequestWorkflowData(needs_input=True),
            mark_pending=True,
        )
        # Set it back to unconfigured
        wr.configured_task_data = None
        # Assign a worker without configuring the task for it
        wr.worker = self.playground.create_worker()
        wr.save()
        with self.assertRaisesRegex(
            StatusChangeError, r"Unconfigured Wait task marked running"
        ):
            wr.mark_running()

    def test_wait_new_child(self) -> None:
        workflow = self.playground.create_workflow()
        package = self.playground.create_upload_artifacts().upload
        wr = workflow.create_child_wait(
            task_name="externaldebsign",
            task_data=ExternalDebsignData(unsigned=package.pk),
            workflow_data=WorkRequestWorkflowData(needs_input=True),
        )
        self.assertEqual(wr.status, WorkRequest.Statuses.BLOCKED)
        self.assertIsNone(wr.configured_task_data)


class LookupEnvironmentTests(TestCase):
    """test :py:class:`WorkRequest` lookup_environment function."""

    scenario = scenarios.DefaultContext()
    work_request: ClassVar[WorkRequest]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.work_request = cls.playground.create_worker_task()

    def mock_lookup_single(self, **kwargs: Any) -> None:
        self.lookup_single = self.enterContext(
            mock.patch.object(self.work_request, "lookup_single", **kwargs)
        )

    def patch_executor(
        self, backend: BackendType, image_category: ExecutorImageCategory
    ) -> mock.MagicMock:
        """Patch executor for `backend` and return its mock."""
        executor = mock.MagicMock(spec=ExecutorInterface)
        executor.image_category = image_category
        self.enterContext(
            mock.patch.dict(
                "debusine.tasks.executors.base._backends", {backend: executor}
            )
        )
        return executor

    def test_get_environment_int(self) -> None:
        lookup = 1
        self.mock_lookup_single(return_value="RETURN")
        self.assertEqual(self.work_request.lookup_environment(lookup), "RETURN")
        self.lookup_single.assert_called_once_with(
            lookup,
            default_category=None,
            expect_type=LookupChildType.ARTIFACT,
        )

    def test_get_environment_with_architecture_and_backend(self) -> None:
        self.mock_lookup_single(return_value="RETURN")
        self.assertEqual(
            self.work_request.lookup_environment(
                "debian/match:codename=bookworm",
                backend=BackendType.UNSHARE,
            ),
            "RETURN",
        )
        self.lookup_single.assert_called_once_with(
            "debian/match:codename=bookworm:format=tarball:backend=unshare",
            default_category=None,
            expect_type=LookupChildType.ARTIFACT,
        )

    def test_get_environment_with_architecture_from_worker(self) -> None:
        self.patch_executor(BackendType.UNSHARE, ExecutorImageCategory.IMAGE)
        self.mock_lookup_single(return_value="RETURN")
        self.assertEqual(
            self.work_request.lookup_environment(
                "debian/match:codename=bookworm",
                architecture="amd64",
                backend=BackendType.UNSHARE,
                try_variant="",
            ),
            "RETURN",
        )
        self.lookup_single.assert_called_once_with(
            "debian/match:codename=bookworm:architecture=amd64:"
            "format=image:backend=unshare:variant=",
            default_category=None,
            expect_type=LookupChildType.ARTIFACT,
        )

    def test_get_environment_with_architecture_missing(self) -> None:
        self.patch_executor(BackendType.UNSHARE, ExecutorImageCategory.IMAGE)
        self.mock_lookup_single(return_value="RETURN")
        self.assertEqual(
            self.work_request.lookup_environment(
                "debian/match:codename=bookworm",
                backend=BackendType.UNSHARE,
                try_variant="",
            ),
            "RETURN",
        )
        self.lookup_single.assert_called_once_with(
            "debian/match:codename=bookworm:format=image:"
            "backend=unshare:variant=",
            default_category=None,
            expect_type=LookupChildType.ARTIFACT,
        )

    def test_get_environment_tries_variant(self) -> None:
        self.patch_executor(BackendType.UNSHARE, ExecutorImageCategory.TARBALL)
        self.mock_lookup_single(return_value="RETURN")
        self.assertEqual(
            self.work_request.lookup_environment(
                "debian/match:codename=bookworm",
                architecture="amd64",
                backend=BackendType.UNSHARE,
                try_variant="samplebasetaskwithexecutorandarchitecture",
            ),
            "RETURN",
        )
        self.lookup_single.assert_called_once_with(
            "debian/match:codename=bookworm:architecture=amd64:"
            "format=tarball:backend=unshare:"
            "variant=samplebasetaskwithexecutorandarchitecture",
            default_category=None,
            expect_type=LookupChildType.ARTIFACT,
        )

    def test_get_environment_without_try_variant(self) -> None:
        self.patch_executor(BackendType.UNSHARE, ExecutorImageCategory.TARBALL)
        self.mock_lookup_single(return_value="RETURN")
        self.assertEqual(
            self.work_request.lookup_environment(
                "debian/match:codename=bookworm",
                architecture="amd64",
                backend=BackendType.UNSHARE,
                try_variant=None,
            ),
            "RETURN",
        )
        self.lookup_single.assert_called_once_with(
            "debian/match:codename=bookworm:architecture=amd64:"
            "format=tarball:backend=unshare",
            default_category=None,
            expect_type=LookupChildType.ARTIFACT,
        )
