supervisor/tests/docker/test_manager.py

"""Test Docker manager."""

from http import HTTPStatus
from pathlib import Path
import re
from unittest.mock import AsyncMock, MagicMock, patch

import aiodocker
from aiodocker.containers import DockerContainer
from aiodocker.networks import DockerNetwork
from awesomeversion import AwesomeVersion
import pytest

from supervisor.const import DNS_SUFFIX, ENV_SUPERVISOR_CPU_RT
from supervisor.coresys import CoreSys
from supervisor.docker.const import (
    LABEL_MANAGED,
    DockerMount,
    MountBindOptions,
    MountType,
)
from supervisor.docker.manager import CommandReturn, DockerAPI, PullLogEntry
from supervisor.exceptions import (
    DockerContainerNotFoundError,
    DockerContainerNotRunningError,
    DockerError,
    DockerNoSpaceOnDevice,
    DockerNotFound,
    DockerRegistryRateLimitExceeded,
    DockerStatsTimeoutError,
    DockerStatsUnknownError,
    DockerTimeoutError,
)


async def test_run_command_success(docker: DockerAPI, container: DockerContainer):
    """Test successful command execution."""
    # Mock container and its methods
    container.wait.return_value = {"StatusCode": 0}
    container.log.return_value = ["command output"]

    # Execute the command
    result = await docker.run_command(
        image="alpine", tag="3.18", command=["echo", "hello"], stdout=True, stderr=True
    )

    # Verify the result
    assert isinstance(result, CommandReturn)
    assert result.exit_code == 0
    assert result.log == ["command output"]

    # Verify docker.containers.run was called correctly
    docker.containers.create.assert_called_once_with(
        {
            "Image": "alpine:3.18",
            "Labels": {"supervisor_managed": ""},
            "OpenStdin": False,
            "StdinOnce": False,
            "AttachStdin": False,
            "AttachStdout": False,
            "AttachStderr": False,
            "HostConfig": {
                "NetworkMode": "hassio",
                "Init": False,
                "Privileged": False,
                "Dns": ["172.30.32.3"],
                "DnsSearch": ["local.hass.io"],
                "DnsOptions": ["timeout:10"],
            },
            "Cmd": ["echo", "hello"],
        },
        name=None,
    )
    container.start.assert_called_once()

    # Verify container cleanup
    container.delete.assert_called_once_with(force=True, v=True)


async def test_run_command_pulls_image_when_not_found(
    docker: DockerAPI, container: DockerContainer
):
    """Test that run_command pulls the image when it doesn't exist locally."""
    # Mock image inspect to raise NOT_FOUND first, then succeed (after pull)
    docker.images.inspect.side_effect = [
        aiodocker.DockerError(HTTPStatus.NOT_FOUND, {"message": "No such image"}),
    ]
    # Mock pull to return successfully (with stream=False, returns a list)
    docker.images.pull = AsyncMock(return_value=[{}])
    container.wait.return_value = {"StatusCode": 0}
    container.log.return_value = ["output"]

    # Execute the command
    result = await docker.run_command(
        image="alpine", tag="3.18", command=["echo", "hello"]
    )

    # Verify pull was called
    docker.images.pull.assert_called_once_with("alpine", tag="3.18")

    # Verify the command still executed successfully
    assert result.exit_code == 0
    assert result.log == ["output"]


async def test_run_command_no_pull_when_image_exists(
    docker: DockerAPI, container: DockerContainer
):
    """Test that run_command doesn't pull if image already exists."""
    # Default mock already returns image data on inspect
    container.wait.return_value = {"StatusCode": 0}
    container.log.return_value = ["output"]

    # Execute the command
    result = await docker.run_command(
        image="alpine", tag="3.18", command=["echo", "hello"]
    )

    # Verify inspect was called but pull was NOT called
    docker.images.inspect.assert_called_once_with("alpine:3.18")
    docker.images.pull.assert_not_called()

    # Verify the command executed successfully
    assert result.exit_code == 0


async def test_run_command_inspect_error_propagates(docker: DockerAPI):
    """Test that non-NOT_FOUND errors from image inspect are propagated."""
    # Mock image inspect to raise a different error (e.g., server error)
    docker.images.inspect.side_effect = aiodocker.DockerError(
        HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "Server error"}
    )

    # Execute the command and expect DockerError
    with pytest.raises(DockerError, match="Can't inspect image alpine:latest"):
        await docker.run_command(image="alpine", command=["echo", "hello"])

    # Verify pull was NOT called since the error wasn't NOT_FOUND
    docker.images.pull.assert_not_called()


