# 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 Debusine class."""

import base64
import contextlib
import datetime as dt
import hashlib
import io
import json
import logging
import re
import tarfile
import urllib
from collections.abc import AsyncGenerator
from pathlib import Path
from typing import Any, override
from unittest import IsolatedAsyncioTestCase, mock
from unittest.mock import call
from urllib.parse import quote, urlencode

import aiohttp
import pydantic
import pydantic_core
import requests
import responses

from debusine.artifacts import LocalArtifact, Upload, WorkRequestDebugLogs
from debusine.artifacts.local_artifact import NewLocalFile, NewRemoteFile
from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianSourcePackage,
    DebianUpload,
    DebusineTaskConfiguration,
    EmptyArtifactData,
    TaskTypes,
    WorkRequestResults,
)
from debusine.assets import AssetCategory, KeyPurpose, SigningKeyData
from debusine.client import task_configuration
from debusine.client.debusine import (
    Debusine,
    ServerNotificationDecodeError,
    serialize_local_artifact,
)
from debusine.client.exceptions import (
    TokenDisabledError,
    UnexpectedResponseError,
)
from debusine.client.models import (
    ArtifactResponse,
    AssetPermissionCheckResponse,
    AssetResponse,
    AssetsResponse,
    CollectionData,
    CollectionDataNew,
    CollectionIdentifierResponse,
    CollectionItemData,
    CollectionItemResponse,
    CollectionRelationResponse,
    CollectionRelationSetResponse,
    CollectionRelationType,
    CollectionRoles,
    CollectionTokenGrantRequest,
    CollectionTokenGrantResponse,
    CreateWorkflowRequest,
    FileRequest,
    FileResponse,
    FilesResponseType,
    LookupChildType,
    LookupMultipleResponse,
    LookupResultType,
    LookupSingleResponseArtifact,
    OnWorkRequestCompleted,
    RelationCreateRequest,
    RelationResponse,
    RelationType,
    RelationsResponse,
    RuntimeParameter,
    TaskConfigurationCollection,
    TaskConfigurationCollectionUpdateResults,
    WorkRequestExternalDebsignRequest,
    WorkRequestRequest,
    WorkflowTemplateData,
    WorkflowTemplateDataNew,
    WorkspaceInheritanceChain,
    WorkspaceInheritanceChainElement,
)
from debusine.client.server_notifications import ServerNotifications
from debusine.tasks.models import OutputData
from debusine.test import TestCase
from debusine.test.test_utils import (
    create_artifact_response,
    create_listing_response,
    create_remote_artifact,
    create_work_request_response,
)


class SerializeLocalArtifactTests(TestCase):
    """Tests for ``serialize_local_artifact``."""

    @override
    def setUp(self) -> None:
        super().setUp()

        # Register a custom artifact category, taking care that it doesn't
        # pollute other tests.
        class TestArtifact(LocalArtifact[EmptyArtifactData]):
            """Custom artifact for testing."""

            _category = ArtifactCategory.TEST

        def unregister_test_artifact() -> None:
            del LocalArtifact._local_artifacts_category_to_class[
                ArtifactCategory.TEST
            ]

        self.addCleanup(unregister_test_artifact)

        self.test_artifact_class = TestArtifact
        self.category = TestArtifact._category
        self.data = EmptyArtifactData()

        self.local_artifact = TestArtifact(
            category=self.category, data=self.data
        )

    def test_structure(self) -> None:
        """``serialize_local_artifact`` returns the expected dictionary."""
        workspace = "workspace"
        self.assertEqual(
            serialize_local_artifact(self.local_artifact, workspace=workspace),
            {
                "category": self.category,
                "data": self.data.model_dump(),
                "expire_at": None,
                "files": {},
                "workspace": workspace,
                "work_request": None,
            },
        )

    def test_guesses_content_type(self) -> None:
        """``serialize_local_artifact`` can guess Content-Type."""
        files = []
        expected_files: list[tuple[Path, str]] = []
        for suffix, content_type in (
            (".buildinfo", "text/plain; charset=utf-8"),
            (".dsc", "text/plain; charset=utf-8"),
            (".changes", "text/plain; charset=utf-8"),
            (".md", "text/markdown; charset=utf-8"),
        ):
            file = self.create_temporary_file(suffix=suffix)
            files.append(NewLocalFile(file=file))
            expected_files.append((file, content_type))
        for contents, content_type in (
            (b"text\n", "text/plain; charset=us-ascii"),
            (
                Path(__file__).read_bytes(),
                "text/x-script.python; charset=utf-8",
            ),
            (b"\0\1\2\3\4\5\6\7", "application/octet-stream; charset=binary"),
        ):
            file = self.create_temporary_file(contents=contents)
            files.append(NewLocalFile(file=file))
            expected_files.append((file, content_type))
        artifact = self.test_artifact_class._create(data=self.data, files=files)

        serialized = serialize_local_artifact(artifact, workspace=None)

        self.assertEqual(
            serialized["files"],
            {
                file.name: FileRequest.create_from(
                    file, content_type=content_type
                ).model_dump()
                for file, content_type in expected_files
            },
        )

    def test_set_content_type(self) -> None:
        """``serialize_local_artifact`` accepts an explicit Content-Type."""
        file = self.create_temporary_file()
        artifact = self.test_artifact_class._create(
            data=self.data,
            files=[NewLocalFile(file=file, content_type="text/x-foo")],
        )

        serialized = serialize_local_artifact(artifact, workspace=None)

        self.assertEqual(
            serialized["files"],
            {
                file.name: FileRequest.create_from(
                    file, content_type="text/x-foo"
                ).model_dump()
            },
        )

    def test_no_workspace(self) -> None:
        """``serialize_local_artifact`` returns dictionary without workspace."""
        self.assertNotIn(
            "workspace",
            serialize_local_artifact(self.local_artifact, workspace=None),
        )

    def test_validate_model_on_serialization(self) -> None:
        """``serialize_local_artifact`` raises ValueError: invalid artifact."""
        file = self.create_temporary_file(suffix=".changes")
        artifact = Upload(
            files={file.name: file},
            category=Upload._category,
            data=DebianUpload(
                type="dpkg",
                changes_fields={
                    "Architecture": "amd64",
                    "Files": [
                        {"name": "hello_1.0_amd64.deb"},
                    ],
                },
            ),
        )

        # Remove the only file. The "artifact" object will not validate
        artifact.files.popitem()

        with self.assertRaisesRegex(
            ValueError, r"Expecting 1 \.changes file in \[\]"
        ):
            serialize_local_artifact(artifact, workspace="System")

    def test_with_work_request(self) -> None:
        """``serialize_local_artifact`` includes ``work_request``."""
        workspace = "workspace"
        work_request = 5

        serialized = serialize_local_artifact(
            self.local_artifact, workspace=workspace, work_request=work_request
        )

        self.assertEqual(serialized["work_request"], work_request)

    def test_remote_files(self) -> None:
        """``serialize_local_artifact`` handles remote files."""
        changes_file = self.create_temporary_file(suffix=".changes")
        tar_file = self.create_temporary_file(suffix=".tar.xz")
        artifact = self.test_artifact_class._create(
            data=self.data,
            files=[
                NewLocalFile(file=changes_file),
                NewRemoteFile(
                    path_in_artifact=tar_file.name,
                    file_request=FileRequest.create_from(tar_file),
                ),
            ],
        )

        serialized = serialize_local_artifact(artifact, workspace=None)

        self.assertEqual(
            serialized["files"],
            {
                file.name: FileRequest.create_from(
                    file, content_type=content_type
                ).model_dump()
                for file, content_type in (
                    (changes_file, "text/plain; charset=utf-8"),
                    (tar_file, None),
                )
            },
        )

    def test_with_expire_at(self) -> None:
        """``serialize_local_artifact`` includes ``expire_at``."""
        expire_at = dt.datetime.now(tz=dt.UTC) + dt.timedelta(days=1)

        serialized = serialize_local_artifact(
            self.local_artifact, workspace="workspace", expire_at=expire_at
        )

        self.assertEqual(
            serialized["expire_at"], pydantic_core.to_jsonable_python(expire_at)
        )


