# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Tests for the debusine Cli WorkRequest commands."""

import contextlib
import datetime as dt
import json
import re
import stat
from collections.abc import Generator
from functools import partial
from pathlib import Path
from typing import Any, override
from unittest import mock
from unittest.mock import MagicMock

import responses
import yaml

from debusine.artifacts.models import ArtifactCategory, WorkRequestResults
from debusine.client.cli import Cli
from debusine.client.commands.base import Command
from debusine.client.commands.tests.base import BaseCliTests
from debusine.client.commands.work_requests import WorkRequestModelCommand
from debusine.client.debusine import Debusine
from debusine.client.exceptions import DebusineError
from debusine.client.models import (
    ArtifactResponse,
    WorkRequestExternalDebsignRequest,
    WorkRequestRequest,
    WorkRequestResponse,
)
from debusine.test.test_utils import (
    create_artifact_response,
    create_file_response,
    create_remote_artifact,
    create_work_request_response,
)
from debusine.utils import calculate_hash


class WorkRequestModelCommandTests(BaseCliTests):
    """Tests for :py:class:`WorkRequestModelCommand`."""

    @Command.preserve_registry()
    def _command(self) -> WorkRequestModelCommand:
        class ConcreteWorkRequestModelCommand(WorkRequestModelCommand):
            """Concrete version of WorkRequestModelCommand."""

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

        return self.command(
            ConcreteWorkRequestModelCommand, workspace="workspace"
        )

    def test_list_rich(self) -> None:
        wr = create_work_request_response()
        command = self._command()
        webui_url = command.debusine.webui_url(wr.get_link_url("webui_self"))
        with (
            mock.patch(
                "debusine.client.commands.work_requests.Table",
                return_value=(table := mock.MagicMock()),
            ),
            mock.patch(
                "debusine.client.commands.work_requests.rich.print"
            ) as rich_print,
            mock.patch("rich.json.JSON.from_data", side_effect=lambda x: x),
        ):
            stderr, stdout = self.capture_output(
                partial(command._list_rich, [wr])
            )

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        rich_print.assert_called_once()

        self.assertEqual(table.add_row.call_count, 1)
        call = table.add_row.call_args_list[0]
        self.assertEqual(call.args[0], f"[link={webui_url}]#{wr.id}[/]")
        self.assertEqual(call.args[1], f"{wr.task_type} {wr.task_name}")
        self.assertEqual(call.args[2], wr.status)
        self.assertEqual(
            call.args[3], f"Created: {command._format_timestamp(wr.created_at)}"
        )
        self.assertEqual(call.args[4], command._format_int(wr.worker))
        self.assertEqual(call.args[5], wr.task_data)

    def test_list_rich_full_data(self) -> None:
        wr = create_work_request_response()
        wr.started_at = wr.created_at + dt.timedelta(minutes=2)
        wr.completed_at = wr.started_at + dt.timedelta(minutes=3)
        wr.result = WorkRequestResults.SUCCESS
        command = self._command()
        webui_url = command.debusine.webui_url(wr.get_link_url("webui_self"))
        with (
            mock.patch(
                "debusine.client.commands.work_requests.Table",
                return_value=(table := mock.MagicMock()),
            ),
            mock.patch(
                "debusine.client.commands.work_requests.rich.print"
            ) as rich_print,
            mock.patch("rich.json.JSON.from_data", side_effect=lambda x: x),
        ):
            stderr, stdout = self.capture_output(
                partial(command._list_rich, [wr])
            )

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        rich_print.assert_called_once()

        self.assertEqual(table.add_row.call_count, 1)
        call = table.add_row.call_args_list[0]
        self.assertEqual(call.args[0], f"[link={webui_url}]#{wr.id}[/]")
        self.assertEqual(call.args[1], f"{wr.task_type} {wr.task_name}")
        self.assertEqual(call.args[2], f"{wr.status} {wr.result}")
        self.assertEqual(
            call.args[3],
            "\n".join(
                [
                    f"Created: {command._format_timestamp(wr.created_at)}",
                    f"Started: {command._format_timestamp(wr.started_at)}",
                    f"Completed: {command._format_timestamp(wr.completed_at)}",
                ]
            ),
        )
        self.assertEqual(call.args[4], command._format_int(wr.worker))
        self.assertEqual(call.args[5], wr.task_data)

    def test_format_timestamp(self) -> None:
        command = self._command()
        for value, expected in (
            (None, None),
            (dt.datetime(2026, 1, 1, tzinfo=dt.UTC), "2026-01-01 00:00:00"),
            (
                dt.datetime(2026, 1, 1, microsecond=42, tzinfo=dt.UTC),
                "2026-01-01 00:00:00",
            ),
        ):
            with self.subTest(value=repr(value)):
                self.assertEqual(command._format_timestamp(value), expected)

    def test_format_int(self) -> None:
        command = self._command()
        for value, expected in ((None, None), (0, "0"), (42, "42")):
            with self.subTest(value=value):
                self.assertEqual(command._format_int(value), expected)

    def test_show_rich(self) -> None:
        wr = create_work_request_response()
        command = self._command()
        webui_url = command.debusine.webui_url(wr.get_link_url("webui_self"))
        with (
            mock.patch(
                "debusine.client.commands.work_requests.Table",
                return_value=(table := mock.MagicMock()),
            ),
            mock.patch(
                "debusine.client.commands.work_requests.rich.print"
            ) as rich_print,
            mock.patch("rich.json.JSON.from_data", side_effect=lambda x: x),
        ):
            stderr, stdout = self.capture_output(
                partial(command._show_rich, wr)
            )

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        rich_print.assert_called_once()

        self.assertEqual(table.add_row.call_count, 6)
        rows = [
            call_args_list.args
            for call_args_list in table.add_row.call_args_list
        ]
        self.assertEqual(
            rows[0], ("ID:", f"#{wr.id} [link={webui_url}]{webui_url}[/]")
        )
        self.assertEqual(rows[1], ("Task:", f"{wr.task_type} {wr.task_name}"))
        self.assertEqual(rows[2], ("Workspace:", f"{wr.scope}/{wr.workspace}"))
        self.assertEqual(
            rows[3], ("Created at:", command._format_timestamp(wr.created_at))
        )
        self.assertEqual(rows[4], ("Status:", wr.status))
        self.assertEqual(
            rows[5],
            ("Task data:", wr.task_data),
        )

    def test_show_rich_full_data(self) -> None:
        wr = create_work_request_response()
        wr.started_at = wr.created_at + dt.timedelta(minutes=2)
        wr.completed_at = wr.started_at + dt.timedelta(minutes=3)
        wr.duration = 5
        wr.result = WorkRequestResults.SUCCESS
        wr.worker = 42
        wr.priority_base = 2
        wr.priority_adjustment = 3
        wr.dynamic_task_data = {"dynamic": True}
        wr.artifacts = [1, 2, 3]
        command = self._command()
        webui_url = command.debusine.webui_url(wr.get_link_url("webui_self"))
        with (
            mock.patch(
                "debusine.client.commands.work_requests.Table",
                return_value=(table := mock.MagicMock()),
            ),
            mock.patch(
                "debusine.client.commands.work_requests.rich.print"
            ) as rich_print,
            mock.patch("rich.json.JSON.from_data", side_effect=lambda x: x),
        ):
            stderr, stdout = self.capture_output(
                partial(command._show_rich, wr)
            )

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        rich_print.assert_called_once()

        self.assertEqual(table.add_row.call_count, 13)
        rows = [
            call_args_list.args
            for call_args_list in table.add_row.call_args_list
        ]
        self.assertEqual(
            rows[0], ("ID:", f"#{wr.id} [link={webui_url}]{webui_url}[/]")
        )
        self.assertEqual(rows[1], ("Task:", f"{wr.task_type} {wr.task_name}"))
        self.assertEqual(rows[2], ("Workspace:", f"{wr.scope}/{wr.workspace}"))
        self.assertEqual(
            rows[3], ("Created at:", command._format_timestamp(wr.created_at))
        )
        self.assertEqual(
            rows[4], ("Started at:", command._format_timestamp(wr.started_at))
        )
        self.assertEqual(
            rows[5],
            ("Completed at:", command._format_timestamp(wr.completed_at)),
        )
        self.assertEqual(
            rows[6],
            ("Duration:", command._format_float(wr.duration)),
        )
        self.assertEqual(rows[7], ("Status:", f"{wr.status} {wr.result}"))
        self.assertEqual(
            rows[8],
            ("Worker:", command._format_int(wr.worker)),
        )
        self.assertEqual(rows[9], ("Priority:", "2 + 3"))
        self.assertEqual(
            rows[10],
            ("Task data:", wr.task_data),
        )
        self.assertEqual(
            rows[11],
            ("Dynamic task data:", wr.dynamic_task_data),
        )
        self.assertEqual(
            rows[12],
            ("Artifacts:", wr.artifacts),
        )