async def test_run_command_docker_exception(docker: DockerAPI):
    """Test command execution when Docker raises an exception."""
    # Mock docker containers.run to raise aiodocker.DockerError
    docker.containers.create.side_effect = err = aiodocker.DockerError(
        HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "Docker error"}
    )

    # Execute the command and expect DockerError
    with pytest.raises(
        DockerError,
        match=re.escape(
            f"Can't execute command: Can't create container from alpine:latest: {str(err)}"
        ),
    ):
        await docker.run_command(image="alpine", command="test")


async def test_run_command_cleanup_on_exception(
    docker: DockerAPI, container: DockerContainer
):
    """Test that container cleanup happens even when an exception occurs."""
    container.wait.side_effect = aiodocker.DockerError(500, {"message": "Wait failed"})

    # Execute the command and expect DockerError
    with pytest.raises(DockerError):
        await docker.run_command(image="alpine", command="test")

    # Verify container cleanup still happened
    container.delete.assert_called_once_with(force=True, v=True)


async def test_run_command_custom_stdout_stderr(
    docker: DockerAPI, container: DockerContainer
):
    """Test command execution with custom stdout/stderr settings."""
    # Mock container and its methods
    container.wait.return_value = {"StatusCode": 0}
    container.log.return_value = ["output"]

    # Execute the command with custom stdout/stderr
    result = await docker.run_command(
        image="alpine", command="test", stdout=False, stderr=True
    )

    # Verify container.logs was called with the correct parameters
    container.log.assert_called_once_with(stdout=False, stderr=True, follow=False)

    # Verify the result
    assert result.exit_code == 0
    assert result.log == ["output"]


async def test_run_command_with_mounts(docker: DockerAPI):
    """Test command execution with mounts are correctly converted."""
    # Create test mounts
    mounts = [
        DockerMount(
            type=MountType.BIND,
            source="/dev",
            target="/dev",
            read_only=True,
            bind_options=MountBindOptions(read_only_non_recursive=True),
        ),
        DockerMount(
            type=MountType.VOLUME,
            source="my_volume",
            target="/data",
            read_only=False,
        ),
    ]

    # Execute the command with mounts
    result = await docker.run_command(image="alpine", command="test", mounts=mounts)

    # Verify the result
    assert result.exit_code == 0

    # Check that mounts were converted correctly
    docker.containers.create.assert_called_once_with(
        {
            "Image": "alpine:latest",
            "Labels": {LABEL_MANAGED: ""},
            "OpenStdin": False,
            "StdinOnce": False,
            "AttachStdin": False,
            "AttachStdout": False,
            "AttachStderr": False,
            "Cmd": "test",
            "HostConfig": {
                "NetworkMode": docker.network.name,
                "Init": False,
                "Privileged": False,
                "Dns": [str(docker.network.dns)],
                "DnsSearch": [DNS_SUFFIX],
                "DnsOptions": ["timeout:10"],
                "Mounts": [
                    {
                        "Type": "bind",
                        "Source": "/dev",
                        "Target": "/dev",
                        "ReadOnly": True,
                        "BindOptions": {"ReadOnlyNonRecursive": True},
                    },
                    {
                        "Type": "volume",
                        "Source": "my_volume",
                        "Target": "/data",
                        "ReadOnly": False,
                    },
                ],
            },
        },
        name=None,
    )


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_container_with_cidfile(
    coresys: CoreSys, docker: DockerAPI, container: DockerContainer
):
    """Test container creation with cidfile and bind mount."""
    container.id = "test_container_id_12345"
    container.show.return_value = mock_metadata = {"Id": container.id}

    container_name = "test_container"
    cidfile_path = coresys.config.path_cid_files / f"{container_name}.cid"
    extern_cidfile_path = coresys.config.path_extern_cid_files / f"{container_name}.cid"

    # Execute run with a container name
    result = await docker.run("test_image", tag="latest", name=container_name)

    # Check the container creation parameters
    docker.containers.create.assert_called_once()
    create_config = docker.containers.create.call_args.args[0]

    assert "HostConfig" in create_config
    assert "Mounts" in create_config["HostConfig"]
    assert {
        "Type": "bind",
        "Source": str(extern_cidfile_path),
        "Target": "/run/cid",
        "ReadOnly": True,
    } in create_config["HostConfig"]["Mounts"]

    # Verify container start was called
    container.start.assert_called_once()

    # Verify cidfile was written with container ID
    assert cidfile_path.exists()
    assert cidfile_path.read_text() == container.id

    assert result == mock_metadata


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_container_with_leftover_cidfile(
    coresys: CoreSys, docker: DockerAPI, container: DockerContainer
):
    """Test container creation removes leftover cidfile before creating new one."""
    container.id = "test_container_id_12345"
    container.show.return_value = mock_metadata = {"Id": container.id}

    container_name = "test_container"
    cidfile_path = coresys.config.path_cid_files / f"{container_name}.cid"

    # Create a leftover cidfile
    cidfile_path.touch()

    # Execute run with a container name
    result = await docker.run("test_image", tag="latest", name=container_name)

    # Verify container was created
    docker.containers.create.assert_called_once()

    # Verify new cidfile was written with container ID
    assert cidfile_path.exists()
    assert cidfile_path.read_text() == container.id

    assert result == mock_metadata


