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

import asyncio
import functools
import logging
import re
import secrets
import signal
import socket
import subprocess
from collections.abc import AsyncGenerator, Coroutine
from pathlib import Path
from textwrap import dedent
from typing import Any, AnyStr, Literal, Never, override
from unittest import mock
from unittest.mock import MagicMock

import aiohttp
from aiohttp import RequestInfo
from aiohttp.client_reqrep import ConnectionKey
from aiohttp.web import Response
from aiohttp.web_exceptions import (
    HTTPForbidden,
    HTTPInternalServerError,
    HTTPServiceUnavailable,
)
from multidict import CIMultiDict, CIMultiDictProxy
from yarl import URL

import debusine.tasks.models as task_models
from debusine.artifacts.models import WorkRequestResults
from debusine.client.server_notifications import ServerNotifications
from debusine.tasks import Sbuild, TaskConfigError
from debusine.tasks.models import OutputData, OutputDataError, WorkerType
from debusine.test import TestCase
from debusine.worker import Worker, _worker
from debusine.worker.config import ConfigHandler
from debusine.worker.system_information import system_metadata
from debusine.worker.task_runner import TaskRunner
from debusine.worker.tests import server
from debusine.worker.tests.server import ServerConfig


class WorkerConfigTestCase(TestCase, server.DebusineAioHTTPTestCase):
    """Test case with a worker configuration."""

    @override
    async def setUpAsync(self) -> None:
        await super().setUpAsync()
        self.api_url = str(self.server.make_url('')) + "/api"

        self.temp_config_directory = self.create_temp_config_directory(
            {'General': {'api-url': self.api_url}}
        )

        self.config = ConfigHandler(
            directories=[self.temp_config_directory.as_posix()]
        )


class WorkerTestCase(WorkerConfigTestCase):
    """Base test case for Worker tests."""

    @override
    async def setUpAsync(self) -> None:
        await super().setUpAsync()
        self.default_sigint_handler = signal.getsignal(signal.SIGINT)
        self.default_sigterm_handler = signal.getsignal(signal.SIGTERM)

    @override
    def tearDown(self) -> None:
        # Restore signal handlers. Cli.execute() changes them
        signal.signal(signal.SIGINT, self.default_sigint_handler)
        signal.signal(signal.SIGTERM, self.default_sigterm_handler)
        super().tearDown()


class WorkerRegistrationTests(WorkerTestCase):
    """Test worker registration."""

    worker: Worker

    @override
    async def setUpAsync(self) -> None:
        await super().setUpAsync()
        self.worker = await self.enterAsyncContext(
            Worker(
                worker_type=WorkerType.EXTERNAL,
                working_directory=self.temp_config_directory.as_posix(),
                config=self.config,
            )
        )

        # Don't risk triggering local hooks
        self.worker.AFTER_WORK_REQUEST_HOOK_DIR = Path("/nonexistent")

    def patch_http_method(self, http_method: str, **kwargs: Any) -> MagicMock:
        """Patch a specific HTTP method of Worker's HTTP Client."""
        # setup_worker must be called first.
        assert self.worker is not None

        mocked = self.enterContext(
            mock.patch.object(
                self.worker._async_http_client, http_method, autospec=True
            )
        )
        for key, value in kwargs.items():
            setattr(mocked, key, value)
        return mocked

    async def test_registration_success(self) -> None:
        """
        Worker client registers successfully.

        Client creates a token, sends it to the server and save to the
        configuration.
        """
        # No token in the configuration
        self.assertIsNone(self.config.token)

        # Worker connects: it will create a token, send and save
        await self.worker.require_token()

        # A token has been retrieved
        self.assertIsNotNone(self.config.token)

        registration_request = self.server_requests[-1]
        self.assertEqual(registration_request.path, '/api/1.0/worker/register/')

        # Server side received expected registration data
        self.assertEqual(
            registration_request.json_content,
            {
                'token': self.config.token,
                'fqdn': socket.getfqdn(),
                'worker_type': WorkerType.EXTERNAL,
            },
        )
        assert registration_request.headers is not None
        self.assertNotIn('token', registration_request.headers)

    async def test_registration_success_activation_token(self) -> None:
        """Worker client registers successfully with an activation token."""
        activation_token = secrets.token_hex(32)
        activation_token_path = Path(self.config._activation_token_file)
        activation_token_path.write_bytes(activation_token.encode())  # noqa: ASYNC240
        activation_token_path.chmod(0o600)  # noqa: ASYNC240
        self.assertEqual(self.config.activation_token, activation_token)

        # No token in the configuration
        self.assertIsNone(self.config.token)

        # Worker connects: it will create a token, send and save
        await self.worker.require_token()

        # A token has been retrieved
        self.assertIsNotNone(self.config.token)

        registration_request = self.server_requests[-1]
        self.assertEqual(registration_request.path, '/api/1.0/worker/register/')

        # Server side received expected registration data
        self.assertEqual(
            registration_request.json_content,
            {
                'token': self.config.token,
                'fqdn': socket.getfqdn(),
                'worker_type': WorkerType.EXTERNAL,
            },
        )
        assert registration_request.headers is not None
        self.assertEqual(
            registration_request.headers['token'], activation_token
        )

    async def test_registration_failure(self) -> None:
        """
        Worker client registration failure.

        The server returns 403, the client raises WorkerRegistrationError.
        """
        self.server_config.register_response_status_code = (
            HTTPForbidden.status_code
        )

        log_message = (
            'Could not register. Server HTTP status code: 403 Body: error'
        )

        with (
            self.assertLogsContains(log_message, logger="debusine.worker"),
            self.assertRaisesSystemExit(3),
        ):
            await self.worker.require_token()

    async def test_client_connector_error(self) -> None:
        """Worker cannot connect to the server: server unreachable."""
        connection_key = mock.create_autospec(spec=ConnectionKey)
        mock_post = self.patch_http_method(
            "post",
            side_effect=aiohttp.client_exceptions.ClientConnectorError(
                connection_key, OSError("Unreachable")
            ),
        )
        log_message = "Could not register. Server unreachable: "
        with (
            self.assertLogsContains(log_message, logger="debusine.worker"),
            self.assertRaisesSystemExit(3),
        ):
            await self.worker.require_token()

        self.assertTrue(mock_post.called)

    async def test_connection_read_error(self) -> None:
        """Worker cannot read data from the server (low level disconnection)."""
        mock_post = self.patch_http_method(
            "post",
            side_effect=aiohttp.client_exceptions.ClientResponseError(
                RequestInfo(
                    URL('https://test.com'),
                    '',
                    CIMultiDictProxy(CIMultiDict()),
                    URL('https://test.com'),
                ),
                (),
            ),
        )
        log_message = (
            'Could not register. Server HTTP status code: 0 Body: Not available'
        )
        with (
            self.assertLogsContains(log_message, logger="debusine.worker"),
            self.assertRaisesSystemExit(3),
        ):
            await self.worker.require_token()

        self.assertTrue(mock_post.called)

    async def test_connection_rejected(self) -> None:
        """Worker tries to connect, server returns 500."""
        self.server_config.register_response_status_code = (
            HTTPInternalServerError.status_code
        )

        log_message = (
            'Could not register. Server HTTP status code: 500 Body: error'
        )

        with (
            self.assertLogsContains(log_message, logger="debusine.worker"),
            self.assertRaisesSystemExit(3),
        ):
            await self.worker.require_token()