class CliCreateWorkRequestTests(BaseCliTests):
    """Tests for the Cli functionality related to create work requests."""

    def patch_work_request_create(self) -> MagicMock:
        """
        Patch Debusine.work_request_create.

        Does not set return_value / side_effect.
        """
        patcher_work_request_create = mock.patch.object(
            Debusine, 'work_request_create', autospec=True
        )
        mocked_work_request_create = patcher_work_request_create.start()
        self.addCleanup(patcher_work_request_create.stop)

        return mocked_work_request_create

    def test_create_work_request_invalid_task_name(self) -> None:
        """Cli parse the CLI and stdin to create a WorkRequest bad task_name."""
        mocked_work_request_create = self.patch_work_request_create()
        task_name = "task-name"
        mocked_work_request_create.side_effect = DebusineError(
            title=f"invalid {task_name}"
        )

        self.enterContext(
            self.patch_sys_stdin_read(
                "distribution: stretch\n"
                "package: http://..../package_1.2-3.dsc\n"
            )
        )

        cli = self.create_cli(["work-request", "create", task_name])

        exception = self.assertShowsError(cli.execute)
        self.assertDebusineError(exception, {"title": f"invalid {task_name}"})

    def assert_create_work_request(
        self,
        cli: Cli,
        mocked_post_work_request: MagicMock,
        work_request_request: WorkRequestRequest,
    ) -> None:
        """
        Call cli.execute() and assert the work request is created.

        Assert expected stderr/stdout and network call.
        """
        with (
            mock.patch(
                "debusine.client.commands.base.Command.feedback"
            ) as feedback,
            mock.patch(
                "debusine.client.commands.work_requests"
                ".WorkRequestModelCommand.show"
            ) as show,
        ):
            stderr, stdout = self.capture_output(cli.execute)

        work_request_id = mocked_post_work_request.return_value.id

        feedback.assert_called_once_with("Work request started:")
        show.assert_called_once()
        self.assertEqual(show.call_args.args[0].id, work_request_id)

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")

        actual_work_request_request = mocked_post_work_request.mock_calls[0][1][
            1
        ]

        # Verify WorkRequestRequest is as expected
        assert actual_work_request_request == work_request_request

    def verify_create_work_request_scenario(
        self,
        cli_args: list[str],
        workspace: str = "developers",
        event_reactions: dict[str, Any] = {},
        use_stdin: bool = False,
    ) -> None:
        """
        Test create-work-request using workspace/stdin/stdin-data.

        :param cli_args: first parameter is task_name. create-work-request
          is prepended to them.
        :param workspace: workspace to use or None if not specified in the
          cli_args
        :param use_stdin: to pass the generated data: use_stdin or a file.
        """
        task_name = cli_args[0]

        mocked_post_work_request = self.patch_work_request_create()
        work_request_id = 11
        data = {
            "distribution": "stretch",
            "package": "http://..../package_1.2-3.dsc",
        }
        mocked_post_work_request.return_value = create_work_request_response(
            base_url=self.get_base_url(self.default_server),
            scope=self.servers[self.default_server]["scope"],
            workspace=workspace,
            id=work_request_id,
        )

        serialized_data = yaml.safe_dump(data)

        if use_stdin:
            self.enterContext(self.patch_sys_stdin_read(serialized_data))
            data_option = "-"
        else:
            data_file = self.create_temporary_file(
                contents=serialized_data.encode("utf-8")
            )
            data_option = str(data_file)

        if event_reactions:
            event_reactions_file = self.create_temporary_file(
                contents=yaml.safe_dump(event_reactions).encode("utf-8")
            )
            cli_args += ["--event-reactions", str(event_reactions_file)]

        cli = self.create_cli(
            ["work-request", "create", *cli_args, "--data", data_option]
        )

        expected_work_request_request = WorkRequestRequest(
            task_name=task_name,
            workspace=workspace,
            task_data=data,
            event_reactions=event_reactions,
        )
        self.assert_create_work_request(
            cli,
            mocked_post_work_request,
            expected_work_request_request,
        )

    def test_create_work_request_specific_workspace(self) -> None:
        """Cli parse the command line and use workspace."""
        workspace_name = "Testing"
        args = ["sbuild", "--workspace", workspace_name]
        self.verify_create_work_request_scenario(args, workspace=workspace_name)

    def test_create_work_request_event_reactions(self) -> None:
        """Cli parse the command line and provide even_reactions."""
        args = ["sbuild"]
        self.verify_create_work_request_scenario(
            args, event_reactions={'on_success': [{'send-notifications': {}}]}
        )

    def test_create_work_request_success_data_from_file(self) -> None:
        """Cli parse the command line and read file to create a work request."""
        args = ["sbuild"]
        self.verify_create_work_request_scenario(args)

    def test_create_work_request_success_data_from_stdin(self) -> None:
        """Cli parse the command line and stdin to create a work request."""
        args = ["sbuild"]
        self.verify_create_work_request_scenario(args, use_stdin=True)

    def test_create_work_request_data_is_empty(self) -> None:
        """Cli parse command line to create a work request with empty data."""
        empty_file = self.create_temporary_file()
        cli = self.create_cli(
            ["work-request", "create", "sbuild", "--data", str(empty_file)]
        )

        stderr, stdout = self.capture_output(
            cli.execute, assert_system_exit_code=3
        )

        self.assertEqual(
            stderr, "Error: data must be a dictionary. It is empty\n"
        )
        self.assertEqual(stdout, "")

    def test_create_work_request_yaml_errors_failed(self) -> None:
        """cli.execute() deal with different invalid task_data."""
        work_requests = [
            {
                "task_data": "test:\n  name: a-name\n"
                "    first-name: some first name",
                "comment": "yaml.safe_load raises ScannerError",
            },
            {
                "task_data": "input:\n  source_url: https://example.com\n )",
                "comment": "yaml.safe_load raises ParserError",
            },
        ]

        for work_request in work_requests:
            task_data = work_request["task_data"]
            with self.subTest(task_data), self.patch_sys_stdin_read(task_data):
                cli = self.create_cli(["work-request", "create", "task-name"])
                stderr, _ = self.capture_output(
                    cli.execute, assert_system_exit_code=3
                )

                self.assertRegex(stderr, "^Error parsing YAML:")
                self.assertRegex(
                    stderr,
                    "Fix the YAML data\n$",
                )


