mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-06-21 17:36:30 +00:00
77fd1b4017
* Retry run if image missing and handle fixup * Fix lint and run error test * Remove retry and just capture exception
250 lines
8.8 KiB
Python
250 lines
8.8 KiB
Python
"""Test Docker interface."""
|
|
import asyncio
|
|
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.addons import Addon
|
|
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,
|
|
DockerNotFound,
|
|
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"))
|
|
await asyncio.sleep(0)
|
|
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
|
|
)
|
|
await asyncio.sleep(0)
|
|
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"))
|
|
|
|
|
|
@pytest.mark.parametrize("err", [DockerException(), RequestException()])
|
|
async def test_image_pull_fail(
|
|
coresys: CoreSys, capture_exception: Mock, err: Exception
|
|
):
|
|
"""Test failure to pull image."""
|
|
coresys.docker.images.pull.side_effect = err
|
|
with pytest.raises(DockerError):
|
|
await coresys.homeassistant.core.instance.install(
|
|
AwesomeVersion("2022.7.3"), arch=CpuArch.AMD64
|
|
)
|
|
|
|
capture_exception.assert_called_once_with(err)
|
|
|
|
|
|
async def test_run_missing_image(
|
|
coresys: CoreSys,
|
|
install_addon_ssh: Addon,
|
|
container: MagicMock,
|
|
capture_exception: Mock,
|
|
path_extern,
|
|
):
|
|
"""Test run captures the exception when image is missing."""
|
|
coresys.docker.containers.create.side_effect = [NotFound("missing"), MagicMock()]
|
|
container.status = "stopped"
|
|
install_addon_ssh.data["image"] = "test_image"
|
|
|
|
with pytest.raises(DockerNotFound):
|
|
await install_addon_ssh.instance.run()
|
|
|
|
capture_exception.assert_called_once()
|