@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
async def test_stop_container_with_cidfile_cleanup(
    coresys: CoreSys, docker: DockerAPI, container: DockerContainer
):
    """Test container stop with cidfile cleanup."""
    container_name = "test_container"
    cidfile_path = coresys.config.path_cid_files / f"{container_name}.cid"

    # Create a cidfile
    cidfile_path.touch()

    # Call stop_container with remove_container=True
    await docker.stop_container(timeout=10, remove_container=True, name=container_name)

    # Verify container operations
    container.stop.assert_called_once_with(t=10)
    container.delete.assert_called_once_with(force=True, v=True)

    assert not cidfile_path.exists()


async def test_stop_container_without_removal_no_cidfile_cleanup(
    docker: DockerAPI, container: DockerContainer
):
    """Test container stop without removal doesn't clean up cidfile."""
    container_name = "test_container"

    # Mock the containers.get method and cidfile cleanup
    with patch("pathlib.Path.unlink") as mock_unlink:
        # Call stop_container with remove_container=False
        await docker.stop_container(container_name, timeout=10, remove_container=False)

        # Verify container operations
        container.stop.assert_called_once_with(t=10)
        container.delete.assert_not_called()

        # Verify cidfile cleanup was NOT called
        mock_unlink.assert_not_called()


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_cidfile_cleanup_handles_oserror(
    coresys: CoreSys, docker: DockerAPI, container: DockerContainer
):
    """Test that cidfile cleanup handles OSError gracefully."""
    container_name = "test_container"
    cidfile_path = coresys.config.path_cid_files / f"{container_name}.cid"

    # Create a cidfile
    cidfile_path.touch()

    # Mock the containers.get method and cidfile cleanup to raise OSError
    with (
        patch("pathlib.Path.is_dir", return_value=False),
        patch("pathlib.Path.is_file", return_value=True),
        patch(
            "pathlib.Path.unlink", side_effect=OSError("File not found")
        ) as mock_unlink,
    ):
        # Call stop_container - should not raise exception
        await docker.stop_container(container_name, timeout=10, remove_container=True)

        # Verify container operations completed
        container.stop.assert_called_once_with(t=10)
        container.delete.assert_called_once_with(force=True, v=True)

        # Verify cidfile cleanup was attempted
        mock_unlink.assert_called_once_with(missing_ok=True)


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_container_with_leftover_cidfile_directory(
    coresys: CoreSys, docker: DockerAPI, container: DockerContainer
):
    """Test container creation removes leftover cidfile directory before creating new one.

    This can happen when Docker auto-starts a container with restart policy
    before Supervisor could write the CID file, causing Docker to create
    the bind mount source as a directory.
    """
    container.id = "test_container_id_12345"
    container.show.return_value = mock_metadata = {"Id": container.id}

    container_name = "test_container"
    cidfile_path = coresys.config.path_cid_files / f"{container_name}.cid"

    # Create a leftover directory (simulating Docker's behavior)
    cidfile_path.mkdir()
    assert cidfile_path.is_dir()

    # Execute run with a container name
    result = await docker.run("test_image", tag="latest", name=container_name)

    # Verify container was created
    docker.containers.create.assert_called_once()

    # Verify new cidfile was written as a file (not directory)
    assert cidfile_path.exists()
    assert cidfile_path.is_file()
    assert cidfile_path.read_text() == container.id

    assert result == mock_metadata


async def test_repair(
    coresys: CoreSys, caplog: pytest.LogCaptureFixture, container: DockerContainer
):
    """Test repair API."""
    coresys.docker.docker.networks.get.side_effect = [
        hassio := MagicMock(spec=DockerNetwork),
        host := MagicMock(spec=DockerNetwork),
    ]
    hassio.show.return_value = {
        "Containers": {
            "good": {"Name": "good"},
            "corrupt": {"Name": "corrupt"},
            "fail": {"Name": "fail"},
        }
    }
    host.show.return_value = {"Containers": {}}
    coresys.docker.containers.get.side_effect = [
        container,
        aiodocker.DockerError(HTTPStatus.NOT_FOUND, {"message": "corrupt"}),
        aiodocker.DockerError(HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "fail"}),
    ]

    await coresys.docker.repair()

    coresys.docker.docker.containers.prune.assert_called_once()
    coresys.docker.docker.images.prune.assert_called_once_with(
        filters={"dangling": "false"}
    )
    coresys.docker.docker.images.prune_builds.assert_called_once()
    coresys.docker.docker.volumes.prune.assert_called_once()
    coresys.docker.docker.networks.prune.assert_called_once()
    hassio.disconnect.assert_called_once_with({"Container": "corrupt", "Force": True})
    host.disconnect.assert_not_called()
    assert "Docker fatal error on container fail on hassio" in caplog.text