class CliManageWorkRequestTests(BaseCliTests):
    """Tests for CLI manage-work-request subcommand."""

    def test_no_changes(self) -> None:
        """manage-work-request fails if no changes are specified."""
        cli = self.create_cli(["work-request", "manage", "1"])

        stderr, stdout = self.capture_output(
            cli.execute, assert_system_exit_code=3
        )

        self.assertEqual(stderr, "Error: no changes specified\n")
        self.assertEqual(stdout, "")

    def test_set_priority_adjustment(self) -> None:
        """manage-work-request sets the priority adjustment."""
        debusine = self.patch_build_debusine_object()
        cli = self.create_cli(
            ["work-request", "manage", "--set-priority-adjustment", "100", "1"]
        )

        stderr, stdout = self.capture_output(cli.execute)

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        debusine.return_value.work_request_update.assert_called_once_with(
            1, priority_adjustment=100
        )


class CliRetryWorkRequestTests(BaseCliTests):
    """Tests for CLI retry-work-request subcommand."""

    def test_retry(self) -> None:
        """retry-work-request calls the retry method."""
        debusine = self.patch_build_debusine_object()
        cli = self.create_cli(["work-request", "retry", "1"])

        stderr, stdout = self.capture_output(cli.execute)

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        debusine.return_value.work_request_retry.assert_called_once_with(
            1,
        )