class DebusineTests(TestCase):
    """Tests for the Debusine class."""

    @override
    def setUp(self) -> None:
        super().setUp()
        self.api_url = "https://debusine.debian.org/api"

        self.token = "token"
        self.scope = "scope"
        self.logger = mock.create_autospec(spec=logging.Logger)
        # Add a trailing slash to the API URL to ensure that all parts of
        # the client strip it correctly.
        self.debusine = Debusine(
            self.api_url + "/", self.token, scope=self.scope, logger=self.logger
        )

    def test_webui_url(self) -> None:
        for api_url, path, expected in (
            (
                "https://debusine.debian.org/api",
                "/a/b",
                "https://debusine.debian.org/a/b",
            ),
            (
                "https://debusine.debian.org/subdir/api",
                "/a/b",
                "https://debusine.debian.org/subdir/a/b",
            ),
            (
                "https://debusine.debian.org/",
                "/a/b",
                "https://debusine.debian.org/a/b",
            ),
        ):
            for test_api_url in (api_url, api_url + "/"):
                for test_path in (path, path.lstrip("/")):
                    with self.subTest(api_url=test_api_url, path=test_path):
                        debusine = Debusine(
                            test_api_url,
                            self.token,
                            logger=self.logger,
                            scope="scope",
                        )
                        self.assertEqual(
                            debusine.webui_url(test_path), expected
                        )

    @responses.activate
    def test_work_request_iter(self) -> None:
        """Debusine.work_request_iter yields work requests."""
        work_requests = [
            create_work_request_response(id=1),
            create_work_request_response(id=2),
        ]
        listing_response = create_listing_response(*work_requests)

        responses.add(
            responses.GET,
            f'{self.debusine.api_url}/work-request/',
            json=listing_response.model_dump(mode="json"),
        )

        self.assertEqual(list(self.debusine.work_request_iter()), work_requests)
        self.assert_token_key_included_in_all_requests('token')

    @responses.activate
    def test_work_request_get(self) -> None:
        """Debusine.work_request_get returns the work request."""
        work_request_id = 10
        work_request_response = create_work_request_response(id=work_request_id)

        responses.add(
            responses.GET,
            f'{self.debusine.api_url}/work-request/{work_request_id}/',
            json=work_request_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.work_request_get(work_request_id),
            work_request_response,
        )
        self.assert_token_key_included_in_all_requests('token')

    @responses.activate
    def test_work_request_update(self) -> None:
        """Debusine.work_request_update patches the work request."""
        work_request_id = 10
        work_request_response = create_work_request_response(
            id=work_request_id, priority_adjustment=100
        )

        responses.add(
            responses.PATCH,
            f"{self.debusine.api_url}/work-request/{work_request_id}/",
            json=work_request_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.work_request_update(
                work_request_id, priority_adjustment=100
            ),
            work_request_response,
        )
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_work_request_retry(self) -> None:
        """Debusine.work_request_retry triggers a retry."""
        work_request_id = 10
        work_request_response = create_work_request_response(
            id=work_request_id + 1,
        )

        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/work-request/{work_request_id}/retry/",
            json=work_request_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.work_request_retry(work_request_id),
            work_request_response,
        )
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_work_request_abort(self) -> None:
        """Debusine.work_request_abort triggers an abort."""
        work_request_id = 10
        work_request_response = create_work_request_response(id=work_request_id)

        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/work-request/{work_request_id}/abort/",
            json=work_request_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.work_request_abort(work_request_id),
            work_request_response,
        )
        self.assert_token_key_included_in_all_requests("token")

    def verify_work_request_create_success(
        self, workspace: str | None = None
    ) -> None:
        """
        Verify Debusine.work_request_create().

        :param workspace: if it's None: assert that the client's request
          does not contain "workspace" (it's assigned by the server). The
          server response contains workspace="System".

          If it's not None: the request contains "workspace" with the
          correct value.
        """
        expected_work_request = workspace if workspace is not None else "System"

        # response always contains "workspace". It's "system" if it was
        # no in the request
        expected_work_request_response = create_work_request_response(
            workspace=expected_work_request
        )

        responses.add(
            responses.POST,
            f'{self.debusine.api_url}/work-request/',
            json=expected_work_request_response.model_dump(mode="json"),
        )

        work_request_fields: dict[str, Any] = {
            "task_name": "sbuild",
            "task_data": {},
            "event_reactions": {},
            "workspace": workspace,
        }
        # these fields are always posted
        fields_to_post = {"task_name", "task_data", "event_reactions"}

        # "workspace" is posted only if it's not None
        if workspace is not None:
            fields_to_post.add("workspace")

        # work_request_to_post can contain "workspace=None". The method
        # Debusine.work_request_create will remove this key if it's None
        work_request_to_post = WorkRequestRequest(**work_request_fields)

        actual_work_request_response = self.debusine.work_request_create(
            work_request_to_post
        )
        self.assertEqual(
            actual_work_request_response, expected_work_request_response
        )

        self.assert_token_key_included_in_all_requests(self.token)

        # The posted (over the wire) WorkRequestRequest does not contain
        # "workspace" if it was None
        self.assertEqual(
            json.loads(responses.calls[0].request.body or "null"),
            work_request_to_post.model_dump(include=fields_to_post),
        )

    @responses.activate
    def test_work_request_create_success(self) -> None:
        """
        Post a new work request for workspace "Testing".

        The server will assign it to "Testing".
        """
        self.verify_work_request_create_success(workspace="Testing")

    def test_work_request_completed_path(self) -> None:
        """Debusine.work_request_completed_path() return expected path."""
        work_request_id = 67
        self.assertEqual(
            Debusine.work_request_completed_path(work_request_id),
            f"/work-request/{work_request_id}/completed/",
        )

    @responses.activate
    def test_work_request_completed_update(self) -> None:
        """work_request_completed_path() make the expected HTTP request."""
        work_request_id = 68

        responses.add(
            responses.PUT,
            f"{self.api_url}/1.0"
            + self.debusine.work_request_completed_path(work_request_id),
            "",
        )

        self.debusine.work_request_completed_update(
            work_request_id, WorkRequestResults.SUCCESS, OutputData()
        )

    @responses.activate
    def test_work_request_create_no_workspace_success(self) -> None:
        """
        Post a new work request without a workspace.

        Assert that the request does not contain "workspace" (not even
        None, as the server expects no workspace field and assigns it to the
        default).
        """
        self.verify_work_request_create_success(workspace=None)

    @responses.activate
    def test_work_request_external_debsign_get(self) -> None:
        """
        Debusine.work_request_external_debsign_get returns the work request.

        It uses a different endpoint from `Debusine.work_request_get` in
        order to ensure that dynamic task data is resolved.
        """
        work_request_id = 10
        work_request_response = create_work_request_response(
            id=work_request_id,
            task_type="Wait",
            task_name="externaldebsign",
            task_data={"unsigned": "lookup"},
            dynamic_task_data={"unsigned_id": 1},
        )

        responses.add(
            responses.GET,
            f'{self.debusine.api_url}/work-request/{work_request_id}/'
            f'external-debsign/',
            json=work_request_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.work_request_external_debsign_get(work_request_id),
            work_request_response,
        )
        self.assert_token_key_included_in_all_requests('token')

    @responses.activate
    def test_work_request_external_debsign_complete(self) -> None:
        """
        Debusine.work_request_external_debsign_complete sends the request.

        It provides external `debsign` data to the work request.
        """
        work_request_id = 10
        external_debsign_request = WorkRequestExternalDebsignRequest(
            signed_artifact=2
        )
        work_request_response = create_work_request_response(id=work_request_id)

        responses.add(
            responses.POST,
            f'{self.debusine.api_url}/work-request/{work_request_id}/'
            f'external-debsign/',
            json=work_request_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.work_request_external_debsign_complete(
                work_request_id, external_debsign_request
            ),
            work_request_response,
        )
        self.assert_token_key_included_in_all_requests('token')

    def create_workflow_template_data(
        self,
        id: int = 42,  # noqa: A002
    ) -> WorkflowTemplateData:
        return WorkflowTemplateData(
            id=id,
            name=f"pipeline{id}",
            task_name="debian_pipeline",
            static_parameters={
                "qa_suite": "sid@debian:suite",
            },
            runtime_parameters={
                "qa_suite": RuntimeParameter.ANY,
            },
        )

    @responses.activate
    def test_workflow_template_create(self) -> None:
        workspace = "workspace"
        workflow_template_response = self.create_workflow_template_data()
        workflow_template_body = dict(workflow_template_response)
        workflow_template_body.pop("id")
        workflow_template = WorkflowTemplateDataNew(**workflow_template_body)

        responses.add(
            responses.POST,
            f'{self.debusine.api_url}/workflow-template/{workspace}/',
            status=requests.codes.created,
            json=workflow_template_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.workflow_template_create(
                workspace, workflow_template
            ),
            workflow_template_response,
        )
        self.assert_token_key_included_in_all_requests('token')

    @responses.activate
    def test_workflow_template_iter(self) -> None:
        workspace = "workspace"
        workflow_templates = [
            self.create_workflow_template_data(id=1),
            self.create_workflow_template_data(id=2),
        ]
        listing_response = create_listing_response(*workflow_templates)

        responses.add(
            responses.GET,
            f'{self.debusine.api_url}/workflow-template/{workspace}/',
            json=listing_response.model_dump(mode="json"),
        )

        self.assertEqual(
            list(self.debusine.workflow_template_iter(workspace)),
            workflow_templates,
        )
        self.assert_token_key_included_in_all_requests('token')

    @responses.activate
    def test_workflow_template_get(self) -> None:
        workspace = "workspace"
        workflow_template_response = self.create_workflow_template_data()

        responses.add(
            responses.GET,
            (
                f'{self.debusine.api_url}/workflow-template/{workspace}/'
                f'{workflow_template_response.id}/'
            ),
            json=workflow_template_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.workflow_template_get(
                workspace, workflow_template_response.id
            ),
            workflow_template_response,
        )
        self.assert_token_key_included_in_all_requests('token')

    @responses.activate
    def test_workflow_template_update(self) -> None:
        workspace = "workspace"
        workflow_template_response = self.create_workflow_template_data()

        responses.add(
            responses.PATCH,
            (
                f"{self.debusine.api_url}/workflow-template/{workspace}/"
                f"{workflow_template_response.id}/"
            ),
            json=workflow_template_response.model_dump(mode="json"),
        )

        self.assertEqual(
            self.debusine.workflow_template_update(
                workspace,
                workflow_template_response,
            ),
            workflow_template_response,
        )
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_workflow_template_delete(self) -> None:
        workspace = "workspace"
        responses.add(
            responses.DELETE,
            f"{self.debusine.api_url}/workflow-template/{workspace}/test/",
            status=requests.codes.no_content,
        )

        self.debusine.workflow_template_delete(workspace, "test")
        self.assert_token_key_included_in_all_requests("token")

    def verify_workflow_create_success(
        self,
        template_name: str = "test",
        task_name: str = "",
        workspace: str | None = None,
    ) -> None:
        """
        Verify Debusine.workflow_create().

        :param workspace: if it's None: assert that the client's request
          does not contain "workspace" (it's assigned by the server). The
          server response contains workspace="System".

          If it's not None: the request contains "workspace" with the
          correct value.
        """
        # Response always contains "workspace".  It's "system" if it was not
        # in the request.
        expected_workflow_response = create_work_request_response(
            workspace=(workspace if workspace is not None else "System")
        )

        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/workflow/",
            json=expected_workflow_response.model_dump(mode="json"),
            status=requests.codes.created,
        )

        workflow_fields: dict[str, Any] = {
            "template_name": template_name,
            "task_name": task_name,
            "task_data": {},
            "workspace": workspace,
        }
        # these fields are always posted
        fields_to_post = {"template_name", "task_name", "task_data"}

        # "workspace" is posted only if it's not None
        if workspace is not None:
            fields_to_post.add("workspace")

        # workflow_to_post can contain "workspace=None". The method
        # Debusine.workflow_create will remove this key if it's None.
        workflow_to_post = CreateWorkflowRequest(**workflow_fields)

        actual_workflow_response = self.debusine.workflow_create(
            workflow_to_post
        )
        self.assertEqual(actual_workflow_response, expected_workflow_response)

        self.assert_token_key_included_in_all_requests(self.token)

        # The posted (over the wire) CreateWorkflowRequest does not contain
        # "workspace" if it was None.
        self.assertEqual(
            json.loads(responses.calls[0].request.body or "null"),
            workflow_to_post.model_dump(include=fields_to_post),
        )

    @responses.activate
    def test_workflow_create_success(self) -> None:
        """Post a new workflow."""
        self.verify_workflow_create_success(workspace="Testing")

    @responses.activate
    def test_workflow_create_no_workspace_success(self) -> None:
        """Post a new workflow with no workspace."""
        self.verify_workflow_create_success(workspace=None)

    @responses.activate
    def test_workflow_create_no_template_success(self) -> None:
        """Post a new workflow with no template."""
        self.verify_workflow_create_success(template_name="", task_name="test")

    def files_for_artifact(self) -> dict[str, Path]:
        """Return dictionary with files to upload in an artifact."""
        return {
            "README": self.create_temporary_file(),
            ".dirstamp": self.create_temporary_file(),
        }

    @responses.activate
    def test_artifact_get(self) -> None:
        """Test artifact_get() return ArtifactResponse."""
        artifact_id = 2
        workspace = "test"
        download_artifact_tar_gz_url = "https://example.com/path/"
        category = "Testing"
        created_at = dt.datetime.now(dt.UTC)
        data: dict[str, Any] = {}

        get_artifact_response = create_artifact_response(
            id=artifact_id,
            workspace=workspace,
            category=category,
            created_at=created_at,
            data=data,
            download_tar_gz_url=download_artifact_tar_gz_url,
        )

        responses.add(
            responses.GET,
            f"{self.api_url}/1.0/artifact/{get_artifact_response.id}/",
            json=get_artifact_response.model_dump(mode="json"),
        )

        artifact = self.debusine.artifact_get(artifact_id)

        self.assert_token_key_included_in_all_requests(self.token)

        self.assertEqual(
            artifact,
            create_artifact_response(
                id=artifact_id,
                category=category,
                data=data,
                created_at=created_at,
                workspace=workspace,
                download_tar_gz_url=download_artifact_tar_gz_url,
            ),
        )

    @responses.activate
    def test_artifact_get_with_workspace(self) -> None:
        """Test artifact_get() with a workspace."""
        artifact_id = 2
        workspace = "test"
        download_artifact_tar_gz_url = "https://example.com/path/"
        category = "Testing"
        created_at = dt.datetime.now(dt.UTC)
        data: dict[str, Any] = {}

        get_artifact_response = create_artifact_response(
            id=artifact_id,
            workspace=workspace,
            category=category,
            created_at=created_at,
            data=data,
            download_tar_gz_url=download_artifact_tar_gz_url,
        )

        responses.add(
            responses.GET,
            (
                f"{self.api_url}/1.0/artifact/{workspace}/"
                f"{get_artifact_response.id}/"
            ),
            json=get_artifact_response.model_dump(mode="json"),
        )

        artifact = self.debusine.artifact_get(artifact_id, workspace=workspace)

        self.assert_token_key_included_in_all_requests(self.token)

        self.assertEqual(
            artifact,
            create_artifact_response(
                id=artifact_id,
                category=category,
                data=data,
                created_at=created_at,
                workspace=workspace,
                download_tar_gz_url=download_artifact_tar_gz_url,
            ),
        )

    @responses.activate
    def test_artifact_create_success(self) -> None:
        """Test artifact_create succeeds."""
        files = self.files_for_artifact()

        workspace = "test"
        work_request = 5
        expire_at = dt.datetime.now(tz=dt.UTC) + dt.timedelta(days=1)

        artifact_response = create_artifact_response(
            id=2,
            workspace=workspace,
            category="Testing",
            created_at=dt.datetime.now(dt.UTC),
            data={},
            files_to_upload=list(files.keys()),
        )

        responses.add(
            responses.POST,
            f"{self.api_url}/1.0/artifact/",
            json=artifact_response.model_dump(mode="json"),
            status=requests.codes.created,
        )

        artifact_to_post = WorkRequestDebugLogs(
            category=ArtifactCategory.WORK_REQUEST_DEBUG_LOGS,
            files=files,
            data=EmptyArtifactData(),
        )

        actual_artifact_response = self.debusine.artifact_create(
            artifact_to_post,
            workspace=workspace,
            work_request=work_request,
            expire_at=expire_at,
        )

        self.assertEqual(actual_artifact_response, artifact_response)

        self.assert_token_key_included_in_all_requests(self.token)

        assert responses.calls[0].request.body is not None
        self.assertEqual(
            json.loads(responses.calls[0].request.body),
            serialize_local_artifact(
                artifact_to_post,
                workspace=workspace,
                work_request=work_request,
                expire_at=expire_at,
            ),
        )

    @responses.activate
    def test_upload_files(self) -> None:
        """Client upload files to the server (Debusine.upload_files)."""
        file_relative = self.create_temporary_file(prefix="main.c")

        file_paths = [
            self.create_temporary_file(prefix="README"),
            self.create_temporary_file(prefix="AUTHORS?"),
            file_relative,
        ]

        artifact_id = 2

        urls = []
        for file_path in file_paths:
            file_path_encoded = urllib.parse.quote(file_path.name)
            url = (
                f"{self.api_url}/{Debusine.API_VERSION}/artifact/"
                f"{artifact_id}/files/{file_path_encoded}/"
            )
            responses.add(responses.PUT, url)
            urls.append(url)

        files_to_upload = {}
        for file_path in file_paths:
            files_to_upload[file_path.name] = file_path

        files_to_upload[file_relative.name] = file_relative.relative_to(
            file_relative.parent
        )

        self.debusine.upload_files(
            artifact_id, files_to_upload, file_relative.parent
        )

        for url in urls:
            self.assertTrue(responses.assert_call_count(url, 1))
        self.assertEqual(
            self.logger.info.mock_calls,
            [
                call(
                    "Uploading %s to %s",
                    path,
                    self.debusine._url_for_file_in_artifact(
                        artifact_id, path.name
                    ),
                )
                for path in file_paths
            ],
        )

    def test_upload_files_invalid_base_directory(self) -> None:
        """Base directory is None but there is a relative file path."""
        file_path = "src/main.c"

        files_to_upload = {"some/file/main.c": Path(file_path)}

        artifact_id = 2

        with self.assertRaisesRegex(
            ValueError,
            f"{file_path} is relative: base_directory parameter cannot be None",
        ):
            self.debusine.upload_files(artifact_id, files_to_upload, None)

    @responses.activate
    def test_upload_artifact(self) -> None:
        """upload_artifact() uploads the artifact and associated files."""
        category = ArtifactCategory.WORK_REQUEST_DEBUG_LOGS
        workspace = "the-workspace"
        work_request = 6

        data = EmptyArtifactData()
        filename = "README.txt"
        file_path = self.create_temporary_file()
        files = {
            filename: file_path,
            "already-uploaded-file.txt": self.create_temporary_file(),
        }

        local_artifact = WorkRequestDebugLogs(
            category=category, data=data, files=files
        )

        remote_id = 2

        # Create the artifact request
        responses.add(
            "POST",
            f"{self.debusine.api_url}/artifact/",
            create_artifact_response(
                base_url="https://debusine.debian.org/",
                id=remote_id,
                workspace=workspace,
                files_to_upload=[filename],
            ).model_dump_json(),
            status=requests.codes.created,
        )

        # Upload the file request
        responses.add(
            "PUT",
            f"{self.debusine.api_url}/artifact/{remote_id}/files/{filename}/",
        )
        remote_artifact = self.debusine.upload_artifact(
            local_artifact, workspace=workspace, work_request=work_request
        )
        self.assertEqual(
            remote_artifact,
            create_remote_artifact(
                base_url="https://debusine.debian.org/",
                id=remote_id,
                workspace=workspace,
            ),
        )

        self.assertEqual(len(responses.calls), 2)

    @responses.activate
    def test_relation_create(self) -> None:
        """relation_create() makes an HTTP post."""
        artifact_id = 1
        target_id = 2

        url = self.api_url + "/1.0/artifact-relation/"

        response_body = RelationResponse(
            id=7,
            artifact=artifact_id,
            target=target_id,
            type=RelationType.EXTENDS,
        )
        responses.add(
            responses.POST,
            url,
            json=dict(response_body),
        )

        self.debusine.relation_create(
            artifact_id, target_id, RelationType.EXTENDS
        )

        responses.assert_call_count(url, 1)

        assert responses.calls[0].request.body is not None
        request_body = RelationCreateRequest.model_validate_json(
            responses.calls[0].request.body
        )

        self.assertEqual(
            request_body,
            RelationCreateRequest(
                artifact=artifact_id,
                target=target_id,
                type=RelationType.EXTENDS,
            ),
        )

    def test_relation_list_no_ids(self) -> None:
        """relation_list() requires an artifact or target ID."""
        with self.assertRaisesRegex(
            ValueError, "Exactly one of artifact_id or target_id must be set"
        ):
            self.debusine.relation_list()

    def test_relation_list_both_ids(self) -> None:
        """relation_list() requires only one of artifact or target ID."""
        with self.assertRaisesRegex(
            ValueError, "Exactly one of artifact_id or target_id must be set"
        ):
            self.debusine.relation_list(artifact_id=1, target_id=2)

    @responses.activate
    def test_relation_list_artifact(self) -> None:
        """relation_list() with an artifact ID makes an HTTP GET."""
        url = self.api_url + "/1.0/artifact-relation/?artifact=1"
        relation_response = RelationResponse(
            id=7, artifact=1, target=2, type=RelationType.EXTENDS
        )
        responses.add(responses.GET, url, json=[relation_response.model_dump()])

        self.assertEqual(
            self.debusine.relation_list(artifact_id=1),
            RelationsResponse([relation_response]),
        )

        responses.assert_call_count(url, 1)

    @responses.activate
    def test_relation_list_target(self) -> None:
        """relation_list() with a target ID makes an HTTP GET."""
        url = self.api_url + "/1.0/artifact-relation/?target_artifact=2"
        relation_response = RelationResponse(
            id=7, artifact=1, target=2, type=RelationType.EXTENDS
        )
        responses.add(responses.GET, url, json=[relation_response.model_dump()])

        self.assertEqual(
            self.debusine.relation_list(target_id=2),
            RelationsResponse([relation_response]),
        )

        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_relation_list(self) -> None:
        """collection_relation_list() makes an HTTP GET."""
        workspace = "System"
        url = f"{self.api_url}/1.0/collection-relation/{workspace}/"
        collection_relation_response = CollectionRelationResponse(
            source_collection=CollectionIdentifierResponse(
                name="sid",
                category=CollectionCategory.SUITE,
            ),
            target_collection=CollectionIdentifierResponse(
                name="trixie",
                category=CollectionCategory.SUITE,
            ),
            type=CollectionRelationType.SUITE_REQUIRES,
            position=1,
        )

        responses.add(
            responses.GET,
            url,
            json=create_listing_response(
                collection_relation_response
            ).model_dump(mode="json"),
        )

        self.assertEqual(
            list(self.debusine.collection_relation_list(workspace)),
            [collection_relation_response],
        )

        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_relation_list_with_filters(self) -> None:
        """collection_relation_list() with filters makes an HTTP GET."""
        workspace = "System"
        url = (
            f"{self.api_url}/1.0/collection-relation/{workspace}/"
            "?source=sid%40debian%3Asuite&target=trixie%40debian%3Asuite"
            "&type=suite-requires"
        )
        collection_relation_response = CollectionRelationResponse(
            source_collection=CollectionIdentifierResponse(
                name="sid",
                category=CollectionCategory.SUITE,
            ),
            target_collection=CollectionIdentifierResponse(
                name="trixie",
                category=CollectionCategory.SUITE,
            ),
            type=CollectionRelationType.SUITE_REQUIRES,
            position=1,
        )

        etag = '"870803b953e5fee3e"'
        responses.add(
            responses.GET,
            url,
            json=create_listing_response(
                collection_relation_response
            ).model_dump(mode="json"),
            adding_headers={"ETag": etag},
        )

        self.assertEqual(
            list(
                self.debusine.collection_relation_list(
                    workspace,
                    source="sid@debian:suite",
                    target="trixie@debian:suite",
                    relation_type=CollectionRelationType.SUITE_REQUIRES,
                )
            ),
            [collection_relation_response],
        )

        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_relation_list_for_edit(self) -> None:
        workspace = "System"
        url = (
            f"{self.api_url}/1.0/collection-relation/{workspace}/"
            "?source=sid%40debian%3Asuite&type=suite-requires"
        )
        relation_1 = CollectionRelationResponse(
            source_collection=CollectionIdentifierResponse(
                name="sid",
                category=CollectionCategory.SUITE,
            ),
            target_collection=CollectionIdentifierResponse(
                name="trixie",
                category=CollectionCategory.SUITE,
            ),
            type=CollectionRelationType.SUITE_REQUIRES,
            position=1,
        )
        relation_2 = CollectionRelationResponse(
            source_collection=CollectionIdentifierResponse(
                name="sid",
                category=CollectionCategory.SUITE,
            ),
            target_collection=CollectionIdentifierResponse(
                name="trixie-backports",
                category=CollectionCategory.SUITE,
            ),
            type=CollectionRelationType.SUITE_REQUIRES,
            position=2,
        )

        etag = '"first-page-etag"'
        responses.add(
            responses.GET,
            url,
            json=create_listing_response(relation_1, relation_2).model_dump(
                mode="json"
            ),
            adding_headers={"ETag": etag},
        )

        relations, returned_etag = (
            self.debusine.collection_relation_list_for_edit(
                workspace,
                source="sid@debian:suite",
                relation_type=CollectionRelationType.SUITE_REQUIRES,
            )
        )

        self.assertEqual(relations, [relation_1, relation_2])
        self.assertEqual(returned_etag, etag)

    @responses.activate
    def test_collection_relation_list_for_edit_raise_multiple_pages_error(
        self,
    ) -> None:
        workspace = "System"
        url = (
            f"{self.api_url}/1.0/collection-relation/{workspace}/"
            "?source=sid%40debian%3Asuite&type=suite-requires"
        )
        next_url = f"{url}&page=2"
        first_relation = CollectionRelationResponse(
            source_collection=CollectionIdentifierResponse(
                name="sid",
                category=CollectionCategory.SUITE,
            ),
            target_collection=CollectionIdentifierResponse(
                name="trixie",
                category=CollectionCategory.SUITE,
            ),
            type=CollectionRelationType.SUITE_REQUIRES,
            position=1,
        )
        first_etag = '"first-page-etag"'
        responses.add(
            responses.GET,
            url,
            json=create_listing_response(
                first_relation, next_url=next_url
            ).model_dump(mode="json"),
            adding_headers={"ETag": first_etag},
        )

        with self.assertRaisesRegex(
            UnexpectedResponseError,
            re.escape(
                "Expected a single paginated response page from "
                f"{url}, but server returned additional pages"
            ),
        ):
            self.debusine.collection_relation_list_for_edit(
                workspace,
                source="sid@debian:suite",
                relation_type=CollectionRelationType.SUITE_REQUIRES,
            )

    @responses.activate
    def test_collection_relation_set(self) -> None:
        workspace = "Debian"
        source_collection = "personal@debian:suite"
        relation_type = CollectionRelationType.SUITE_REQUIRES

        targets = ["trixie@debian:suite", "trixie-backports@debian:suite"]

        url = (
            f"{self.api_url}/1.0/collection-relation/{workspace}/"
            f"{quote(source_collection)}/{relation_type}/"
        )

        expected_response = CollectionRelationSetResponse(
            targets=[
                CollectionIdentifierResponse(
                    name="trixie",
                    category=CollectionCategory.SUITE,
                )
            ]
        )

        responses.add(
            responses.PUT,
            url,
            body=expected_response.model_dump_json(),
        )

        etag = '"87d154802e0803b953e5fee3ea0bb2a84b361b10730e68646b0bdab"'
        actual = self.debusine.collection_relation_set(
            workspace,
            source="personal@debian:suite",
            targets=targets,
            relation_type=CollectionRelationType.SUITE_REQUIRES,
            if_match=etag,
        )

        responses.assert_call_count(url, 1)

        self.assertEqual(actual, expected_response)

        self.assertEqual(
            responses.calls[0].request.headers["If-Match"],
            etag,
        )

    def add_get_download_artifact_response(
        self, files: int = 0, file_size: int = 10240, directories: bool = False
    ) -> ArtifactResponse:
        """
        Call responses.add("GET", "/api/1.0/artifact) with an Artifact.

        Return the ArtifactResponse that has been created.
        """
        artifact_id = 17
        artifact_base_url = (
            f"https://debusine.example.com/artifact/{artifact_id}/"
        )
        url_download_artifact = f"{artifact_base_url}?archive=tar.gz"

        directory = "directory/sub-directory/" if directories else ""

        response_files = {
            f"{directory}file{i}": FileResponse(
                size=file_size,
                checksums={"sha256": "abc123"},
                type="file",
                url=f"{artifact_base_url}{directory}file{i}",
            )
            for i in range(files)
        }

        # Set endpoint to download artifact's information
        url_artifact_information = f"{self.api_url}/1.0/artifact/{artifact_id}/"
        artifact_response = create_artifact_response(
            id=artifact_id,
            download_tar_gz_url=url_download_artifact,
            files=response_files,
        )
        responses.add(
            "GET",
            url_artifact_information,
            body=artifact_response.model_dump_json(),
        )

        return artifact_response

    @responses.activate
    def test_download_artifact_failed(self) -> None:
        """download_artifact() download an invalid tar file Raise exception."""
        artifact_response = self.add_get_download_artifact_response()
        responses.add(
            responses.GET,
            str(artifact_response.download_tar_gz_url),
            b"Invalid tar",
        )

        with self.assertRaisesRegex(
            RuntimeError, r"Error untarring https://.*\. Returncode: "
        ):
            self.debusine.download_artifact(
                artifact_response.id,
                self.create_temporary_directory(),
                tarball=False,
            )

    def create_tar(self) -> tuple[io.BytesIO, Path]:
        """
        Create a tar file with a file inside.

        :return: tuple [tarFile, Path of a file]
        """
        compressed = io.BytesIO()
        with tarfile.open(fileobj=compressed, mode="w|gz") as tar_file:
            file = self.create_temporary_file()
            tar_file.add(file, arcname=file.name)

        compressed.seek(0)

        return compressed, file

    @responses.activate
    def test_download_artifact_untar(self) -> None:
        """download_artifact() download the artifact and untar it."""
        compressed, file = self.create_tar()
        downloaded_directory = self.create_temporary_directory()
        expected_artifact_response = self.add_get_download_artifact_response()
        responses.add(
            responses.GET,
            str(expected_artifact_response.download_tar_gz_url),
            compressed.getvalue(),
        )

        stdout = io.StringIO()

        with contextlib.redirect_stdout(stdout):
            actual_artifact_response = self.debusine.download_artifact(
                expected_artifact_response.id,
                downloaded_directory,
                tarball=False,
            )

        self.assertEqual(
            file.read_bytes(),
            (downloaded_directory / file.name).read_bytes(),
        )

        self.assertEqual(actual_artifact_response, expected_artifact_response)

        self.logger.info.assert_has_calls(
            [
                call(
                    "Downloading artifact and uncompressing into %s",
                    downloaded_directory,
                ),
                call(file.name),
            ]
        )

    @responses.activate
    def test_download_artifact_tarball(self) -> None:
        """download_artifact() download the artifact as .tar.gz."""
        compressed, _ = self.create_tar()
        downloaded_directory = self.create_temporary_directory()

        expected_artifact_response = self.add_get_download_artifact_response()
        artifact_id = expected_artifact_response.id
        responses.add(
            responses.GET,
            str(expected_artifact_response.download_tar_gz_url),
            compressed.getvalue(),
        )

        self.debusine.download_artifact(
            expected_artifact_response.id,
            downloaded_directory,
            tarball=True,
        )

        artifact_file = downloaded_directory / f"artifact-{artifact_id}.tar.gz"

        self.assertEqual(compressed.getvalue(), artifact_file.read_bytes())

        self.logger.info.assert_called_with(
            "Artifact downloaded: %s", artifact_file
        )

    @responses.activate
    def test_download_artifact_single_file_individually(self) -> None:
        """download_artifact() downloads single-file artifacts individually."""
        contents = b"file contents"
        downloaded_directory = self.create_temporary_directory()
        artifact_response = self.add_get_download_artifact_response(files=1)
        destination = downloaded_directory / "file0"

        responses.add(
            responses.GET,
            str(artifact_response.files["file0"].url),
            contents,
        )

        self.debusine.download_artifact(
            artifact_response.id, downloaded_directory
        )

        self.assertEqual(destination.read_bytes(), contents)
        self.logger.info.assert_called_with(
            "Downloading artifact files into %s", downloaded_directory
        )

    @responses.activate
    def test_download_artifact_small_files_as_tar(self) -> None:
        """download_artifact() downloads many small files as a tar."""
        compressed, file = self.create_tar()
        downloaded_directory = self.create_temporary_directory()
        artifact_response = self.add_get_download_artifact_response(files=3)

        responses.add(
            responses.GET,
            str(artifact_response.download_tar_gz_url),
            compressed.getvalue(),
        )

        self.debusine.download_artifact(
            artifact_response.id, downloaded_directory
        )

        self.assertEqual(
            file.read_bytes(),
            (downloaded_directory / file.name).read_bytes(),
        )

    @responses.activate
    def test_download_artifact_single_file_with_parent_directory(self) -> None:
        """download_artifact() handles directories in single-file downloads."""
        contents = b"file contents"
        downloaded_directory = self.create_temporary_directory()
        artifact_response = self.add_get_download_artifact_response(
            files=1, directories=True
        )
        destination = downloaded_directory / "directory/sub-directory/file0"

        responses.add(
            responses.GET,
            str(artifact_response.files["directory/sub-directory/file0"].url),
            contents,
        )

        self.debusine.download_artifact(
            artifact_response.id, downloaded_directory
        )

        self.assertEqual(destination.read_bytes(), contents)
        self.logger.info.assert_called_with(
            "Downloading artifact files into %s", downloaded_directory
        )

    def test_should_download_as_tarball_single_file(self) -> None:
        """_should_download_as_tarball() single file artifact."""
        artifact_response = self.add_get_download_artifact_response(files=1)
        self.assertFalse(
            self.debusine._should_download_as_tarball(artifact_response)
        )

    def test_should_download_as_tarball_small_files(self) -> None:
        """When artifact contains only small files, download tar."""
        artifact_response = self.add_get_download_artifact_response(files=3)
        self.assertTrue(
            self.debusine._should_download_as_tarball(artifact_response)
        )

    def test_should_download_as_tarball_large_files(self) -> None:
        """When artifact contains a large file, download separately."""
        artifact_response = self.add_get_download_artifact_response(files=2)
        artifact_response.files["file1"].size = 30 * 1024 * 1024
        self.assertFalse(
            self.debusine._should_download_as_tarball(artifact_response)
        )

    def test_should_download_as_tarball_large_overall_size(self) -> None:
        """When artifact would be a large tar, download separately."""
        artifact_response = self.add_get_download_artifact_response(
            files=20, file_size=5 * 1024 * 1024
        )
        self.assertFalse(
            self.debusine._should_download_as_tarball(artifact_response)
        )

    def add_get_download_artifact_file_response(
        self, contents: bytes
    ) -> ArtifactResponse:
        """
        Call responses.add("GET", "/api/1.0/artifact) with an Artifact.

        Return the ArtifactResponse that has been created.
        """
        artifact_id = 17
        url_system_tar_xz = (
            f"https://debusine.example.com/artifact/{artifact_id}/system.tar.xz"
        )

        # Set endpoint to download artifact's information
        url_artifact_information = f"{self.api_url}/1.0/artifact/{artifact_id}/"
        artifact_response = create_artifact_response(
            id=artifact_id,
            files=FilesResponseType(
                {
                    "system.tar.xz": FileResponse(
                        size=len(contents),
                        checksums={
                            "sha256": hashlib.sha256(contents).hexdigest()
                        },
                        type="file",
                        url=url_system_tar_xz,
                    )
                }
            ),
        )
        responses.add(
            "GET",
            url_artifact_information,
            body=artifact_response.model_dump_json(),
        )

        return artifact_response

    @responses.activate
    def test_download_artifact_file_missing_path(self) -> None:
        """download_artifact_file() requires path_in_artifact to exist."""
        artifact_response = self.add_get_download_artifact_file_response(b"")
        with self.assertRaisesRegex(
            ValueError,
            f"No file 'nonexistent' in artifact {artifact_response.id}",
        ):
            self.debusine.download_artifact_file(
                artifact_response.id,
                "nonexistent",
                self.create_temporary_directory() / "nonexistent",
            )

    @responses.activate
    def test_download_artifact_file(self) -> None:
        """download_artifact_file() downloads the requested file."""
        contents = b"File contents"
        downloaded_directory = self.create_temporary_directory()
        artifact_response = self.add_get_download_artifact_file_response(
            contents
        )
        responses.add(
            responses.GET,
            str(artifact_response.files["system.tar.xz"].url),
            contents,
        )
        destination = downloaded_directory / "system.tar.xz"

        self.debusine.download_artifact_file(
            artifact_response.id, "system.tar.xz", destination
        )

        self.assertEqual(destination.read_bytes(), contents)
        self.logger.info.assert_called_with(
            "Artifact file downloaded: %s", destination
        )

    @responses.activate
    def test_download_artifact_file_with_artifact_response(self) -> None:
        """download_artifact_file() accepts an existing ArtifactResponse."""
        contents = b"File contents"
        downloaded_directory = self.create_temporary_directory()
        artifact_response = self.add_get_download_artifact_file_response(
            contents
        )
        responses.add(
            responses.GET,
            str(artifact_response.files["system.tar.xz"].url),
            contents,
        )
        destination = downloaded_directory / "system.tar.xz"

        self.debusine.download_artifact_file(
            artifact_response, "system.tar.xz", destination
        )

        self.assertEqual(destination.read_bytes(), contents)
        self.logger.info.assert_called_with(
            "Artifact file downloaded: %s", destination
        )

    @responses.activate
    def test_asset_create_success(self) -> None:
        """Test asset_create succeeds."""
        workspace = "test"
        work_request = 5

        asset_data = SigningKeyData(
            purpose=KeyPurpose.OPENPGP_UPLOAD,
            fingerprint="ABC123",
            public_key=base64.b64encode(b"PUBLIC KEY").decode(),
            description="Description",
        )
        asset_request = {
            "category": AssetCategory.SIGNING_KEY,
            "workspace": workspace,
            "data": asset_data.model_dump(),
            "work_request": work_request,
        }
        asset_response_json = {
            "id": 99,
            "category": AssetCategory.SIGNING_KEY,
            "workspace": workspace,
            "data": asset_data.model_dump(),
            "work_request": work_request,
        }
        asset_response = AssetResponse(
            id=99,
            category=AssetCategory.SIGNING_KEY,
            workspace=workspace,
            data=asset_data.model_dump(),
            work_request=work_request,
        )

        responses.add(
            responses.POST,
            f"{self.api_url}/1.0/asset/",
            json=asset_response_json,
            status=requests.codes.created,
        )

        actual_asset_response = self.debusine.asset_create(
            category=AssetCategory.SIGNING_KEY,
            workspace=workspace,
            data=asset_data,
            work_request=work_request,
        )

        self.assertEqual(actual_asset_response, asset_response)

        self.assert_token_key_included_in_all_requests(self.token)

        assert responses.calls[0].request.body is not None
        self.assertEqual(
            json.loads(responses.calls[0].request.body), asset_request
        )

    def assert_list_assets(
        self,
        url: str,
        asset_id: int | None = None,
        work_request: int | None = None,
        workspace: str | None = None,
    ) -> None:
        """Call client.asset_list with the specified options, assert results."""
        asset_data = SigningKeyData(
            purpose=KeyPurpose.OPENPGP_UPLOAD,
            fingerprint="ABC123",
            public_key=base64.b64encode(b"PUBLIC KEY").decode(),
            description="Description",
        )
        assets_response = AssetsResponse(
            [
                AssetResponse(
                    id=99,
                    category=AssetCategory.SIGNING_KEY,
                    workspace="workspace",
                    data=asset_data.model_dump(),
                    work_request=12,
                )
            ]
        )
        assets_response_json = [
            {
                "id": 99,
                "category": AssetCategory.SIGNING_KEY,
                "workspace": "workspace",
                "data": asset_data.model_dump(),
                "work_request": 12,
            }
        ]
        responses.add(
            responses.GET,
            url,
            json=assets_response_json,
            status=requests.codes.ok,
        )

        actual_asset_response = self.debusine.asset_list(
            asset_id=asset_id, workspace=workspace, work_request=work_request
        )

        self.assertEqual(actual_asset_response, assets_response)

        self.assert_token_key_included_in_all_requests(self.token)

    @responses.activate
    def test_asset_list_by_id(self) -> None:
        """Test asset_list by asset_id."""
        self.assert_list_assets(
            url=f"{self.api_url}/1.0/asset/?asset=99", asset_id=99
        )

    @responses.activate
    def test_asset_list_by_workspace(self) -> None:
        """Test asset_list by workspace."""
        self.assert_list_assets(
            url=f"{self.api_url}/1.0/asset/?workspace=workspace",
            workspace="workspace",
        )

    @responses.activate
    def test_asset_list_by_work_request(self) -> None:
        """Test asset_list by work_request_id."""
        self.assert_list_assets(
            url=f"{self.api_url}/1.0/asset/?work_request=12",
            work_request=12,
        )

    @responses.activate
    def test_asset_list_without_filter(self) -> None:
        """Test asset_list without a filter."""
        with self.assertRaisesRegex(
            ValueError,
            r"At least one of asset, work_request, and workspace must be "
            r"specified",
        ):
            self.debusine.asset_list()

    @responses.activate
    def test_asset_permission_check(self) -> None:
        permission_check_response = AssetPermissionCheckResponse(
            has_permission=True,
            username="testuser",
            user_id=123,
            resource={"package": "foo"},
        )
        responses.add(
            responses.POST,
            (
                f"{self.debusine.api_url}/asset/debusine%3Asigning-key/"
                f"openpgp%3AABC123/sign_with/"
            ),
            json=permission_check_response.model_dump(),
        )
        response = self.debusine.asset_permission_check(
            asset_category="debusine:signing-key",
            asset_slug="openpgp:ABC123",
            permission_name="sign_with",
            artifact_id=12,
            work_request_id=13,
            workspace="workspace",
        )
        self.assertEqual(response, permission_check_response)
        self.assert_token_key_included_in_all_requests(self.token)

    @responses.activate
    def test_lookup_single(self) -> None:
        """lookup_single() performs the requested lookup."""
        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/lookup/single/",
            json={
                "result_type": LookupResultType.ARTIFACT,
                "collection_item": None,
                "artifact": 1,
                "collection": None,
            },
        )

        response = self.debusine.lookup_single(
            lookup="collection/artifact",
            work_request=1,
            workspace="workspace",
            expect_type=LookupChildType.ARTIFACT,
            default_category=CollectionCategory.TEST,
        )

        self.assertEqual(
            response,
            LookupSingleResponseArtifact(artifact=1),
        )
        assert responses.calls[0].request.body is not None
        self.assertEqual(
            json.loads(responses.calls[0].request.body),
            {
                "lookup": "collection/artifact",
                "work_request": 1,
                "workspace": "workspace",
                "default_category": CollectionCategory.TEST,
                "expect_type": LookupChildType.ARTIFACT,
            },
        )

    @responses.activate
    def test_lookup_multiple(self) -> None:
        """lookup_multiple() performs the requested lookup."""
        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/lookup/multiple/",
            json=[
                {
                    "result_type": LookupResultType.ARTIFACT,
                    "collection_item": {
                        "name": "name",
                        "data": {"some": "data"},
                    },
                    "artifact": artifact_id,
                    "collection": None,
                    "parent_collection_lookup": "collection",
                }
                for artifact_id in (1, 2)
            ],
        )

        response = self.debusine.lookup_multiple(
            lookup=[{"collection": "collection"}],
            work_request=1,
            workspace="workspace",
            expect_type=LookupChildType.ARTIFACT,
            default_category=CollectionCategory.TEST,
        )

        self.assertEqual(
            response,
            LookupMultipleResponse(
                [
                    LookupSingleResponseArtifact(
                        collection_item=CollectionItemResponse(
                            name="name",
                            data={"some": "data"},
                        ),
                        artifact=artifact_id,
                        parent_collection_lookup="collection",
                    )
                    for artifact_id in (1, 2)
                ]
            ),
        )
        assert responses.calls[0].request.body is not None
        self.assertEqual(
            json.loads(responses.calls[0].request.body),
            {
                "lookup": [{"collection": "collection"}],
                "work_request": 1,
                "workspace": "workspace",
                "default_category": CollectionCategory.TEST,
                "expect_type": LookupChildType.ARTIFACT,
            },
        )

    @responses.activate
    def test_streaming_download(self) -> None:
        """Test _streaming_download() sets headers and returns the response."""
        url = pydantic.AnyUrl("https://example.net/stream")
        body = b"abc123"
        responses.add(responses.GET, str(url), body)
        r = self.debusine._streaming_download(url)

        self.assertEqual(r.content, body)
        self.assertEqual(r.request.headers["Token"], self.debusine.token)

    @responses.activate
    def test_fetch_task_configuration_collection_empty(self) -> None:
        """Querying /task-configuration/ returns an empty collection."""
        responses.add(
            responses.GET,
            f"{self.debusine.api_url}/task-configuration/System/default/",
            json={
                "collection": {"id": 1, "name": "default", "data": {}},
                "items": [],
            },
        )
        repository = self.debusine.fetch_task_configuration_collection(
            workspace="System", name="default"
        )
        self.assertEqual(repository.entries, [])

    @responses.activate
    def test_fetch_task_configuration_collection(self) -> None:
        config = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop", comment="noop"
        )
        template = DebusineTaskConfiguration(
            template="template", comment="template"
        )
        responses.add(
            responses.GET,
            f"{self.debusine.api_url}/task-configuration/System/default/",
            json={
                "collection": {"id": 1, "name": "default", "data": {}},
                "items": [config.model_dump(), template.model_dump()],
            },
        )
        repository = self.debusine.fetch_task_configuration_collection(
            workspace="System", name="default"
        )
        self.assertCountEqual(repository.entries, [config, template])

    @responses.activate
    def test_push_task_configuration_collection_without_manifest(self) -> None:
        repo = task_configuration.LocalTaskConfigurationRepository(
            self.create_temporary_directory()
        )
        with self.assertRaisesRegex(
            task_configuration.InvalidRepository, "repository has no manifest"
        ):
            self.debusine.push_task_configuration_collection(
                repo=repo, dry_run=False
            )

    @responses.activate
    def test_push_task_configuration_collection_empty(self) -> None:
        result = TaskConfigurationCollectionUpdateResults(
            added=1, removed=2, updated=3, unchanged=5
        )
        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/task-configuration/System/default/",
            json=result.model_dump(),
        )
        repo = task_configuration.LocalTaskConfigurationRepository(
            self.create_temporary_directory()
        )
        repo.manifest = repo.Manifest(
            workspace="System",
            collection=TaskConfigurationCollection(
                id=1, name="default", data={}
            ),
        )
        result = self.debusine.push_task_configuration_collection(
            repo=repo, dry_run=False
        )
        self.assertIsInstance(result, TaskConfigurationCollectionUpdateResults)
        self.assertEqual(result.added, 1)
        self.assertEqual(result.removed, 2)
        self.assertEqual(result.updated, 3)

    @responses.activate
    def test_push_task_configuration_collection(self) -> None:
        config = task_configuration.Item(
            task_type=TaskTypes.WORKER, task_name="noop", comment="noop"
        )
        template = task_configuration.Item(
            template="template", comment="template"
        )
        result = TaskConfigurationCollectionUpdateResults(
            added=2, removed=3, updated=4, unchanged=5
        )
        responses.add(
            responses.POST,
            f"{self.debusine.api_url}/task-configuration/System/default/",
            json=result.model_dump(),
        )
        repo = task_configuration.LocalTaskConfigurationRepository(
            self.create_temporary_directory()
        )
        repo.manifest = repo.Manifest(
            workspace="System",
            collection=TaskConfigurationCollection(
                id=1, name="default", data={}
            ),
        )
        ef = task_configuration.EntryFile(Path("config.yaml"))
        ef.entries.append(config)
        repo.files[ef.relpath] = ef
        ef = task_configuration.EntryFile(Path("template.yaml"))
        ef.entries.append(template)
        repo.files[ef.relpath] = ef
        result = self.debusine.push_task_configuration_collection(
            repo=repo, dry_run=False
        )
        self.assertIsInstance(result, TaskConfigurationCollectionUpdateResults)
        self.assertEqual(result.added, 2)
        self.assertEqual(result.removed, 3)
        self.assertEqual(result.updated, 4)

    @responses.activate
    def test_push_task_configuration_dry_run(self) -> None:
        url = f"{self.debusine.api_url}/task-configuration/System/default/"
        result = TaskConfigurationCollectionUpdateResults(
            added=2, removed=3, updated=4, unchanged=5
        )
        responses.add(responses.POST, url, json=result.model_dump())
        repo = task_configuration.LocalTaskConfigurationRepository(
            self.create_temporary_directory()
        )
        repo.manifest = repo.Manifest(
            workspace="System",
            collection=TaskConfigurationCollection(
                id=1, name="default", data={}
            ),
        )

        self.debusine.push_task_configuration_collection(
            repo=repo, dry_run=False
        )
        self.debusine.push_task_configuration_collection(
            repo=repo, dry_run=True
        )

        responses.assert_call_count(url, 2)
        assert responses.calls[0].request.body is not None
        self.assertFalse(json.loads(responses.calls[0].request.body)["dry_run"])
        assert responses.calls[1].request.body is not None
        self.assertTrue(json.loads(responses.calls[1].request.body)["dry_run"])

    @responses.activate
    def test_get_workspace_inheritance(self) -> None:
        url = f"{self.debusine.api_url}/workspace/System/inheritance/"
        chain = [
            WorkspaceInheritanceChainElement(
                id=3, scope="scope1", workspace="workspace1"
            ),
            WorkspaceInheritanceChainElement(
                id=4, scope="scope2", workspace="workspace2"
            ),
        ]
        result = WorkspaceInheritanceChain(chain=chain)
        responses.add(responses.GET, url, json=result.model_dump())
        res = self.debusine.get_workspace_inheritance("System")
        self.assertEqual(res.chain, chain)
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_set_workspace_inheritance(self) -> None:
        url = f"{self.debusine.api_url}/workspace/System/inheritance/"
        result = WorkspaceInheritanceChain(chain=[])
        responses.add(responses.POST, url, json=result.model_dump())
        res = self.debusine.set_workspace_inheritance(
            "System", chain=WorkspaceInheritanceChain(chain=[])
        )
        self.assertEqual(res.chain, [])
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_get(self) -> None:
        url = f"{self.debusine.api_url}/collection/System/debian%3Asuite/sid/"
        result = CollectionData(
            id=42,
            name="sid",
            category=CollectionCategory.SUITE,
            full_history_retention_period=None,
            metadata_only_retention_period=None,
            data={"test": True},
        )
        responses.add(responses.GET, url, json=result.model_dump())
        res = self.debusine.collection_get("System", "debian:suite", "sid")
        self.assertEqual(res, result)
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_get_timedelta(self) -> None:
        url = f"{self.debusine.api_url}/collection/System/debian%3Asuite/sid/"
        result = CollectionData(
            id=42,
            name="sid",
            category=CollectionCategory.SUITE,
            full_history_retention_period=3,
            metadata_only_retention_period=4,
            data={"test": True},
        )
        responses.add(responses.GET, url, json=result.model_dump())
        res = self.debusine.collection_get(
            "System", result.category, result.name
        )
        self.assertEqual(res, result)
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_get_by_id(self) -> None:
        url = f"{self.debusine.api_url}/collection/System/42/"
        result = CollectionData(
            id=42,
            name="sid",
            category=CollectionCategory.SUITE,
            full_history_retention_period=3,
            metadata_only_retention_period=4,
            data={"test": True},
        )
        responses.add(responses.GET, url, json=result.model_dump())
        res = self.debusine.collection_get_by_id("System", 42)
        self.assertEqual(res, result)
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_create(self) -> None:
        url = f"{self.debusine.api_url}/collection/System/"
        data = CollectionDataNew(
            name="sid",
            category=CollectionCategory.SUITE,
            full_history_retention_period=7,
            metadata_only_retention_period=14,
            data={"test": 2},
        )
        result = CollectionData(id=41, **data.model_dump())
        responses.add(
            responses.POST,
            url,
            json=result.model_dump(),
            status=requests.codes.created,
        )
        res = self.debusine.collection_create("System", data)
        self.assertEqual(res, result)
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_update(self) -> None:
        url = f"{self.debusine.api_url}/collection/System/42/"
        result = CollectionData(
            id=42,
            name="sid",
            category=CollectionCategory.SUITE,
            full_history_retention_period=7,
            metadata_only_retention_period=14,
            data={"test": 2},
        )
        responses.add(responses.PATCH, url, json=result.model_dump())
        res = self.debusine.collection_update("System", result)
        self.assertEqual(res, result)
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_iter(self) -> None:
        url = f"{self.debusine.api_url}/collection/System/"
        data = CollectionData(
            id=42,
            name="sid",
            category=CollectionCategory.SUITE,
            full_history_retention_period=None,
            metadata_only_retention_period=None,
            data={"test": True},
        )
        listing_response = create_listing_response(data)
        responses.add(
            responses.GET, url, json=listing_response.model_dump(mode="json")
        )
        res = list(self.debusine.collection_iter("System"))
        self.assertEqual(res, [data])
        responses.assert_call_count(url, 1)

    @responses.activate
    def test_collection_item_get(self) -> None:
        url = (
            f"{self.debusine.api_url}/collection/System/debian%3Asuite/sid"
            f"/item/hello_1.0/"
        )
        result = CollectionItemData(
            name="hello_1.0",
            category=ArtifactCategory.SOURCE_PACKAGE,
            data=DebianSourcePackage(
                name="hello", version="1.0", type="dpkg", dsc_fields={}
            ).model_dump(),
        )
        responses.add(responses.GET, url, json=result.model_dump())
        res = self.debusine.collection_item_get(
            workspace="System",
            collection_category=CollectionCategory.SUITE,
            collection_name="sid",
            name="hello_1.0",
        )
        self.assertEqual(res, result)
        responses.assert_call_count(url, 1)
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_collection_item_delete(self) -> None:
        url = (
            f"{self.debusine.api_url}/collection/System/debian%3Asuite/sid"
            f"/item/hello_1.0/"
        )
        responses.add(responses.DELETE, url, status=requests.codes.no_content)
        self.debusine.collection_item_delete(
            workspace="System",
            collection_category=CollectionCategory.SUITE,
            collection_name="sid",
            name="hello_1.0",
        )
        responses.assert_call_count(url, 1)
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_collection_item_iter(self) -> None:
        for kwargs, url_suffix in (
            ({}, ""),
            ({"name__startswith": "hello"}, "?name__startswith=hello"),
            (
                {"category": ArtifactCategory.SOURCE_PACKAGE},
                "?category=debian%3Asource-package",
            ),
        ):
            with (
                self.subTest(kwargs=str(kwargs)),
                responses.RequestsMock() as rsps,
            ):
                url = (
                    f"{self.debusine.api_url}/collection/System/debian%3Asuite"
                    f"/sid/item/"
                ) + url_suffix
                data = CollectionItemData(
                    name="hello_1.0",
                    category=ArtifactCategory.SOURCE_PACKAGE,
                    data=DebianSourcePackage(
                        name="hello", version="1.0", type="dpkg", dsc_fields={}
                    ).model_dump(),
                )
                listing_response = create_listing_response(data)
                rsps.add(
                    responses.GET,
                    url,
                    json=listing_response.model_dump(mode="json"),
                )
                res = list(
                    self.debusine.collection_item_iter(
                        workspace="System",
                        collection_category=CollectionCategory.SUITE,
                        collection_name="sid",
                        **kwargs,
                    )
                )
                self.assertEqual(res, [data])
                rsps.assert_call_count(url, 1)
                self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_collection_token_grant_iter(self) -> None:
        url = (
            f"{self.debusine.api_url}/collection/System/debusine%3Atest/test"
            f"/token/"
        )
        grants = [
            CollectionTokenGrantResponse(
                token_type="external-user",
                token_hash=token_hash,
                username=username,
                comment="a comment",
                role=CollectionRoles.VIEWER,
            )
            for token_hash, username in (
                ("0" * 64, "test1"),
                ("1" * 64, "test2"),
            )
        ]
        listing_response = create_listing_response(*grants)

        responses.add(
            responses.GET, url, json=listing_response.model_dump(mode="json")
        )

        self.assertEqual(
            list(
                self.debusine.collection_token_grant_iter(
                    workspace="System",
                    collection_category=CollectionCategory.TEST,
                    collection_name="test",
                )
            ),
            grants,
        )
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_collection_token_grant_create(self) -> None:
        url = (
            f"{self.debusine.api_url}/collection/System/debusine%3Atest/test"
            f"/token/"
        )
        expected_grant_response = CollectionTokenGrantResponse(
            token_type="external-user",
            token_hash="0" * 64,
            token_key="9" * 64,
            username="test",
            comment="a comment",
            role=CollectionRoles.VIEWER,
        )
        responses.add(
            responses.POST,
            url,
            status=requests.codes.created,
            json=expected_grant_response.model_dump(mode="json"),
        )
        request = CollectionTokenGrantRequest(
            username="test",
            comment="a comment",
            role=CollectionRoles.VIEWER,
        )

        actual_grant_response = self.debusine.collection_token_grant_create(
            workspace="System",
            collection_category=CollectionCategory.TEST,
            collection_name="test",
            request=request,
        )
        self.assertEqual(actual_grant_response, expected_grant_response)
        self.assert_token_key_included_in_all_requests("token")
        self.assertEqual(
            json.loads(responses.calls[0].request.body or "null"),
            request.model_dump(exclude_unset=True),
        )

    @responses.activate
    def test_collection_token_grant_delete(self) -> None:
        url = (
            f"{self.debusine.api_url}/collection/System/debusine%3Atest/test"
            f"/token/{'0' * 64}/viewer/"
        )

        responses.add(responses.DELETE, url, status=requests.codes.no_content)

        self.debusine.collection_token_grant_delete(
            workspace="System",
            collection_category=CollectionCategory.TEST,
            collection_name="test",
            token_hash="0" * 64,
            role=CollectionRoles.VIEWER,
        )
        self.assert_token_key_included_in_all_requests("token")

    @responses.activate
    def test_emit_metric(self) -> None:
        url = f"{self.debusine.api_url}/open-metrics/emit/"
        responses.add(responses.POST, url, status=requests.codes.no_content)

        self.debusine.emit_metric(
            metric_type="counter",
            name="debusine_test",
            labels={"foo": "bar"},
            value=1,
        )

        responses.assert_call_count(url, 1)
        assert responses.calls[0].request.body is not None
        self.assertEqual(
            json.loads(responses.calls[0].request.body),
            {
                "metric_type": "counter",
                "name": "debusine_test",
                "labels": {"foo": "bar"},
                "value": 1,
            },
        )