async def test_repair_failures(coresys: CoreSys, caplog: pytest.LogCaptureFixture):
    """Test repair proceeds best it can through failures."""
    fail_err = aiodocker.DockerError(
        HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "fail"}
    )
    coresys.docker.docker.containers.prune.side_effect = fail_err
    coresys.docker.docker.images.prune.side_effect = fail_err
    coresys.docker.docker.images.prune_builds.side_effect = fail_err
    coresys.docker.docker.volumes.prune.side_effect = fail_err
    coresys.docker.docker.networks.prune.side_effect = fail_err
    coresys.docker.docker.networks.get.side_effect = missing_err = (
        aiodocker.DockerError(HTTPStatus.NOT_FOUND, {"message": "missing"})
    )

    await coresys.docker.repair()

    assert f"Error for containers prune: {fail_err!s}" in caplog.text
    assert f"Error for images prune: {fail_err!s}" in caplog.text
    assert f"Error for builds prune: {fail_err!s}" in caplog.text
    assert f"Error for volumes prune: {fail_err!s}" in caplog.text
    assert f"Error for networks prune: {fail_err!s}" in caplog.text
    assert f"Error for networks hassio prune: {missing_err!s}" in caplog.text
    assert f"Error for networks host prune: {missing_err!s}" in caplog.text


@pytest.mark.parametrize("log_starter", [("Loaded image ID"), ("Loaded image")])
async def test_import_image(coresys: CoreSys, tmp_path: Path, log_starter: str):
    """Test importing an image into docker."""
    (test_tar := tmp_path / "test.tar").touch()
    coresys.docker.images.import_image = AsyncMock(
        return_value=[{"stream": f"{log_starter}: imported"}]
    )
    coresys.docker.images.inspect.return_value = {"Id": "imported"}

    image = await coresys.docker.import_image(test_tar)

    assert image["Id"] == "imported"
    coresys.docker.images.inspect.assert_called_once_with("imported")


async def test_import_image_error(coresys: CoreSys, tmp_path: Path):
    """Test failure importing an image into docker."""
    (test_tar := tmp_path / "test.tar").touch()
    coresys.docker.images.import_image = AsyncMock(
        return_value=[{"errorDetail": {"message": "fail"}}]
    )

    with pytest.raises(DockerError, match="Can't import image from tar: fail"):
        await coresys.docker.import_image(test_tar)

    coresys.docker.images.inspect.assert_not_called()


async def test_import_multiple_images_in_tar(
    coresys: CoreSys, tmp_path: Path, caplog: pytest.LogCaptureFixture
):
    """Test importing an image into docker."""
    (test_tar := tmp_path / "test.tar").touch()
    coresys.docker.images.import_image = AsyncMock(
        return_value=[
            {"stream": "Loaded image: imported-1"},
            {"stream": "Loaded image: imported-2"},
        ]
    )

    assert await coresys.docker.import_image(test_tar) is None

    assert "Unexpected image count 2 while importing image from tar" in caplog.text
    coresys.docker.images.inspect.assert_not_called()


@pytest.mark.parametrize(
    ("rt_file_exists", "rt_env", "rt_supported"),
    [(False, "1", False), (True, "0", False), (True, "1", True)],
)
async def test_info(
    monkeypatch: pytest.MonkeyPatch,
    docker: DockerAPI,
    rt_file_exists: bool,
    rt_env: str,
    rt_supported: bool,
):
    """Test docker system info."""
    docker.docker.system.info.return_value = {
        "ServerVersion": "2.0.0",
        "Driver": "example",
        "LoggingDriver": "example",
        "CgroupVersion": "2",
    }
    monkeypatch.setenv(ENV_SUPERVISOR_CPU_RT, rt_env)
    with patch("supervisor.docker.manager.Path.exists", return_value=rt_file_exists):
        await docker.post_init()
    assert docker.info.version == AwesomeVersion("2.0.0")
    assert docker.info.storage == "example"
    assert docker.info.logging == "example"
    assert docker.info.cgroup == "2"
    assert docker.info.support_cpu_realtime is rt_supported