class CliAbortWorkRequestTests(BaseCliTests):
    """Tests for CLI abort-work-request subcommand."""

    def test_abort(self) -> None:
        """abort-work-request calls the abort method."""
        debusine = self.patch_build_debusine_object()
        cli = self.create_cli(["work-request", "abort", "1"])

        stderr, stdout = self.capture_output(cli.execute)

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")
        debusine.return_value.work_request_abort.assert_called_once_with(
            1,
        )


class CliWorkRequestListTests(BaseCliTests):
    """Tests for CLI list-work-requests method."""

    @override
    def setUp(self) -> None:
        self.artifact = create_artifact_response(id=1)
        super().setUp()

    def test_list_work_requests_success(self) -> None:
        """CLI lists work requests and prints them."""
        responses = [
            create_work_request_response(id=1),
            create_work_request_response(id=2),
        ]

        def work_request_iter(
            _self: Debusine,
        ) -> Generator[WorkRequestResponse]:
            yield from responses

        self.enterContext(
            mock.patch.object(
                Debusine,
                "work_request_iter",
                autospec=True,
                side_effect=work_request_iter,
            )
        )

        cli = self.create_cli(["work-request", "list"])

        with mock.patch(
            "debusine.client.commands.work_requests.List.list"
        ) as list_:
            stderr, stdout = self.capture_output(cli.execute)

        list_.assert_called_with(responses)

        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")