class DebusineAsyncTests(IsolatedAsyncioTestCase, TestCase):
    """
    Tests for Debusine client asynchronous functionality.

    Use server.DebusineAioHTTPTestCase to mock the server (instead of the
    responses library used by DebusineTests).
    """

    WORK_REQUEST_RESULT = "success"
    COMPLETED_AT = dt.datetime.now(dt.UTC)

    @override
    def setUp(self) -> None:
        super().setUp()
        self.api_url = "https://debusine.example.org/api"
        self.token = "token"
        self.logger = mock.create_autospec(spec=logging.Logger)
        self.debusine = Debusine(self.api_url, self.token, logger=self.logger)

    def on_complete_endpoint(
        self, *, params: dict[str, str] | None = None
    ) -> str:
        """Return the WebSocket on-completed-url endpoint."""
        url = "1.0/work-request/on-completed/"
        if params is not None:
            url += "?" + urlencode(params)
        return url

    def mock_server_notifications(
        self, payloads: list[dict[str, Any]]
    ) -> mock.MagicMock:
        # Mock the server notification reader
        async def mock_messages() -> AsyncGenerator[dict[str, Any]]:
            for payload in payloads:
                yield payload

        server_notifications = mock.MagicMock(spec=ServerNotifications)
        server_notifications.__aenter__.return_value = server_notifications
        server_notifications.__aexit__.return_value = None
        server_notifications.messages = mock_messages
        return self.enterContext(
            mock.patch(
                "debusine.client.server_notifications.ServerNotifications",
                return_value=server_notifications,
            )
        )

    def test_on_work_request_completed_calls_async(self) -> None:
        command = ["this-is-the-command"]

        with mock.patch.object(
            self.debusine, "on_work_request_completed_async"
        ) as on_work_request_completed_async:
            self.debusine.on_work_request_completed(
                workspaces=None,
                command=command,
                working_directory=Path.cwd(),
                last_completed_at=None,
            )

        on_work_request_completed_async.assert_awaited_with(
            workspaces=None,
            command=command,
            working_directory=Path.cwd(),
            last_completed_at=None,
        )

    async def test_on_work_request_completed_with_workspaces(self) -> None:
        workspace = "lts"
        command = ["this-is-the-command"]

        server_notifications = self.mock_server_notifications([])

        await self.debusine.on_work_request_completed_async(
            workspaces=[workspace],
            last_completed_at=None,
            command=command,
            working_directory=Path.cwd(),
        )

        server_notifications.assert_called_with(
            self.debusine,
            endpoint=self.on_complete_endpoint(
                params={"workspaces": workspace}
            ),
        )

    async def test_on_work_request_completed_last_completed_at(self) -> None:
        """
        on_work_request_completed call _wait_and_execute with last_completed.

        The file in last_completed_at existed and its contents is read.
        """
        command = ["this-is-the-command"]
        last_completed_at = dt.datetime.now(dt.UTC).isoformat()

        last_completed_at_file = self.create_temporary_file(
            contents=json.dumps(
                {"last_completed_at": last_completed_at}
            ).encode("utf-8")
        )

        server_notifications = self.mock_server_notifications([])

        await self.debusine.on_work_request_completed_async(
            last_completed_at=last_completed_at_file,
            command=command,
            working_directory=Path.cwd(),
        )

        server_notifications.assert_called_with(
            self.debusine,
            endpoint=self.on_complete_endpoint(
                params={"completed_at_since": last_completed_at}
            ),
        )

    async def test_on_work_request_completed_last_completed_at_does_not_exist(
        self,
    ) -> None:
        """
        on_work_request_completed call _wait_and_execute with last_completed.

        The file in last_completed_at did not exist.
        """
        last_completed_at_file = self.create_temporary_file()
        last_completed_at_file.unlink()

        command = ["this-is-the-command"]

        server_notifications = self.mock_server_notifications([])

        await self.debusine.on_work_request_completed_async(
            last_completed_at=last_completed_at_file,
            command=command,
            working_directory=Path.cwd(),
        )

        server_notifications.assert_called_with(
            self.debusine,
            endpoint=self.on_complete_endpoint(),
        )

    async def test_on_work_request_completed_invalid_text(
        self,
    ) -> None:
        _on_work_request_completed = self.enterContext(
            mock.patch.object(self.debusine, "_on_work_request_completed")
        )
        self.mock_server_notifications([{}])
        await self.debusine.on_work_request_completed_async(
            command=["/bin/true"], working_directory=Path.cwd()
        )
        _on_work_request_completed.assert_not_called()
        self.logger.warning.assert_called_once_with(
            "invalid text in msg_content received from the server: %r ", None
        )

    async def test_on_work_request_completed_connected(
        self,
    ) -> None:
        _on_work_request_completed = self.enterContext(
            mock.patch.object(self.debusine, "_on_work_request_completed")
        )
        self.mock_server_notifications([{"text": "connected"}])
        await self.debusine.on_work_request_completed_async(
            command=["/bin/true"], working_directory=Path.cwd()
        )
        _on_work_request_completed.assert_not_called()
        self.logger.info.assert_called_once_with("Connected!")

    async def test_on_work_request_completed_work_request_completed(
        self,
    ) -> None:
        self.enterContext(
            mock.patch.object(self.debusine, "_read_last_completed_at")
        )
        _on_work_request_completed = self.enterContext(
            mock.patch.object(self.debusine, "_on_work_request_completed")
        )
        payload = OnWorkRequestCompleted(
            work_request_id=23,
            result=self.WORK_REQUEST_RESULT,
            completed_at=self.COMPLETED_AT,
        ).model_dump(mode="json")
        self.mock_server_notifications(
            [{"text": "work_request_completed", **payload}]
        )
        await self.debusine.on_work_request_completed_async(
            command=[mock.sentinel.command],
            working_directory=mock.sentinel.working_directory,
            last_completed_at=mock.sentinel.last_completed_at,
        )
        _on_work_request_completed.assert_called_once_with(
            command=[mock.sentinel.command],
            working_directory=mock.sentinel.working_directory,
            last_completed_at=mock.sentinel.last_completed_at,
            msg_content={"text": "work_request_completed", **payload},
        )

    async def test_on_work_request_completed_work_request_completed_twice(
        self,
    ) -> None:
        self.enterContext(
            mock.patch.object(self.debusine, "_read_last_completed_at")
        )
        _on_work_request_completed = self.enterContext(
            mock.patch.object(self.debusine, "_on_work_request_completed")
        )
        payload = OnWorkRequestCompleted(
            work_request_id=23,
            result=self.WORK_REQUEST_RESULT,
            completed_at=self.COMPLETED_AT,
        ).model_dump(mode="json")
        self.mock_server_notifications(
            [
                payload1 := {
                    "text": "work_request_completed",
                    **payload,
                    "work_request_id": 1,
                },
                payload2 := {
                    "text": "work_request_completed",
                    **payload,
                    "work_request_id": 2,
                },
            ]
        )
        await self.debusine.on_work_request_completed_async(
            command=[mock.sentinel.command],
            working_directory=mock.sentinel.working_directory,
            last_completed_at=mock.sentinel.last_completed_at,
        )
        _on_work_request_completed.assert_has_calls(
            [
                call(
                    command=[mock.sentinel.command],
                    working_directory=mock.sentinel.working_directory,
                    last_completed_at=mock.sentinel.last_completed_at,
                    msg_content=payload1,
                ),
                call(
                    command=[mock.sentinel.command],
                    working_directory=mock.sentinel.working_directory,
                    last_completed_at=mock.sentinel.last_completed_at,
                    msg_content=payload2,
                ),
            ]
        )

    def test__on_work_request_completed(self) -> None:
        completed_at = dt.datetime(2026, 1, 2, tzinfo=dt.UTC)
        popen = self.enterContext(mock.patch("subprocess.Popen"))
        write_last_completed_at = self.enterContext(
            mock.patch.object(self.debusine, "write_last_completed_at")
        )
        self.debusine._on_work_request_completed(
            command=[mock.sentinel.command],
            working_directory=mock.sentinel.working_directory,
            last_completed_at=mock.sentinel.last_completed_at,
            msg_content=OnWorkRequestCompleted(
                work_request_id=12,
                result=WorkRequestResults.SUCCESS,
                completed_at=completed_at,
            ).model_dump(),
        )
        popen.assert_called_once_with(
            [mock.sentinel.command, "12", "success"],
            cwd=mock.sentinel.working_directory,
        )
        write_last_completed_at.assert_called_once_with(
            mock.sentinel.last_completed_at, completed_at
        )

    def test__on_work_request_completed_no_last_completed_at(self) -> None:
        completed_at = dt.datetime(2026, 1, 2, tzinfo=dt.UTC)
        popen = self.enterContext(mock.patch("subprocess.Popen"))
        write_last_completed_at = self.enterContext(
            mock.patch.object(self.debusine, "write_last_completed_at")
        )
        self.debusine._on_work_request_completed(
            command=[mock.sentinel.command],
            working_directory=mock.sentinel.working_directory,
            last_completed_at=None,
            msg_content=OnWorkRequestCompleted(
                work_request_id=12,
                result=WorkRequestResults.SUCCESS,
                completed_at=completed_at,
            ).model_dump(),
        )
        popen.assert_called_once_with(
            [mock.sentinel.command, "12", "success"],
            cwd=mock.sentinel.working_directory,
        )
        write_last_completed_at.assert_not_called()

    async def test_write_last_completed_at(self) -> None:
        last_completed_at = self.create_temporary_file()
        self.debusine.write_last_completed_at(
            last_completed_at, self.COMPLETED_AT
        )
        expected = (
            json.dumps(
                {"last_completed_at": self.COMPLETED_AT.isoformat()}, indent=2
            )
            + "\n"
        )

        self.assertEqual(last_completed_at.read_text(), expected)

    async def test_decode_ws_message_success(self) -> None:
        msg = aiohttp.WSMessage(
            aiohttp.WSMsgType.TEXT, '{"text": "foo", "value": 1}', None
        )
        payload = self.debusine.decode_ws_message(msg)
        self.assertEqual(payload, {"text": "foo", "value": 1})

    async def test_decode_ws_message_invalid_msg_type(self) -> None:
        with self.assertRaisesRegex(
            TypeError,
            "is not an instance of aiohttp.http_websocket.WSMessage",
        ):
            self.debusine.decode_ws_message(
                "Something not an instance of WMessage"  # type: ignore[arg-type]
            )

    async def test_decode_ws_message_non_text_type(self) -> None:
        msg = aiohttp.WSMessage(aiohttp.WSMsgType.PING, None, None)
        with self.assertRaisesRegex(
            ServerNotificationDecodeError, "Message is not a TEXT message"
        ):
            self.debusine.decode_ws_message(msg)

    async def test_decode_ws_message_invalid_json(self) -> None:
        msg = aiohttp.WSMessage(aiohttp.WSMsgType.TEXT, '{[', None)
        with self.assertRaisesRegex(
            ServerNotificationDecodeError, "does not decode as JSON"
        ):
            self.debusine.decode_ws_message(msg)

    async def test_decode_ws_message_json_not_a_dict(self) -> None:
        msg = aiohttp.WSMessage(aiohttp.WSMsgType.TEXT, '[]', None)
        with self.assertRaisesRegex(
            ServerNotificationDecodeError, "does not contain a dict"
        ):
            self.debusine.decode_ws_message(msg)

    async def test_decode_ws_message_raise_token_disabled_error_disabled(
        self,
    ) -> None:
        msg = aiohttp.WSMessage(
            aiohttp.WSMsgType.TEXT,
            '{"reason": "Disabled token", "reason_code": "TOKEN_DISABLED"}',
            None,
        )
        with self.assertRaisesRegex(TokenDisabledError, "Disabled token"):
            self.debusine.decode_ws_message(msg)

    async def test_decode_ws_message_raise_token_disabled_error_unknown(
        self,
    ) -> None:
        msg = aiohttp.WSMessage(
            aiohttp.WSMsgType.TEXT,
            '{"reason_code": "TOKEN_DISABLED"}',
            None,
        )
        with self.assertRaisesRegex(TokenDisabledError, "Unknown"):
            self.debusine.decode_ws_message(msg)

    async def test_decode_ws_message_unsupported_dict_contents(self) -> None:
        msg = aiohttp.WSMessage(aiohttp.WSMsgType.TEXT, '{"foo": "bar"}', None)
        with self.assertRaisesRegex(
            ServerNotificationDecodeError, "Unsupported message payload"
        ):
            self.debusine.decode_ws_message(msg)

    async def test_decode_ws_message_unsupported_error_reason(self) -> None:
        msg = aiohttp.WSMessage(
            aiohttp.WSMsgType.TEXT,
            '{"reason": "Disabled token",'
            '"reason_code": "SOMETHING_NOT_IMPLEMENTED"}',
            None,
        )
        with self.assertRaisesRegex(
            ServerNotificationDecodeError, "Unsupported error reason code"
        ):
            self.debusine.decode_ws_message(msg)