@pytest.mark.parametrize(
    "error_message",
    [
        "toomanyrequests: retry-after: 1.265943ms, allowed: 44000/minute",
        "error pulling image configuration: download failed after attempts=1: "
        "toomanyrequests: retry-after: 358.998µs, allowed: 44000/minute",
    ],
)
def test_pull_log_entry_exception_detects_rate_limit(error_message: str):
    """Test PullLogEntry detects toomanyrequests in stream errors as rate limit.

    The streaming pull protocol does not preserve HTTP status codes - rate
    limit errors only surface as text. Cover both the bare and wrapped forms
    seen in production Sentry data.
    """
    entry = PullLogEntry(job_id="x", error=error_message)
    assert isinstance(entry.exception, DockerRegistryRateLimitExceeded)


def test_pull_log_entry_exception_no_space_left():
    """Test PullLogEntry returns DockerNoSpaceOnDevice for disk-full errors."""
    entry = PullLogEntry(
        job_id="x", error="failed to register layer: no space left on device"
    )
    assert isinstance(entry.exception, DockerNoSpaceOnDevice)


def test_pull_log_entry_exception_generic_error():
    """Test PullLogEntry returns plain DockerError for other errors."""
    entry = PullLogEntry(job_id="x", error="something else broke")
    err = entry.exception
    assert isinstance(err, DockerError)
    assert not isinstance(err, DockerRegistryRateLimitExceeded)
    assert not isinstance(err, DockerNoSpaceOnDevice)


# ---------------------------------------------------------------------------
# Timeout error path tests
# ---------------------------------------------------------------------------


async def test_post_init_system_info_timeout(docker: DockerAPI):
    """Test post_init raises DockerTimeoutError when system.info times out."""
    docker.docker.system.info.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout while getting Docker information"
    ):
        await docker.post_init()


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_container_create_timeout(coresys: CoreSys):
    """Test run raises DockerTimeoutError when containers.create times out."""
    coresys.docker.containers.create.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout creating container from alpine:latest"
    ):
        await coresys.docker.run("alpine", name="test", tag="latest")


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_container_start_timeout(coresys: CoreSys):
    """Test run raises DockerTimeoutError when container.start times out."""
    coresys.docker.containers.create.return_value.start.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout starting"):
        await coresys.docker.run("alpine", name="test", tag="latest")


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_attach_network_failure_logs_warning(
    coresys: CoreSys, caplog: pytest.LogCaptureFixture
):
    """Test run logs warning and continues when attach to hassio-network fails."""
    with (
        patch.object(
            coresys.docker.network,
            "attach_container",
            new=AsyncMock(side_effect=DockerError("failed")),
        ),
        patch.object(
            coresys.docker.network,
            "detach_default_bridge",
            new=AsyncMock(),
        ) as detach_default_bridge,
    ):
        await coresys.docker.run("alpine", name="test", tag="latest")

    assert "Can't attach test to hassio-network!" in caplog.text
    detach_default_bridge.assert_not_called()


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_host_network_get_timeout(coresys: CoreSys):
    """Test run raises DockerTimeoutError when loading host network times out."""
    coresys.docker.docker.networks.get.side_effect = TimeoutError()

    with pytest.raises(
        DockerTimeoutError, match="Timeout getting host network information"
    ):
        await coresys.docker.run(
            "alpine", name="test", tag="latest", network_mode="host"
        )


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_host_network_get_docker_error(coresys: CoreSys):
    """Test run raises DockerError when loading host network fails."""
    coresys.docker.docker.networks.get.side_effect = aiodocker.DockerError(
        HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "fail"}
    )

    with pytest.raises(
        DockerError, match="Can't get host network information from Docker"
    ):
        await coresys.docker.run(
            "alpine", name="test", tag="latest", network_mode="host"
        )


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_host_network_disconnect_timeout(coresys: CoreSys):
    """Test run raises DockerTimeoutError when disconnect from host network times out."""
    host_network = MagicMock(spec=DockerNetwork)
    host_network.show.return_value = {"Containers": {"abc": {"Name": "test"}}}
    host_network.disconnect.side_effect = TimeoutError()
    coresys.docker.docker.networks.get.return_value = host_network

    with pytest.raises(
        DockerTimeoutError,
        match="Timeout disconnecting container test from host network",
    ):
        await coresys.docker.run(
            "alpine", name="test", tag="latest", network_mode="host"
        )


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_host_network_disconnect_non_not_found_error(coresys: CoreSys):
    """Test run raises DockerError when host network disconnect fails with non-404."""
    host_network = MagicMock(spec=DockerNetwork)
    host_network.show.return_value = {"Containers": {"abc": {"Name": "test"}}}
    host_network.disconnect.side_effect = aiodocker.DockerError(
        HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "fail"}
    )
    coresys.docker.docker.networks.get.return_value = host_network

    with pytest.raises(
        DockerError, match="Can't disconnect container test from host network"
    ):
        await coresys.docker.run(
            "alpine", name="test", tag="latest", network_mode="host"
        )