class CliWorkRequestStatusTests(BaseCliTests):
    """Tests for the Cli functionality related to work request status."""

    @override
    def setUp(self) -> None:
        self.artifact = create_artifact_response(
            id=1,
        )
        super().setUp()

    def artifact_get(
        self,
        debusine: None,  # noqa: ARG002, U100
        artifact_id: int,
    ) -> ArtifactResponse:
        """
        Return copy of self.artifact changing its id to artifact_id.

        :param debusine: unused, only to have the same function signature
          as Debusine.artifact_get() (so it can be used in a mock)
        :param artifact_id: if of the returned artifact.
        """
        artifact = self.artifact.model_copy()
        artifact.id = artifact_id
        return artifact

    def test_show_work_request_success(self) -> None:
        """Cli use Debusine to fetch a WorkRequest and prints it to stdout."""
        created_at = dt.datetime.now(tz=dt.UTC)
        work_request_id = 10
        artifact_ids = [1, 2]

        work_request_response = create_work_request_response(
            id=work_request_id,
            created_at=created_at,
            artifacts=artifact_ids,
            task_data={"architecture": "amd64"},
            status="running",
        )

        self.enterContext(
            mock.patch.object(
                Debusine,
                'work_request_get',
                autospec=True,
                return_value=work_request_response,
            )
        )
        self.enterContext(
            mock.patch.object(
                Debusine,
                "artifact_get",
                autospec=True,
                side_effect=self.artifact_get,
            )
        )

        cli = self.create_cli(["work-request", "show", str(work_request_id)])
        _, stdout = self.capture_output(cli.execute)

        expected_artifacts = []

        for artifact_id in artifact_ids:
            expected_artifacts.append(
                self.artifact_get(None, artifact_id).model_dump(mode="json")
            )

        stdout_parsed = yaml.safe_load(stdout)
        stdout_artifacts = stdout_parsed.pop("artifacts", None)
        stdout_parsed["artifacts"] = [a["id"] for a in stdout_artifacts]
        output = WorkRequestResponse(**stdout_parsed)
        self.assertEqual(output, work_request_response)
        self.assertEqual(stdout_artifacts, expected_artifacts)