class WorkerEnvTests(WorkerTestCase):
    """Test Worker environment setup/teardown."""

    worker: Worker

    async def setup_worker(
        self,
        *,
        log_level: str | None = None,
        working_directory: Path | None = None,
        setup_logging: bool = True,
        use_working_directory: bool = True,
    ) -> None:
        kwargs = {}
        if use_working_directory:
            kwargs["working_directory"] = (
                working_directory or self.temp_config_directory
            ).as_posix()

        if not setup_logging:
            # Disable the logging configuration, it makes it hard to test the
            # logs
            self.enterContext(mock.patch.object(Worker, "_setup_logging"))

        self.worker = await self.enterAsyncContext(
            Worker(
                log_level=log_level,
                worker_type=WorkerType.EXTERNAL,
                config=self.config,
                **kwargs,
            )
        )

        # Don't risk triggering local hooks
        self.worker.AFTER_WORK_REQUEST_HOOK_DIR = Path("/nonexistent")

    def patch_config_handler(self) -> MagicMock:
        """
        Patch ConfigHandler class.

        :return: the MagicMock
        """
        config_handler_mocked = self.enterContext(
            mock.patch("debusine.worker._worker.ConfigHandler", autospec=True)
        )
        instance = config_handler_mocked.return_value
        instance.working_directory = Path('/tmp')
        return config_handler_mocked

    def assertQueuedEvents(self, expected: list[str]) -> None:
        """
        Check that the queued events match the given list.

        Events are stringified before comparison
        """
        actual: list[str] = []
        while not self.worker.event_queue.empty():
            evt = self.worker.event_queue.get_nowait()
            assert isinstance(evt, _worker.Event)
            actual.append(str(evt))
        assert actual == expected

    def test_worker_init(self) -> None:
        """Init without any keyword."""
        mocked_config_handler = self.patch_config_handler()

        worker = Worker(
            log_level="WARNING",
            working_directory=self.temp_config_directory.as_posix(),
        )

        # Worker() created a ConfigHandler default object and set it
        # to Worker._config
        self.assertIsNotNone(worker._config)
        mocked_config_handler.assert_called_once_with()

    def test_worker_init_signing(self) -> None:
        """Signing workers use different dirs and require an HTTPS api-url."""
        mocked_config_handler = self.patch_config_handler()

        worker = Worker(
            log_level="WARNING",
            worker_type=WorkerType.SIGNING,
            working_directory=self.temp_config_directory.as_posix(),
        )

        self.assertIsNotNone(worker._config)
        mocked_config_handler.assert_called_once_with(
            directories=[
                str(Path.home() / ".config/debusine/signing"),
                "/etc/debusine/signing",
            ],
            require_https=True,
        )

    async def test_worker_init_use_config_handler(self) -> None:
        """worker.set_config_handler() sets the ConfigHandler."""
        await self.setup_worker(log_level="WARNING")
        assert self.worker is not None

        self.assertIs(self.worker._config, self.config)

    async def test_worker_init_bad_working_directory(self) -> None:
        with (
            self.assertLogsContains(
                "could not create working directory", logger="debusine.worker"
            ),
            self.assertRaisesSystemExit(3),
        ):
            await self.setup_worker(
                working_directory=Path("/nonexistent"),
                setup_logging=False,
            )

    async def test_worker_init_working_directory_from_config(self) -> None:
        mydir = self.temp_config_directory / "foobar"
        self.config['General']['working-directory'] = str(mydir)

        await self.setup_worker(use_working_directory=False)

        assert self.worker is not None
        self.assertEqual(self.worker._workdir, mydir)

    async def test_clean_working_directory(self) -> None:
        """worker._clean_workdir() cleans up working directory."""
        await self.setup_worker()
        assert self.worker is not None
        # Files that will be dropped
        work_request_dir = self.temp_config_directory / "WR-123"
        sample_file = work_request_dir / "sample.txt"
        work_request_dir.mkdir(parents=True)
        sample_file.write_text("sample text")
        # Files that will be kept
        random_dir = self.temp_config_directory / "random"
        random_dir.mkdir()

        self.worker.clean_workdir()

        self.assertFalse(sample_file.exists())
        self.assertFalse(work_request_dir.exists())
        self.assertTrue(random_dir.exists())

    async def test_clean_working_directory_with_failing_rmtree(self) -> None:
        """worker._clean_workdir() fails gracefully when rmtree fails."""
        await self.setup_worker()
        assert self.worker is not None
        # Files that will be dropped
        work_request_dir = self.temp_config_directory / "WR-123"
        work_request_dir.mkdir(parents=True)
        with (
            self.assertLogsContains(
                f"Failed to remove '{work_request_dir}'",
                logger="debusine.worker",
            ),
            mock.patch("shutil.rmtree") as mock_rmtree,
        ):
            mock_rmtree.side_effect = OSError("error")
            self.worker.clean_workdir()

    async def assert_logging_setup(
        self,
        argv_setup: dict[str, Any],
        config_setup: dict[str, Any],
        expected: dict[str, Any],
    ) -> None:
        """
        Assert logging setup for argv_setup and config_setup.

        :param argv_setup: parameters that a user might have passed using
          the CLI
        :param config_setup: parameters that might be set in the config.ini
        :param expected: given the argv_setup and config_setup: what is the
          expected configuration
        """
        if config_setup['log-level'] is not None:
            self.config['General']['log-level'] = logging.getLevelName(
                config_setup['log-level']
            )

        # Normalizes log_level: always using the str representation and not
        # the int
        if argv_setup['log_level'] is not None:
            argv_setup['log_level'] = logging.getLevelName(
                argv_setup['log_level']
            )
        expected['level'] = logging.getLevelName(expected['level'])

        # Patches basicConfig
        patcher = mock.patch(
            'debusine.worker._worker.logging.basicConfig', autospec=True
        )
        mocked_logging = patcher.start()
        self.addCleanup(patcher.stop)

        await self.setup_worker(**argv_setup)

        self.assertTrue(mocked_logging.called)
        called_args = mocked_logging.call_args[1]
        self.assertEqual(called_args['level'], expected['level'])

        # Verify that handlers include JournalHandler
        self.assertIn('handlers', called_args)
        handlers = called_args['handlers']
        self.assertEqual(len(handlers), 1)
        handler_class = type(handlers[0]).__name__
        self.assertIn("JournalHandler", handler_class)

    async def test_setup_logging_no_logging_configuration(self) -> None:
        """Logging system: no explicit setup: use Worker.DEFAULT_LOG_LEVEL."""
        argv_setup = {'log_level': None}
        config_setup = {'log-level': None}
        expected = {'level': Worker.DEFAULT_LOG_LEVEL}
        await self.assert_logging_setup(argv_setup, config_setup, expected)

    async def test_setup_logging_from_constructor(self) -> None:
        """Logging system: use settings from the Worker constructor."""
        argv_setup = {'log_level': logging.DEBUG}
        config_setup = {'log-level': None}
        expected = {'level': logging.DEBUG}

        await self.assert_logging_setup(argv_setup, config_setup, expected)

    async def test_setup_logging_from_config(self) -> None:
        """Logging system: use settings from the configuration file."""
        argv_setup = {'log_level': None}
        config_setup = {'log-level': logging.DEBUG}
        expected = {'level': logging.DEBUG}
        await self.assert_logging_setup(argv_setup, config_setup, expected)

    async def test_setup_logging_worker_constructor_overwrites_config(
        self,
    ) -> None:
        """Logging system: use settings from the Worker constructor."""
        argv_setup = {'log_level': logging.DEBUG}
        config_setup = {'log-level': logging.ERROR}
        expected = {'level': logging.DEBUG}

        await self.assert_logging_setup(argv_setup, config_setup, expected)

    async def test_log_forced_exit(self) -> None:
        """log_forced_exit logs the event."""
        await self.setup_worker()
        assert self.worker is not None

        with self.assertLogsContains(
            "Terminated with signal SIGINT",
            logger="debusine.worker",
            level=logging.INFO,
        ):
            self.worker.log_forced_exit(signal.SIGINT)

    async def test_asyncio_sleep(self) -> None:
        """Worker._asyncio_sleep calls asyncio.sleep."""
        mocked = self.enterContext(mock.patch('asyncio.sleep', autospec=True))
        await Worker._asyncio_sleep(1.5)
        mocked.assert_called_with(1.5)

    async def test_create_task_signal_int_term_handler(self) -> None:
        """
        Test Task returned from Worker._create_task_signal_int_term_handler.

        Assert that the task logs "Terminated with signal ...".
        Other tests on the behaviour of Worker._signal_int_term_handler
        is implemented in the method test_signal_int_term_handler.

        Testing that Worker._create_task_signal_int_term_handler returns
        a Task that will execute Worker._signal_int_term_handler might be
        possible but only if accessing Task._coro private method.
        """
        await self.setup_worker()
        assert self.worker is not None

        test_signals = [signal.SIGINT, signal.SIGTERM]

        for test_signal in test_signals:
            with self.subTest(test_signal=repr(test_signal)):
                sig_handler = self.worker._create_task_signal_int_term_handler(
                    test_signal
                )

                self.assertIsInstance(sig_handler, asyncio.Task)

                with (
                    mock.patch.object(
                        self.worker.event_queue, "put"
                    ) as queue_put,
                    self.assertLogsContains(
                        f"Terminated with signal {test_signal.name}",
                        logger="debusine.worker",
                    ),
                ):
                    await sig_handler
                    queue_put.assert_called_once()
                    self.assertEqual(
                        str(queue_put.call_args.args[0]),
                        f"shutdown: signal {test_signal} received",
                    )

    async def test_setup_signal_int_term_handler(self) -> None:
        """Test that SIGINT/SIGTERM setup is done in Worker.main()."""
        await self.setup_worker()
        assert self.worker is not None

        connect = self.enterContext(
            mock.patch.object(
                self.worker,
                "process_notifications",
                return_value=None,
            )
        )

        add_signal_handler = self.enterContext(
            mock.patch.object(asyncio.get_running_loop(), "add_signal_handler")
        )
        with self.assertRaisesSystemExit(4):
            await self.worker.main()

        add_signal_handler.assert_has_calls(
            [
                mock.call(signal.SIGINT, mock.ANY),
                mock.call(signal.SIGTERM, mock.ANY),
            ]
        )

        connect.assert_awaited()

    async def test_int_term_handler_unexpected_signum(self) -> None:
        await self.setup_worker()
        assert self.worker is not None
        await self.worker._signal_int_term_handler(signal.SIGUSR1)
        self.assertQueuedEvents(["shutdown: signal 10 received"])

    async def test_exit_code(self) -> None:
        await self.setup_worker()
        assert self.worker is not None
        self.enterContext(mock.patch.object(self.worker, "require_token"))
        process_notifications = self.enterContext(
            mock.patch.object(
                self.worker,
                "process_notifications",
            )
        )
        for event, code in (
            (_worker.ServerDisconnectShutdownEvent(), 1),
            (_worker.ReportTaskResultFailedShutdownEvent(), 2),
            (
                _worker.SignalReceivedShutdownEvent(signal.SIGTERM),
                128 + signal.SIGTERM,
            ),
            (
                _worker.SignalReceivedShutdownEvent(signal.SIGINT),
                128 + signal.SIGINT,
            ),
            (
                _worker.SignalReceivedShutdownEvent(signal.SIGUSR1),
                128 + signal.SIGUSR1,
            ),
            (None, 4),
        ):
            with self.subTest(event=str(event)):
                process_notifications.return_value = event
                with self.assertRaisesSystemExit(code):
                    await self.worker.main()