@pytest.mark.usefixtures("path_extern", "tmp_supervisor_data")
async def test_run_metadata_timeout(coresys: CoreSys, container: DockerContainer):
    """Test run() raises DockerTimeoutError when container.show times out after start."""
    container.show.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout inspecting started container"
    ):
        await coresys.docker.run("alpine", name="test", tag="latest")


async def test_run_command_image_inspect_timeout(docker: DockerAPI):
    """Test run_command raises DockerTimeoutError when image inspect times out."""
    docker.images.inspect.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout inspecting image"):
        await docker.run_command(image="alpine", command=["echo", "hi"])


async def test_repair_timeouts(
    coresys: CoreSys, caplog: pytest.LogCaptureFixture, container: DockerContainer
):
    """Test repair logs warnings and continues when prune operations time out."""
    coresys.docker.docker.containers.prune.side_effect = TimeoutError()
    coresys.docker.docker.images.prune.side_effect = TimeoutError()
    coresys.docker.docker.images.prune_builds.side_effect = TimeoutError()
    coresys.docker.docker.volumes.prune.side_effect = TimeoutError()
    coresys.docker.docker.networks.prune.side_effect = TimeoutError()
    # prune_networks is called next; make networks.get raise DockerError to keep it short
    coresys.docker.docker.networks.get.side_effect = aiodocker.DockerError(
        HTTPStatus.NOT_FOUND, {"message": "missing"}
    )

    await coresys.docker.repair()

    assert "Error for containers prune: timed out" in caplog.text
    assert "Error for images prune: timed out" in caplog.text
    assert "Error for builds prune: timed out" in caplog.text
    assert "Error for volumes prune: timed out" in caplog.text
    assert "Error for networks prune: timed out" in caplog.text


async def test_prune_networks_get_timeout(
    coresys: CoreSys, caplog: pytest.LogCaptureFixture
):
    """Test prune_networks raises DockerTimeoutError when networks.get/show times out."""
    coresys.docker.docker.networks.get.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout loading network metadata"):
        await coresys.docker.prune_networks("hassio")


async def test_prune_networks_container_get_timeout(
    coresys: CoreSys, caplog: pytest.LogCaptureFixture
):
    """Test prune_networks raises DockerTimeoutError when containers.get times out."""
    network = MagicMock(spec=DockerNetwork)
    network.show.return_value = {"Containers": {"abc123": {"Name": "test"}}}
    coresys.docker.docker.networks.get.return_value = network
    coresys.docker.containers.get.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout checking container abc123"):
        await coresys.docker.prune_networks("hassio")


async def test_container_is_initialized_timeout(
    docker: DockerAPI, container: DockerContainer
):
    """Test container_is_initialized raises DockerTimeoutError when show times out."""
    container.show.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout getting container"):
        await docker.container_is_initialized(
            "mycontainer", "myimage", AwesomeVersion("1.0")
        )


async def test_stop_container_get_timeout(
    docker: DockerAPI, container: DockerContainer
):
    """Test stop_container raises DockerTimeoutError when stop times out."""
    container.stop.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout stopping container"):
        await docker.stop_container("mycontainer", timeout=10)


async def test_stop_container_not_found(docker: DockerAPI, container: DockerContainer):
    """Test stop_container raises DockerNotFound when container doesn't exist."""
    container.stop.side_effect = aiodocker.DockerError(404, {"message": "missing"})
    with pytest.raises(DockerNotFound):
        await docker.stop_container("mycontainer", timeout=10)

    container.delete.assert_not_called()


async def test_stop_container_already_stopped(
    docker: DockerAPI, container: DockerContainer
):
    """Test stop_container treats a 304 (already stopped) as success, not an error."""
    # Docker returns 304 when the container is already stopped. aiodocker only
    # raises DockerError for 4xx/5xx responses, so this surfaces as a normal
    # return from stop() rather than an exception.
    container.stop.return_value = None

    await docker.stop_container("mycontainer", timeout=10, remove_container=False)

    container.stop.assert_called_once_with(t=10)
    container.delete.assert_not_called()


async def test_start_container_get_timeout(docker: DockerAPI):
    """Test start_container raises DockerTimeoutError when containers.get times out."""
    docker.containers.get.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout getting .* for starting up"):
        await docker.start_container("mycontainer")


async def test_start_container_start_timeout(
    docker: DockerAPI, container: DockerContainer
):
    """Test start_container raises DockerTimeoutError when container.start times out."""
    container.start.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout starting mycontainer"):
        await docker.start_container("mycontainer")


async def test_restart_container_get_timeout(docker: DockerAPI):
    """Test restart_container raises DockerTimeoutError when containers.get times out."""
    docker.containers.get.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout getting container .* for restarting"
    ):
        await docker.restart_container("mycontainer", timeout=10)