class CliProvideSignatureTests(BaseCliTests):
    """Tests for the CLI `provide-signature` command."""

    @contextlib.contextmanager
    def patch_debusine_method(
        self, name: str, **kwargs: Any
    ) -> Generator[mock.MagicMock]:
        """Patch a method on :py:class:`Debusine`."""
        with mock.patch.object(
            Debusine, name, autospec=True, **kwargs
        ) as patcher:
            yield patcher

    @responses.activate
    def verify_provide_signature_scenario(
        self,
        local_changes: bool = False,
        extra_args: list[str] | None = None,
        workspace: str = "System",
    ) -> None:
        """Test a provide-signature scenario."""
        if extra_args is None:
            extra_args = []
        directory = self.create_temporary_directory()
        (tar := directory / "foo_1.0.tar.xz").write_text("tar")
        (dsc := directory / "foo_1.0.dsc").write_text("dsc")
        (buildinfo := directory / "foo_1.0_source.buildinfo").write_text(
            "buildinfo"
        )
        changes = directory / "foo_1.0_source.changes"
        self.write_changes_file(changes, [tar, dsc, buildinfo])
        bare_work_request_response = create_work_request_response(
            id=1,
            task_type="Wait",
            task_name="externaldebsign",
            workspace=workspace,
        )
        full_work_request_response = create_work_request_response(
            id=1,
            task_type="Wait",
            task_name="externaldebsign",
            dynamic_task_data={"unsigned_id": 2},
            workspace=workspace,
        )
        unsigned_artifact_response = create_artifact_response(
            id=2,
            files={
                path.name: create_file_response(
                    size=path.stat().st_size,
                    checksums={"sha256": calculate_hash(path, "sha256").hex()},
                    url=f"https://example.com/{path.name}",
                )
                for path in (tar, dsc, buildinfo, changes)
            },
            workspace=workspace,
        )
        remote_signed_artifact = create_remote_artifact(
            id=3, workspace=workspace
        )
        args = ["work-request", "sign", "1"]
        if local_changes:
            args.extend(["--local-file", str(changes)])
        if extra_args:
            args.extend(["--", *extra_args])
        cli = self.create_cli(args)

        for path in (tar, dsc, buildinfo, changes):
            responses.add(
                responses.GET,
                f"https://example.com/{path.name}",
                body=path.read_bytes(),
            )

        with (
            self.patch_debusine_method(
                "work_request_get", return_value=bare_work_request_response
            ) as mock_work_request_get,
            self.patch_debusine_method(
                "work_request_external_debsign_get",
                return_value=full_work_request_response,
            ) as mock_work_request_external_debsign_get,
            self.patch_debusine_method(
                "artifact_get", return_value=unsigned_artifact_response
            ) as mock_artifact_get,
            mock.patch("subprocess.run") as mock_run,
            self.patch_debusine_method(
                "upload_artifact", return_value=remote_signed_artifact
            ) as mock_upload_artifact,
            self.patch_debusine_method(
                "work_request_external_debsign_complete"
            ) as mock_work_request_external_debsign_complete,
        ):
            stderr, stdout = self.capture_output(cli.execute)

            if local_changes:
                self.assertEqual(stderr, "")
            else:
                self.assertRegex(
                    stderr,
                    r"\A"
                    + "\n".join(
                        [
                            (
                                rf"Artifact file downloaded: "
                                rf".*/{re.escape(path.name)}"
                            )
                            for path in (dsc, buildinfo, changes)
                        ]
                    )
                    + r"\n\Z",
                )
            self.assertEqual(stdout, "")

            # Ensure that the CLI called debusine in the right sequence.
            # The mocks are arranged a bit awkwardly since we want to allow
            # the unmocked Debusine.download_artifact_file to be called.
            debusine_instance = mock_work_request_get.call_args[0][0]
            self.assertIsInstance(debusine_instance, Debusine)
            mock_work_request_get.assert_called_once_with(debusine_instance, 1)
            mock_work_request_external_debsign_get.assert_called_once_with(
                debusine_instance, 1
            )
            mock_artifact_get.assert_called_once_with(debusine_instance, 2)
            mock_run.assert_called_once_with(
                ["debsign", "--re-sign", changes.name, *extra_args],
                cwd=mock.ANY,
                check=True,
            )
            mock_upload_artifact.assert_called_once_with(
                debusine_instance, mock.ANY, workspace=workspace
            )
            signed_local = mock_upload_artifact.call_args[0][1]
            self.assertEqual(signed_local.category, ArtifactCategory.UPLOAD)
            self.assertCountEqual(
                signed_local.files.keys(),
                [changes.name, dsc.name, buildinfo.name],
            )
            self.assertCountEqual(signed_local.remote_files.keys(), [tar.name])
            mock_work_request_external_debsign_complete.assert_called_once_with(
                debusine_instance,
                1,
                WorkRequestExternalDebsignRequest(signed_artifact=3),
            )
        if local_changes:
            for path in (tar, dsc, buildinfo, changes):
                url = f"https://example.com/{path.name}"
                responses.assert_call_count(url, 0)

    def test_debsign(self) -> None:
        """provide-signature calls debsign and posts the output."""
        self.verify_provide_signature_scenario()

    def test_debsign_local_changes(self) -> None:
        """provide-signature uses local .changes."""
        self.verify_provide_signature_scenario(local_changes=True)

    def test_debsign_with_args(self) -> None:
        """provide-signature calls debsign with args and posts the output."""
        self.verify_provide_signature_scenario(extra_args=["-kKEYID"])

    def test_wrong_work_request(self) -> None:
        """provide-signature only works for Wait/externaldebsign requests."""
        debusine = self.patch_build_debusine_object()
        debusine.return_value.work_request_get.return_value = (
            create_work_request_response(task_type="Wait", task_name="delay")
        )
        cli = self.create_cli(["work-request", "sign", "1"])

        stderr, _ = self.capture_output(cli.execute, assert_system_exit_code=3)

        self.assertEqual(
            stderr,
            "Don't know how to provide signature for Wait/delay work request\n",
        )

    def verify_provide_signature_error_scenario(
        self,
        error_regex: str,
        local_file: str | None = None,
        expect_size: dict[str, int] | None = None,
        expect_sha256: dict[str, str] | None = None,
        delete_dsc: bool = False,
    ) -> None:
        """Test a provide-signature failure scenario with local_file."""
        if expect_size is None:
            expect_size = {}
        if expect_sha256 is None:
            expect_sha256 = {}
        directory = self.create_temporary_directory()
        (tar := directory / "foo_1.0.tar.xz").write_text("tar")
        (dsc := directory / "foo_1.0.dsc").write_text("dsc")
        (buildinfo := directory / "foo_1.0_source.buildinfo").write_text(
            "buildinfo"
        )
        changes = directory / "foo_1.0_source.changes"
        self.write_changes_file(changes, [tar, dsc, buildinfo])
        bare_work_request_response = create_work_request_response(
            id=1, task_type="Wait", task_name="externaldebsign"
        )
        full_work_request_response = create_work_request_response(
            id=1,
            task_type="Wait",
            task_name="externaldebsign",
            dynamic_task_data={"unsigned_id": 2},
        )
        unsigned_artifact_response = create_artifact_response(
            id=2,
            files={
                path.name: create_file_response(
                    size=expect_size.get(path.name, path.stat().st_size),
                    checksums={
                        "sha256": expect_sha256.get(
                            path.name, calculate_hash(path, "sha256").hex()
                        )
                    },
                    url=f"https://example.com/{path.name}",
                )
                for path in (tar, dsc, buildinfo, changes)
            },
        )
        if delete_dsc:
            dsc.unlink()
        if local_file is None:
            local_file = str(changes)
        args = ["work-request", "sign", "1", "--local-file", local_file]
        cli = self.create_cli(args)

        with (
            self.patch_debusine_method(
                "work_request_get", return_value=bare_work_request_response
            ),
            self.patch_debusine_method(
                "work_request_external_debsign_get",
                return_value=full_work_request_response,
            ),
            self.patch_debusine_method(
                "artifact_get", return_value=unsigned_artifact_response
            ),
            mock.patch("subprocess.run"),
        ):
            stderr, _ = self.capture_output(
                cli.execute, assert_system_exit_code=3
            )

        self.assertRegex(stderr, error_regex)

    def test_missing_local_changes(self) -> None:
        """provide-signature raises an error for missing --local-file."""
        self.verify_provide_signature_error_scenario(
            r"^--local-file 'nonexistent\.changes' does not exist\.$",
            local_file="nonexistent.changes",
        )

    def test_wrong_local_changes(self) -> None:
        """provide-signature raises an error for the wrong --local-file."""
        directory = self.create_temporary_directory()
        (local_file := directory / "different.changes").write_text("")
        self.verify_provide_signature_error_scenario(
            (
                r"^'[^']+\/different\.changes' is not part of artifact \d+\. "
                r"Expecting 'foo_1.0_source\.changes'$"
            ),
            local_file=str(local_file),
        )

    def test_missing_local_dsc(self) -> None:
        """provide-signature raises an error for missing indirect local-file."""
        self.verify_provide_signature_error_scenario(
            r"^'[^']+\/foo_1.0\.dsc' does not exist\.$",
            delete_dsc=True,
        )

    def test_local_dsc(self) -> None:
        """provide-signature raises an error for the wrong kind of file."""
        self.verify_provide_signature_error_scenario(
            r"^--local-file 'foo\.dsc' is not a \.changes file.$",
            local_file="foo.dsc",
        )

    def test_size_mismatch(self) -> None:
        """provide-signature verifies the size of local files."""
        self.verify_provide_signature_error_scenario(
            r"^'[^']+/foo_1\.0\.dsc' size mismatch \(expected 999 bytes\)$",
            expect_size={"foo_1.0.dsc": 999},
        )

    def test_hash_mismatch(self) -> None:
        """provide-signature verifies the hashes of local files."""
        self.verify_provide_signature_error_scenario(
            (
                r"^'[^']+/foo_1\.0\.dsc' hash mismatch "
                r"\(expected sha256 = abc123\)$"
            ),
            expect_sha256={"foo_1.0.dsc": "abc123"},
        )