class WorkerEventsTests(WorkerTestCase):
    """Test worker event handling."""

    worker: Worker

    @override
    async def setUpAsync(self) -> None:
        await super().setUpAsync()

        token = '6c931875627131b5135b7de3371c44'
        self.server_config.registered_token = token
        self.config.write_token(token)

        # Server notification payloads that will be generated by the mocked
        # ServerNotifications
        self.server_notification_payloads: list[dict[str, Any]] = []

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

        self.server_notifications = mock.MagicMock(spec=ServerNotifications)
        self.server_notifications.__aenter__.return_value = (
            self.server_notifications
        )
        self.server_notifications.__aexit__.return_value = None
        self.server_notifications.messages = mock_messages

        self.enterContext(
            mock.patch(
                "debusine.client.debusine.Debusine.server_notifications",
                return_value=self.server_notifications,
            )
        )

        self.worker = await self.enterAsyncContext(
            Worker(
                worker_type=WorkerType.EXTERNAL,
                working_directory=self.temp_config_directory.as_posix(),
                config=self.config,
            )
        )

        # Don't risk triggering local hooks
        self.worker.AFTER_WORK_REQUEST_HOOK_DIR = Path("/nonexistent")

    def assertQueuedEvents(self, expected: list[str]) -> None:
        """
        Check that the queued events match the given list.

        Events are stringified before comparison
        """
        actual: list[str] = []
        while not self.worker.event_queue.empty():
            evt = self.worker.event_queue.get_nowait()
            assert isinstance(evt, _worker.Event)
            actual.append(str(evt))
        expected = [
            *expected,
            "shutdown: server notification connection closed",
        ]
        assert actual == expected

    async def test_handle_server_notifications_payloads_to_events(self) -> None:
        RESPONSES = ServerConfig.RESPONSES
        self.server_notification_payloads = [
            RESPONSES["connected"],
            RESPONSES["request_dynamic_metadata"],
            RESPONSES["work_request_available"],
        ]
        await self.worker.handle_server_notifications()
        self.assertQueuedEvents(
            [
                "connected",
                "request_dynamic_metadata",
                "work_request_available",
            ]
        )

    async def test_handle_server_notifications_ignore_unsupported_payload(
        self,
    ) -> None:
        self.server_notification_payloads = [
            {"text": "unsupported"},
            ServerConfig.RESPONSES["connected"],
        ]
        with self.assertLogs(
            logger="debusine.worker", level=logging.WARNING
        ) as log:
            await self.worker.handle_server_notifications()
        self.assertQueuedEvents(["connected"])
        self.assertEqual(
            log.output,
            [
                "WARNING:debusine.worker:"
                "Ignoring unsupported or malformed server message:"
                " {'text': 'unsupported'}: unsupported server event"
            ],
        )

    async def test_handle_server_notifications_ignore_payload_without_text(
        self,
    ) -> None:
        self.server_notification_payloads = [
            {},
            ServerConfig.RESPONSES["connected"],
        ]
        with self.assertLogs(
            logger="debusine.worker", level=logging.WARNING
        ) as log:
            await self.worker.handle_server_notifications()
        self.assertQueuedEvents(["connected"])
        self.assertEqual(
            log.output,
            [
                "WARNING:debusine.worker:"
                "Ignoring unsupported or malformed server message:"
                " {}: message payload without 'text'"
            ],
        )

    async def test_process_notifications_shutdown(self) -> None:
        event = _worker.SignalReceivedShutdownEvent(signal.SIGUSR1)
        await self.worker.event_queue.put(event)
        with self.assertLogs(
            logger="debusine.worker", level=logging.INFO
        ) as log:
            self.assertEqual(await self.worker.process_notifications(), event)
        self.assertEqual(
            log.output,
            [
                "INFO:debusine.worker:"
                "Shutdown requested: shutdown: signal 10 received"
            ],
        )

    async def test_process_notifications_connected(self) -> None:
        await self.worker.event_queue.put(_worker.ConnectedEvent())
        with self.assertLogs(
            logger="debusine.worker", level=logging.DEBUG
        ) as log:
            await self.worker.process_notifications()
        self.assertEqual(
            log.output[:2],
            [
                "DEBUG:debusine.worker:Worker event: connected",
                "DEBUG:debusine.worker:Received the 'connected' message",
            ],
        )

    async def test_process_notifications_request_dynamic_metadata(self) -> None:
        await self.worker.event_queue.put(_worker.RequestDynamicMetadataEvent())
        sdm = self.enterContext(
            mock.patch.object(self.worker, "_send_dynamic_metadata")
        )
        await self.worker.process_notifications()
        sdm.assert_called()

    async def test_process_notifications_request_dynamic_metadata_awaited(
        self,
    ) -> None:
        await self.worker.event_queue.put(_worker.RequestDynamicMetadataEvent())

        exception = RuntimeError("expected error")

        async def mock_send_dynamic_metadata() -> None:
            raise exception

        self.enterContext(
            mock.patch.object(
                self.worker,
                "_send_dynamic_metadata",
                side_effect=mock_send_dynamic_metadata,
            )
        )
        with self.assertRaises(ExceptionGroup) as exc:
            await self.worker.process_notifications()
        self.assertIs(exc.exception.exceptions[0], exception)

    async def test_process_notifications_work_request_available(self) -> None:
        await self.worker.event_queue.put(_worker.WorkRequestAvailableEvent())
        sdm = self.enterContext(
            mock.patch.object(self.worker, "_request_and_execute_work_request")
        )
        await self.worker.process_notifications()
        sdm.assert_called()

    async def test_process_notifications_work_request_available_awaited(
        self,
    ) -> None:
        await self.worker.event_queue.put(_worker.WorkRequestAvailableEvent())

        exception = RuntimeError("expected error")

        async def mock_request_and_execute_work_request() -> None:
            raise exception

        self.enterContext(
            mock.patch.object(
                self.worker,
                "_request_and_execute_work_request",
                side_effect=mock_request_and_execute_work_request,
            )
        )
        with self.assertRaises(ExceptionGroup) as exc:
            await self.worker.process_notifications()
        self.assertIs(exc.exception.exceptions[0], exception)

    async def test_process_notifications_unsupported(self) -> None:
        await self.worker.event_queue.put("invalid")  # type: ignore[arg-type]
        with (
            self.assertLogs(
                logger="debusine.worker", level=logging.WARNING
            ) as log,
        ):
            await self.worker.process_notifications()
        self.assertEqual(
            log.output,
            ["WARNING:debusine.worker:Ignoring unsupported event: invalid"],
        )