async def test_container_logs_get_timeout(docker: DockerAPI):
    """Test container_logs raises DockerTimeoutError when containers.get times out."""
    docker.containers.get.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout getting container .* for logs"
    ):
        await docker.container_logs("mycontainer")


async def test_container_stats_not_found(docker: DockerAPI):
    """Test container_stats raises DockerContainerNotFoundError when container doesn't exist."""
    docker.containers.container.return_value.stats.side_effect = aiodocker.DockerError(
        HTTPStatus.NOT_FOUND, {"message": "not found"}
    )
    with pytest.raises(DockerContainerNotFoundError, match="not found"):
        await docker.container_stats("mycontainer")


async def test_container_stats_not_running(
    docker: DockerAPI, container: DockerContainer
):
    """Test container_stats raises DockerContainerNotRunningError when stopped.

    Docker returns a stub response containing only id/name (no online_cpus in
    cpu_stats) for a stopped or restarting container instead of an error, same
    as in one-shot mode.
    """
    stub_response = {
        "id": "abc123",
        "name": "/mycontainer",
        "cpu_stats": {"cpu_usage": {"total_usage": 0}},
        "memory_stats": {},
    }
    container.stats = AsyncMock(return_value=[stub_response])
    with pytest.raises(DockerContainerNotRunningError, match="is not running"):
        await docker.container_stats("mycontainer")


async def test_container_stats_timeout(docker: DockerAPI, container: DockerContainer):
    """Test container_stats raises DockerStatsTimeoutError when the stats call times out."""
    container.stats.side_effect = TimeoutError()
    with pytest.raises(
        DockerStatsTimeoutError, match="Timed out getting stats for container"
    ):
        await docker.container_stats("mycontainer")


async def test_container_stats_unknown_error(
    docker: DockerAPI, container: DockerContainer
):
    """Test container_stats raises DockerStatsUnknownError on unexpected stats error."""
    container.stats.side_effect = aiodocker.DockerError(
        HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "boom"}
    )
    with pytest.raises(DockerStatsUnknownError, match="unknown error"):
        await docker.container_stats("mycontainer")


async def test_container_stats_one_shot(docker: DockerAPI, container: DockerContainer):
    """Test container_stats requests an immediate, un-windowed sample without inspecting the container."""
    stats_payload = {"cpu_stats": {"cpu_usage": {"total_usage": 123}, "online_cpus": 4}}
    docker.containers.get.reset_mock()
    docker.containers.container.reset_mock()
    container.show.reset_mock()

    with patch.object(
        DockerAPI,
        "_query_one_shot_stats",
        AsyncMock(return_value=stats_payload),
    ) as query_one_shot_stats:
        result = await docker.container_stats("mycontainer", one_shot=True)

    assert result == stats_payload
    query_one_shot_stats.assert_called_once_with("mycontainer")
    docker.containers.get.assert_not_called()
    docker.containers.container.assert_not_called()
    container.show.assert_not_called()
    container.stats.assert_not_called()


async def test_container_stats_one_shot_timeout(docker: DockerAPI):
    """Test container_stats one-shot raises DockerStatsTimeoutError on timeout."""
    with (
        patch.object(
            DockerAPI, "_query_one_shot_stats", AsyncMock(side_effect=TimeoutError())
        ),
        pytest.raises(
            DockerStatsTimeoutError, match="Timed out getting stats for container"
        ),
    ):
        await docker.container_stats("mycontainer", one_shot=True)


async def test_container_stats_one_shot_not_found(docker: DockerAPI):
    """Test container_stats one-shot raises DockerContainerNotFoundError when container doesn't exist."""
    with (
        patch.object(
            DockerAPI,
            "_query_one_shot_stats",
            AsyncMock(
                side_effect=aiodocker.DockerError(
                    HTTPStatus.NOT_FOUND, {"message": "gone"}
                )
            ),
        ),
        pytest.raises(DockerContainerNotFoundError, match="not found"),
    ):
        await docker.container_stats("mycontainer", one_shot=True)


async def test_container_stats_one_shot_unknown_error(docker: DockerAPI):
    """Test container_stats one-shot raises DockerStatsUnknownError on unexpected error."""
    with (
        patch.object(
            DockerAPI,
            "_query_one_shot_stats",
            AsyncMock(
                side_effect=aiodocker.DockerError(
                    HTTPStatus.INTERNAL_SERVER_ERROR, {"message": "boom"}
                )
            ),
        ),
        pytest.raises(DockerStatsUnknownError, match="unknown error"),
    ):
        await docker.container_stats("mycontainer", one_shot=True)


