supervisor/tests/docker/test_interface.py

"""Test Docker interface."""
from typing import Any
from unittest.mock import MagicMock, Mock, PropertyMock, call, patch

from awesomeversion import AwesomeVersion
from docker.errors import DockerException, NotFound
from docker.models.containers import Container
from docker.models.images import Image
import pytest
from requests import RequestException

from supervisor.const import BusEvent, CpuArch
from supervisor.coresys import CoreSys
from supervisor.docker.const import ContainerState
from supervisor.docker.interface import DockerInterface
from supervisor.docker.monitor import DockerContainerStateEvent
from supervisor.exceptions import DockerAPIError, DockerError, DockerRequestError


@pytest.fixture(autouse=True)
def mock_verify_content(coresys: CoreSys):
    """Mock verify_content utility during tests."""
    with patch.object(
        coresys.security, "verify_content", return_value=None
    ) as verify_content:
        yield verify_content


@pytest.mark.parametrize(
    "cpu_arch, platform",
    [
        (CpuArch.ARMV7, "linux/arm/v7"),
        (CpuArch.ARMHF, "linux/arm/v6"),
        (CpuArch.AARCH64, "linux/arm64"),
        (CpuArch.I386, "linux/386"),
        (CpuArch.AMD64, "linux/amd64"),
    ],
)
async def test_docker_image_platform(coresys: CoreSys, cpu_arch: str, platform: str):
    """Test platform set correctly from arch."""
    with patch.object(
        coresys.docker.images, "pull", return_value=Mock(id="test:1.2.3")
    ) as pull:
        instance = DockerInterface(coresys)
        await instance.install(AwesomeVersion("1.2.3"), "test", arch=cpu_arch)
        assert pull.call_count == 1
        assert pull.call_args == call("test:1.2.3", platform=platform)


async def test_docker_image_default_platform(coresys: CoreSys):
    """Test platform set using supervisor arch when omitted."""
    with patch.object(
        type(coresys.supervisor), "arch", PropertyMock(return_value="i386")
    ), patch.object(
        coresys.docker.images, "pull", return_value=Mock(id="test:1.2.3")
    ) as pull:
        instance = DockerInterface(coresys)
        await instance.install(AwesomeVersion("1.2.3"), "test")
        assert pull.call_count == 1
        assert pull.call_args == call("test:1.2.3", platform="linux/386")


@pytest.mark.parametrize(
    "attrs,expected",
    [
        ({"State": {"Status": "running"}}, ContainerState.RUNNING),
        ({"State": {"Status": "exited", "ExitCode": 0}}, ContainerState.STOPPED),
        ({"State": {"Status": "exited", "ExitCode": 137}}, ContainerState.FAILED),
        (
            {"State": {"Status": "running", "Health": {"Status": "healthy"}}},
            ContainerState.HEALTHY,
        ),
        (
            {"State": {"Status": "running", "Health": {"Status": "unhealthy"}}},
            ContainerState.UNHEALTHY,
        ),
    ],
)
async def test_current_state(
    coresys: CoreSys, attrs: dict[str, Any], expected: ContainerState
):
    """Test current state for container."""
    container_collection = MagicMock()
    container_collection.get.return_value = Container(attrs)
    with patch(
        "supervisor.docker.manager.DockerAPI.containers",
        new=PropertyMock(return_value=container_collection),
    ):
        assert await coresys.homeassistant.core.instance.current_state() == expected


async def test_current_state_failures(coresys: CoreSys):
    """Test failure states for current state."""
    container_collection = MagicMock()
    with patch(
        "supervisor.docker.manager.DockerAPI.containers",
        new=PropertyMock(return_value=container_collection),
    ):
        container_collection.get.side_effect = NotFound("dne")
        assert (
            await coresys.homeassistant.core.instance.current_state()
            == ContainerState.UNKNOWN
        )

        container_collection.get.side_effect = DockerException()
        with pytest.raises(DockerAPIError):
            await coresys.homeassistant.core.instance.current_state()

        container_collection.get.side_effect = RequestException()
        with pytest.raises(DockerRequestError):
            await coresys.homeassistant.core.instance.current_state()