class CliOnWorkRequestTests(BaseCliTests):
    """Tests for Cli on-work-request-completed."""

    def test_command_does_not_exist(self) -> None:
        """Cli report that command does not exist."""
        command = "/some/file/does/not/exist"
        cli = self.create_cli(["work-request", "on-completed", command])

        stderr, stdout = self.capture_output(
            cli.execute, assert_system_exit_code=3
        )

        self.assertEqual(
            stderr, f'Error: "{command}" does not exist or is not executable\n'
        )
        self.assertEqual(stdout, "")

    def test_command_is_not_executable(self) -> None:
        """Cli report that command is not executable."""
        command = self.create_temporary_file()
        cli = self.create_cli(["work-request", "on-completed", str(command)])

        stderr, stdout = self.capture_output(
            cli.execute, assert_system_exit_code=3
        )

        self.assertEqual(
            stderr, f'Error: "{command}" does not exist or is not executable\n'
        )
        self.assertEqual(stdout, "")

    def debusine_on_work_request_completed_is_called(
        self,
        *,
        workspaces: list[str] | None = None,
        last_completed_at: Path | None = None,
        silent: bool | None = False,
        extra_args: list[str] | None = None,
    ) -> None:
        """Cli call debusine.on_work_request_completed."""
        command = self.create_temporary_file()
        command.chmod(stat.S_IXUSR | stat.S_IRUSR)

        debusine = self.patch_build_debusine_object()

        cli_args = []

        if silent:
            cli_args.extend(["--silent"])

        cli_args.extend(
            ["work-request", "on-completed", str(command), *(extra_args or [])]
        )

        if workspaces:
            cli_args.extend(["--workspace", *workspaces])

        if last_completed_at:
            cli_args.extend(["--last-completed-at", str(last_completed_at)])

        cli = self.create_cli(cli_args)

        stderr, stdout = self.capture_output(cli.execute)

        self.assertEqual(stderr, "")
        self.assertEqual(stdout, "")

        debusine.return_value.on_work_request_completed.assert_called_with(
            workspaces=workspaces,
            last_completed_at=last_completed_at,
            command=[str(command), *(extra_args or [])],
            working_directory=Path.cwd(),
        )

    def test_debusine_on_work_request_completed(self) -> None:
        """Cli call debusine.on_work_request_completed."""
        self.debusine_on_work_request_completed_is_called(silent=True)

    def test_debusine_on_work_request_completed_with_extra_args(self) -> None:
        self.debusine_on_work_request_completed_is_called(
            silent=True, extra_args=["foo", "bar"]
        )

    def test_debusine_on_work_request_completed_with_workspace(self) -> None:
        """Cli call debusine.on_work_request_completed."""
        self.debusine_on_work_request_completed_is_called(workspaces=["System"])

    def test_debusine_on_work_request_completed_create_last_completed(
        self,
    ) -> None:
        """
        Cli call debusine.on_work_request_completed.

        File last_completed_at is created.
        """
        last_completed_at = self.create_temporary_file()
        last_completed_at.unlink()

        self.debusine_on_work_request_completed_is_called(
            last_completed_at=last_completed_at
        )

        expected = json.dumps({"last_completed_at": None}, indent=2) + "\n"
        self.assertEqual(last_completed_at.read_text(), expected)

    def test_debusine_on_work_request_completed_last_completed_existed(
        self,
    ) -> None:
        """Cli call debusine.on_work_request_completed."""
        contents = json.dumps({"last_completed_at": None}).encode("utf-8")
        last_completed_at = self.create_temporary_file(contents=contents)

        self.debusine_on_work_request_completed_is_called(
            last_completed_at=last_completed_at
        )

    def test_debusine_on_work_request_last_completed_no_write_permission(
        self,
    ) -> None:
        """Cli cannot write to last-completed-at file."""
        cannot_write = self.create_temporary_file()
        cannot_write.chmod(0o444)
        cli = self.create_cli(
            [
                "work-request",
                "on-completed",
                "/bin/echo",
                "--last-completed-at",
                str(cannot_write),
            ]
        )

        stderr, _ = self.capture_output(cli.execute, assert_system_exit_code=3)
        self.assertEqual(
            stderr, f'Error: write or read access denied for "{cannot_write}"\n'
        )

    def test_debusine_on_work_request_last_completed_no_read_permission(
        self,
    ) -> None:
        """Cli cannot write to last-completed-at file."""
        cannot_write = self.create_temporary_file()
        cannot_write.chmod(0o333)
        cli = self.create_cli(
            [
                "work-request",
                "on-completed",
                "/bin/echo",
                "--last-completed-at",
                str(cannot_write),
            ]
        )

        stderr, _ = self.capture_output(cli.execute, assert_system_exit_code=3)
        self.assertEqual(
            stderr, f'Error: write or read access denied for "{cannot_write}"\n'
        )

    def test_debusine_on_work_request_dir_last_completed_no_permission(
        self,
    ) -> None:
        """Cli cannot write to last-completed-at directory."""
        directory = self.create_temporary_directory()
        directory.chmod(0o000)
        cli = self.create_cli(
            [
                "work-request",
                "on-completed",
                "/bin/echo",
                "--last-completed-at",
                str(directory / "file.json"),
            ]
        )

        stderr, _ = self.capture_output(cli.execute, assert_system_exit_code=3)
        self.assertEqual(
            stderr, f'Error: write access denied for directory "{directory}"\n'
        )