async def test_container_stats_one_shot_not_running(docker: DockerAPI):
    """Test container_stats one-shot raises DockerContainerNotRunningError for a stopped container.

    Docker returns a stub response containing only id/name (no online_cpus in
    cpu_stats, no memory_stats.usage, no networks) instead of an error for a
    stopped or restarting container, even in one-shot mode.
    """
    stub_response = {
        "id": "abc123",
        "name": "/mycontainer",
        "cpu_stats": {"cpu_usage": {"total_usage": 0}},
        "memory_stats": {},
    }
    with (
        patch.object(
            DockerAPI,
            "_query_one_shot_stats",
            AsyncMock(return_value=stub_response),
        ),
        pytest.raises(DockerContainerNotRunningError, match="is not running"),
    ):
        await docker.container_stats("mycontainer", one_shot=True)


async def test_container_stats_one_shot_empty_response(docker: DockerAPI):
    """Test container_stats one-shot raises DockerStatsUnknownError when Docker returns nothing."""
    with (
        patch.object(
            DockerAPI,
            "_query_one_shot_stats",
            AsyncMock(return_value=None),
        ),
        pytest.raises(DockerStatsUnknownError, match="unknown error"),
    ):
        await docker.container_stats("mycontainer", one_shot=True)


async def test_container_stats_empty_response(
    docker: DockerAPI, container: DockerContainer
):
    """Test container_stats raises DockerStatsUnknownError when Docker returns no samples."""
    container.stats = AsyncMock(return_value=[])
    with pytest.raises(DockerStatsUnknownError, match="unknown error"):
        await docker.container_stats("mycontainer")


async def test_query_one_shot_stats(docker: DockerAPI):
    """Test _query_one_shot_stats queries Docker directly for a one-shot sample."""
    stats_payload = {"cpu_stats": {"online_cpus": 4}}

    class MockResponse:
        async def json(self, *, content_type=None):
            return stats_payload

    class MockQueryCM:
        async def __aenter__(self):
            return MockResponse()

        async def __aexit__(self, exc_type, exc, tb):
            return None

    query_mock = MagicMock(return_value=MockQueryCM())
    docker.docker._query = query_mock  # pylint: disable=protected-access

    result = await docker._query_one_shot_stats(  # pylint: disable=protected-access
        "mycontainer"
    )

    assert result == stats_payload
    query_mock.assert_called_once_with(
        "containers/mycontainer/stats",
        params={"stream": "0", "one-shot": "1"},
    )


async def test_container_run_inside_get_timeout(
    docker: DockerAPI, container: DockerContainer
):
    """Test container_run_inside raises DockerTimeoutError when exec times out."""
    container.exec.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout running command in container"
    ):
        await docker.container_run_inside("mycontainer", "echo hi")


async def test_container_run_inside_exec_timeout(
    docker: DockerAPI, container: DockerContainer
):
    """Test container_run_inside raises DockerTimeoutError when exec stream times out."""
    container.exec.side_effect = TimeoutError()
    with pytest.raises(
        DockerTimeoutError, match="Timeout running command in container"
    ):
        await docker.container_run_inside("mycontainer", "echo hi")


async def test_remove_image_latest_timeout(docker: DockerAPI):
    """Test remove_image raises DockerTimeoutError when deleting latest tag times out."""
    docker.images.delete.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout removing image .*:latest"):
        await docker.remove_image("myimage", AwesomeVersion("1.0"), latest=True)


async def test_remove_image_version_timeout(docker: DockerAPI):
    """Test remove_image raises DockerTimeoutError when deleting version tag times out."""
    # latest delete succeeds (raises NOT_FOUND which is suppressed), version times out
    docker.images.delete.side_effect = [
        aiodocker.DockerError(HTTPStatus.NOT_FOUND, {"message": "not found"}),
        TimeoutError(),
    ]
    with pytest.raises(DockerTimeoutError, match="Timeout removing image .*:1.0"):
        await docker.remove_image("myimage", AwesomeVersion("1.0"), latest=True)


async def test_cleanup_old_images_inspect_timeout(docker: DockerAPI):
    """Test cleanup_old_images raises DockerTimeoutError when inspect times out."""
    docker.images.inspect.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout getting myimage for cleanup"):
        await docker.cleanup_old_images("myimage", AwesomeVersion("1.0"))


async def test_cleanup_old_images_list_timeout(docker: DockerAPI):
    """Test cleanup_old_images raises DockerTimeoutError when image list times out."""
    docker.images.inspect.return_value = {"Id": "abc123"}
    docker.images.list.side_effect = TimeoutError()
    with pytest.raises(DockerTimeoutError, match="Timeout listing images for cleanup"):
        await docker.cleanup_old_images("myimage", AwesomeVersion("1.0"))