@pytest.mark.parametrize(
    "attrs,expected,fired_when_skip_down",
    [
        ({"State": {"Status": "running"}}, ContainerState.RUNNING, True),
        ({"State": {"Status": "exited", "ExitCode": 0}}, ContainerState.STOPPED, False),
        (
            {"State": {"Status": "exited", "ExitCode": 137}},
            ContainerState.FAILED,
            False,
        ),
        (
            {"State": {"Status": "running", "Health": {"Status": "healthy"}}},
            ContainerState.HEALTHY,
            True,
        ),
        (
            {"State": {"Status": "running", "Health": {"Status": "unhealthy"}}},
            ContainerState.UNHEALTHY,
            True,
        ),
    ],
)
async def test_attach_existing_container(
    coresys: CoreSys,
    attrs: dict[str, Any],
    expected: ContainerState,
    fired_when_skip_down: bool,
):
    """Test attaching to existing container."""
    attrs["Id"] = "abc123"
    attrs["Config"] = {}
    container_collection = MagicMock()
    container_collection.get.return_value = Container(attrs)
    with patch(
        "supervisor.docker.manager.DockerAPI.containers",
        new=PropertyMock(return_value=container_collection),
    ), patch.object(type(coresys.bus), "fire_event") as fire_event, patch(
        "supervisor.docker.interface.time", return_value=1
    ):
        await coresys.homeassistant.core.instance.attach(AwesomeVersion("2022.7.3"))
        fire_event.assert_called_once_with(
            BusEvent.DOCKER_CONTAINER_STATE_CHANGE,
            DockerContainerStateEvent("homeassistant", expected, "abc123", 1),
        )

        fire_event.reset_mock()
        await coresys.homeassistant.core.instance.attach(
            AwesomeVersion("2022.7.3"), skip_state_event_if_down=True
        )
        if fired_when_skip_down:
            fire_event.assert_called_once_with(
                BusEvent.DOCKER_CONTAINER_STATE_CHANGE,
                DockerContainerStateEvent("homeassistant", expected, "abc123", 1),
            )
        else:
            fire_event.assert_not_called()


async def test_attach_container_failure(coresys: CoreSys):
    """Test attach fails to find container but finds image."""
    container_collection = MagicMock()
    container_collection.get.side_effect = DockerException()
    image_collection = MagicMock()
    image_config = {"Image": "sha256:abc123"}
    image_collection.get.return_value = Image({"Config": image_config})
    with patch(
        "supervisor.docker.manager.DockerAPI.containers",
        new=PropertyMock(return_value=container_collection),
    ), patch(
        "supervisor.docker.manager.DockerAPI.images",
        new=PropertyMock(return_value=image_collection),
    ), patch.object(
        type(coresys.bus), "fire_event"
    ) as fire_event:
        await coresys.homeassistant.core.instance.attach(AwesomeVersion("2022.7.3"))
        fire_event.assert_not_called()
        assert coresys.homeassistant.core.instance.meta_config == image_config


async def test_attach_total_failure(coresys: CoreSys):
    """Test attach fails to find container or image."""
    container_collection = MagicMock()
    container_collection.get.side_effect = DockerException()
    image_collection = MagicMock()
    image_collection.get.side_effect = DockerException()
    with patch(
        "supervisor.docker.manager.DockerAPI.containers",
        new=PropertyMock(return_value=container_collection),
    ), patch(
        "supervisor.docker.manager.DockerAPI.images",
        new=PropertyMock(return_value=image_collection),
    ), pytest.raises(
        DockerError
    ):
        await coresys.homeassistant.core.instance.attach(AwesomeVersion("2022.7.3"))