class WorkerTests(WorkerTestCase):
    """Test Worker class."""

    worker: Worker

    @override
    async def setUpAsync(self) -> None:
        await super().setUpAsync()

        token = '6c931875627131b5135b7de3371c44'
        self.server_config.registered_token = token
        self.config.write_token(token)

        self.worker = await self.enterAsyncContext(
            Worker(
                working_directory=self.temp_config_directory.as_posix(),
                worker_type=WorkerType.EXTERNAL,
                config=self.config,
            )
        )

        # Don't risk triggering local hooks
        self.worker.AFTER_WORK_REQUEST_HOOK_DIR = Path("/nonexistent")

    def set_events(
        self, *events: _worker.Event, include_quit_main_loop: bool = True
    ) -> None:
        self.enterContext(
            mock.patch.object(self.worker, "handle_server_notifications")
        )
        for event in events:
            self.worker.event_queue.put_nowait(event)
        if include_quit_main_loop:
            self.worker.event_queue.put_nowait(_worker.QuitMainLoopEvent())

    def patch_worker(self, attribute: str, **kwargs: Any) -> mock.AsyncMock:
        """Patch an attribute of the worker object."""
        patcher = mock.patch.object(self.worker, attribute, autospec=True)
        mocked = patcher.start()
        for key, value in kwargs.items():
            setattr(mocked, key, value)
        self.addCleanup(patcher.stop)
        return mocked

    def patch_debusine_work_request_completed_update(self) -> MagicMock:
        """Patch Debusine.work_request_completed_update. Return its mock."""
        # setup_worker must be called first.
        assert self.worker is not None
        return self.enterContext(
            mock.patch(
                "debusine.client.debusine.Debusine.work_request_completed_update",
                autospec=True,
            )
        )

    def assertQueuedEvents(self, expected: list[str]) -> None:
        """
        Check that the queued events match the given list.

        Events are stringified before comparison
        """
        actual: list[str] = []
        while not self.worker.event_queue.empty():
            evt = self.worker.event_queue.get_nowait()
            assert isinstance(evt, _worker.Event)
            actual.append(str(evt))
        assert actual == expected

    async def capture_build_task(self) -> asyncio.Task[None]:
        """Wait for a build task to start, and return it."""
        orig_task_group_create_task = asyncio.TaskGroup.create_task

        # Monitor tasks created during event processing, to capture the first
        # that runs Worker._request_and_execute_work_request
        build_task: asyncio.Task[None] | None = None

        found = asyncio.Event()

        def spy_create_task(
            task_group: asyncio.TaskGroup, awaitable: Coroutine[Any, Any, Never]
        ) -> asyncio.Task[Any]:
            nonlocal build_task
            task = orig_task_group_create_task(task_group, awaitable)
            if (
                awaitable.__qualname__
                == "Worker._request_and_execute_work_request"
            ):
                build_task = task
                found.set()
            return task

        with mock.patch(
            "asyncio.TaskGroup.create_task",
            side_effect=spy_create_task,
            autospec=True,
        ):
            await found.wait()

        assert build_task is not None
        return build_task

    async def test_process_message_dynamic_metadata_request_signing(
        self,
    ) -> None:
        """Signing workers send the correct worker type in dynamic metadata."""
        self.worker._worker_type = WorkerType.SIGNING

        with (
            mock.patch(
                "debusine.worker._worker.analyze_worker_all_executors",
                return_value={},
            ),
            mock.patch(
                "debusine.worker._worker.analyze_external_worker_tasks",
                return_value={},
            ),
            mock.patch(
                "debusine.worker._worker.Worker._get_provided_tags",
                autospec=True,
                return_value=set(),
            ),
        ):
            await self.worker._send_dynamic_metadata()

        assert self.server_latest_request is not None
        self.assertEqual(
            self.server_latest_request.path, "/api/1.0/worker/dynamic-metadata/"
        )
        self.assertTrue(self.server_latest_request.is_authenticated)
        self.assertEqual(
            self.server_latest_request.json_content,
            {
                **system_metadata(task_models.WorkerType.SIGNING),
                "provided_tags": [],
            },
        )

    async def test_send_dynamic_metadata(self) -> None:
        """Test a successful dynamic metadata update."""
        worker_metadata = {"test": True}

        with (
            mock.patch(
                "debusine.worker._worker.Worker._get_worker_metadata",
                autospec=True,
                return_value=worker_metadata,
            ),
            self.assertLogs(
                logger="debusine.worker", level=logging.DEBUG
            ) as logs,
        ):
            await self.worker._send_dynamic_metadata()

        regex = r'Send dynamic metadata'
        found = sum(
            int(re.search(regex, log) is not None) for log in logs.output
        )
        self.assertEqual(found, 1)

        assert self.server_latest_request is not None
        self.assertEqual(
            self.server_latest_request.path, '/api/1.0/worker/dynamic-metadata/'
        )

        self.assertTrue(self.server_latest_request.is_authenticated)

        self.assertEqual(
            self.server_latest_request.json_content, worker_metadata
        )

    async def test_send_dynamic_metadata_failed(self) -> None:
        """Test a failed dynamic metadata update."""
        executors_metadata = {"executor:unshare:available": True}
        tasks_metadata = {"sbuild:version": 1, "sbuild:available": True}

        self.server_config.dynamic_metadata_status_code = (
            HTTPServiceUnavailable.status_code
        )

        with (
            mock.patch(
                "debusine.worker._worker.analyze_worker_all_executors",
                autospec=True,
                return_value=executors_metadata,
            ),
            mock.patch(
                'debusine.worker._worker.analyze_external_worker_tasks',
                autospec=True,
                return_value=tasks_metadata,
            ),
            self.assertLogsContains(
                "ERROR:debusine.worker:"
                "Failed to send dynamic_metadata: (response: 503): ",
                logger="debusine.worker",
            ),
        ):
            await self.worker._send_dynamic_metadata()

        assert self.server_latest_request is not None
        self.assertEqual(
            self.server_latest_request.path, '/api/1.0/worker/dynamic-metadata/'
        )

    async def receive_task_submit_result(
        self,
        execute_result: WorkRequestResults,
        json_result: str,
    ) -> None:
        """Worker fetches the task and validates submitted JSON."""
        runner_execute = self.enterContext(
            mock.patch(
                "debusine.worker.task_runner.TaskRunner.execute",
                return_value=execute_result,
            )
        )

        # No request to any URL...
        self.assertIsNone(self.server_latest_request)

        self.set_events(_worker.WorkRequestAvailableEvent())

        work_request_completed_mocked = (
            self.patch_debusine_work_request_completed_update()
        )

        with self.assertRaisesSystemExit(4):
            await self.worker.main()

        # The WorkRequest is executed...
        self.assertEqual(runner_execute.call_count, 1)

        work_request_completed_mocked.assert_called_once_with(
            mock.ANY, self.work_request.id, json_result, mock.ANY
        )

        output_data = work_request_completed_mocked.call_args.args[3]
        self.assertIsInstance(output_data, OutputData)
        self.assertIsNotNone(output_data.runtime_statistics)

    async def test_task_is_received_executed_submit_success(self) -> None:
        """Receive a task, execute it and submit success completion."""
        await self.receive_task_submit_result(
            WorkRequestResults.SUCCESS, "success"
        )

    async def test_task_is_received_executed_submit_failure(self) -> None:
        """Receive a task, executes it (failure) and submits failure."""
        await self.receive_task_submit_result(
            WorkRequestResults.FAILURE, "failure"
        )

    async def test_print_no_work_request_available(self) -> None:
        """Worker logs 'No work request available'."""
        self.set_events(_worker.WorkRequestAvailableEvent())
        self.server_config.get_next_for_worker = (
            ServerConfig.GetNextForWorkerResponse.NO_WORK_REQUEST_PENDING
        )

        log_message = 'No work request available'

        with self.assertLogsContains(
            log_message, logger="debusine.worker", level=logging.DEBUG
        ):
            await self.worker.process_notifications()

    async def test_log_work_request_ignored_already_running(self) -> None:
        """Worker receive two Work Request available, fetches only one."""
        wait_until_requesting = asyncio.Event()
        wait_until_condition_tested = asyncio.Event()

        async def _request_work_request() -> None:
            wait_until_requesting.set()
            await wait_until_condition_tested.wait()

        self.enterContext(
            mock.patch.object(
                self.worker,
                "_request_work_request",
                side_effect=_request_work_request,
            )
        )

        with self.assertLogsContains(
            "Waited 0 seconds for the end of the former task",
            logger="debusine.worker",
            level=logging.DEBUG,
        ):
            first_request = asyncio.create_task(
                self.worker._request_and_execute_work_request()
            )
            await wait_until_requesting.wait()
        # The first work request now has the lock

        # Log that a Work Request is available (coming from the server)
        # but the Worker ignores it because the worker is already
        # executing a task
        with self.assertLogsContains(
            "Worker is busy and can't execute a new work request",
            logger="debusine.worker",
        ):
            self.patch_worker("_asyncio_sleep")
            await self.worker._request_and_execute_work_request()

        wait_until_condition_tested.set()
        with self.assertLogsContains(
            "No work request available",
            logger="debusine.worker",
            level=logging.INFO,
        ):
            await first_request

    async def test_shutdown_event_aborts_task(self) -> None:
        # Reduce wait_popen polling interval to make the test faster
        orig_wait_popen = Sbuild._wait_popen

        def fast_wait_popen(
            self: Sbuild,
            popen: subprocess.Popen[AnyStr],
            timeout: float,  # noqa: ARG001
        ) -> int:
            return orig_wait_popen(self, popen, 0.05)

        self.enterContext(
            mock.patch(
                "debusine.tasks.Sbuild._wait_popen",
                side_effect=fast_wait_popen,
                autospec=True,
            )
        )

        loop = asyncio.get_running_loop()
        execute_started = asyncio.Event()
        execute_done = asyncio.Event()

        runner_shutdown = self.enterContext(
            self.spy("debusine.worker.task_runner.TaskRunner.shutdown")
        )

        def mock_execute(self: Sbuild) -> WorkRequestResults:
            nonlocal execute_started
            loop.call_soon_threadsafe(execute_started.set)
            self.run_cmd(["sleep", "inf"], working_directory=Path.cwd())
            loop.call_soon_threadsafe(execute_done.set)
            return WorkRequestResults.SUCCESS

        self.enterContext(
            mock.patch(
                "debusine.tasks.Sbuild.execute_logging_exceptions",
                side_effect=mock_execute,
                autospec=True,
            )
        )

        report_task_result = self.enterContext(
            mock.patch.object(self.worker, "report_task_result")
        )

        self.set_events(
            _worker.WorkRequestAvailableEvent(), include_quit_main_loop=False
        )

        async with asyncio.TaskGroup() as tg:
            capture_build_task = tg.create_task(self.capture_build_task())
            tg.create_task(self.worker.process_notifications())
            await execute_started.wait()
            await self.worker.event_queue.put(
                _worker.SignalReceivedShutdownEvent(signal.SIGTERM)
            )

        build_task = capture_build_task.result()
        self.assertTrue(build_task.cancelled())
        report_task_result.assert_not_called()
        runner_shutdown.assert_called()

        runner: TaskRunner = runner_shutdown.call_args.args[0]
        assert runner.task is not None
        self.assertTrue(runner.task.aborted)

        await execute_done.wait()

    async def test_signal_int_term_handler_schedules_shutdown(self) -> None:
        self.enterContext(mock.patch.object(self.worker, "require_token"))

        self.assertTrue(self.worker.event_queue.empty())
        with self.assertLogsContains(
            "Terminated with signal SIGINT", logger="debusine.worker"
        ):
            await self.worker._signal_int_term_handler(signal.SIGINT)
        self.assertQueuedEvents(["shutdown: signal 2 received"])

    async def assert_invalid_get_next_for_worker_logging(
        self, get_next_for_worker: ServerConfig.GetNextForWorkerResponse
    ) -> None:
        """Assert log_message is logged for get_next_for_worker response."""
        self.set_events(_worker.WorkRequestAvailableEvent())
        self.server_config.get_next_for_worker = get_next_for_worker

        assert self.worker is not None

        with self.assertLogsContains(
            "Invalid WorkRequest received from",
            logger="debusine.worker",
            level=logging.WARNING,
        ):
            await self.worker.process_notifications()

    async def test_request_work_request_invalid_work_request_response(
        self,
    ) -> None:
        """Worker logs 'Invalid content of...'."""
        await self.assert_invalid_get_next_for_worker_logging(
            ServerConfig.GetNextForWorkerResponse.INVALID_WORK_REQUEST
        )

    async def test_request_work_request_invalid_json_response(self) -> None:
        """Worker logs 'Invalid JSON response...'."""
        await self.assert_invalid_get_next_for_worker_logging(
            ServerConfig.GetNextForWorkerResponse.INVALID_JSON
        )

    async def test_request_work_request_retries(self) -> None:
        """_request_work_request retires failed requests."""
        self.server_config.get_next_for_worker = Response(
            status=HTTPServiceUnavailable.status_code
        )

        def sleep(delay: float) -> None:  # noqa: ARG001, U100
            self.server_config.get_next_for_worker = (
                ServerConfig.GetNextForWorkerResponse.VALID_WORK_REQUEST
            )

        self.patch_worker("_asyncio_sleep", side_effect=sleep)
        response = await self.worker._request_work_request()
        self.assertIsNotNone(response)

    async def assert_execute_work_request_send_and_log_error(
        self,
        error_msg: str,
        error_code: str,
        worker_type: Literal[
            WorkerType.EXTERNAL, WorkerType.SIGNING
        ] = WorkerType.EXTERNAL,
    ) -> None:
        """Assert error is sent to the server and logged."""
        self.worker._worker_type = worker_type

        work_request_completed_mocked = (
            self.patch_debusine_work_request_completed_update()
        )

        with self.assertLogsContains(error_msg, logger="debusine.worker"):
            await self.worker._request_and_execute_work_request()

        work_request_completed_mocked.assert_called_once_with(
            mock.ANY, mock.ANY, "error", mock.ANY
        )
        _, work_request_id, _, output_data = (
            work_request_completed_mocked.call_args[0]
        )
        self.assertIsInstance(work_request_id, int)
        self.assertIsInstance(output_data, OutputData)
        self.assertEqual(
            output_data.errors,
            [OutputDataError(message=error_msg, code=error_code)],
        )

    async def test_task_configure_error_submit_error(self) -> None:
        """Sbuild.configure fails, Worker submits error."""
        self.enterContext(
            mock.patch.object(
                Sbuild,
                '_configure',
                autospec=True,
                side_effect=TaskConfigError('Invalid schema'),
            )
        )
        await self.assert_execute_work_request_send_and_log_error(
            "Task: sbuild Error configure: Invalid schema", "configure-failed"
        )

    async def test_task_execution_error_submit_error(self) -> None:
        """Sbuild.execute raises TaskConfigError, Worker submits error."""
        self.enterContext(
            mock.patch.object(
                Sbuild,
                'execute',
                autospec=True,
                side_effect=TaskConfigError('Broken'),
            )
        )

        await self.assert_execute_work_request_send_and_log_error(
            'Task: sbuild Error execute: Broken', "execute-failed"
        )

    def assert_functools_partial_equal(
        self, partial1: functools.partial[Any], partial2: functools.partial[Any]
    ) -> None:
        """Assert two functools.partials are semantically equal."""
        # functools.partial does not implement __eq__
        self.assertEqual(partial1.func, partial2.func)
        self.assertEqual(partial1.args, partial2.args)
        self.assertEqual(partial1.keywords, partial2.keywords)

    async def test_main(self) -> None:
        require_token = self.patch_worker("require_token")
        process_notifications = self.patch_worker(
            "process_notifications", return_value=None
        )

        mocked_add_signal_handler = self.enterContext(
            mock.patch.object(
                asyncio.get_running_loop(),
                "add_signal_handler",
                autospec=True,
            )
        )

        with (
            self.assertLogsContains(
                "Debusine worker is starting. Version:",
                logger="debusine.worker",
                level=logging.INFO,
            ),
            self.assertRaisesSystemExit(4),
        ):
            await self.worker.main()

        sigint_call = mocked_add_signal_handler.mock_calls[0][1]
        self.assertEqual(sigint_call[0], signal.SIGINT)
        self.assert_functools_partial_equal(
            sigint_call[1],
            functools.partial(
                self.worker._create_task_signal_int_term_handler, signal.SIGINT
            ),
        )

        sigterm_call = mocked_add_signal_handler.mock_calls[1][1]
        self.assertEqual(sigterm_call[0], signal.SIGTERM)
        self.assert_functools_partial_equal(
            sigterm_call[1],
            functools.partial(
                self.worker._create_task_signal_int_term_handler, signal.SIGTERM
            ),
        )

        self.assertEqual(len(mocked_add_signal_handler.mock_calls), 2)

        require_token.assert_awaited()
        process_notifications.assert_awaited()

    async def test_report_task_result_network_error(self) -> None:
        work_request_completed_mocked = (
            self.patch_debusine_work_request_completed_update()
        )
        work_request_completed_mocked.side_effect = Exception("network error")

        async with TaskRunner(
            self.worker._config,
            self.worker._worker_type,
            self.work_request,
            self.worker._workdir,
        ) as runner:
            with (
                self.assertLogsContains(
                    "Cannot reach server",
                    logger="debusine.worker",
                    level=logging.ERROR,
                ),
            ):
                await self.worker.report_task_result(
                    runner,
                    WorkRequestResults.ERROR,
                    OutputData(
                        errors=[
                            OutputDataError(
                                message="Expected error", code="expected-error"
                            )
                        ]
                    ),
                )

        self.assertQueuedEvents(["shutdown: cannot send results to server"])

    async def test_report_task_result_do_not_send_aborted(self) -> None:
        work_request_completed_update = (
            self.patch_debusine_work_request_completed_update()
        )

        async with TaskRunner(
            self.worker._config,
            self.worker._worker_type,
            self.work_request,
            self.worker._workdir,
        ) as runner:
            await runner.setup_task()
            assert runner.task is not None
            runner.task.abort()
            with self.assertLogs(
                logger="debusine.worker", level=logging.INFO
            ) as log:
                await self.worker.report_task_result(
                    runner, WorkRequestResults.ERROR, OutputData()
                )

        self.assertEqual(
            log.output, ["INFO:debusine.worker:Task: sbuild has been aborted"]
        )
        work_request_completed_update.assert_not_called()

    async def test_report_task_result_runs_after_work_request_hook(
        self,
    ) -> None:
        async with TaskRunner(
            self.worker._config,
            self.worker._worker_type,
            self.work_request,
            self.worker._workdir,
        ) as runner:
            await runner.setup_task()
            assert runner.task is not None
            runner.task.abort()

            with self.assertLogs(
                logger="debusine.worker", level=logging.DEBUG
            ) as log:
                await self.worker.report_task_result(
                    runner, WorkRequestResults.ERROR, OutputData()
                )

        self.assertEqual(
            log.output,
            [
                "INFO:debusine.worker:Task: sbuild has been aborted",
                "DEBUG:debusine.worker:"
                "No hooks in '/nonexistent', skipping execution",
            ],
        )

    async def test_run_after_work_request_hook(self) -> None:
        """Run a simple script via _run_after_work_request_hook."""
        hook_dir = Path(self.create_temporary_directory())
        self.worker.AFTER_WORK_REQUEST_HOOK_DIR = hook_dir

        marker = hook_dir / "marker"
        script = hook_dir / "hook"
        script.write_text(
            dedent(
                f"""\
                #!/bin/sh
                touch {marker}
                """
            )
        )
        script.chmod(0o755)
        await self.worker._run_after_work_request_hook()
        self.assertTrue(marker.exists())

    async def test_run_after_work_request_hook_error(self) -> None:
        """Run a simple failing script via _run_after_work_request_hook."""
        hook_dir = Path(self.create_temporary_directory())
        self.worker.AFTER_WORK_REQUEST_HOOK_DIR = hook_dir

        script = hook_dir / "hook"
        script.write_text(
            dedent(
                """\
                #!/bin/sh
                exit 1
                """
            )
        )
        script.chmod(0o755)
        with self.assertLogsContains(
            f"Hooks failed to execute in {str(hook_dir)!r} .\n"
            "stdout:\n"
            "\n"
            "stderr:\n"
            f"run-parts: {hook_dir}/hook exited with return code 1",
            logger="debusine.worker",
            level=logging.WARNING,
        ):
            await self.worker._run_after_work_request_hook()
