mirror of
https://github.com/home-assistant/supervisor.git
synced 2026-10-01 13:03:39 +01:00
Stefan AgnerandClaude Fable 5.1
9387c4ea65
A failed container or image removal left the plugin persisted as disabled but with its version and image still around, and every later disable request returned early because the plugin was already disabled. Move the removal into a dedicated step that stops the container, removes the image using the stored version and only then forgets the version. disable() always runs it, so a repeated request retries the cleanup, and the disabled load path runs it as well to catch a disable interrupted by a Supervisor exit. The step no longer relies on DockerInterface.remove, which derives the version from container metadata that is not available before an attach. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
267 lines
9.5 KiB
Python
267 lines
9.5 KiB
Python
"""Test multicast plugin."""
|
|
|
|
from unittest.mock import AsyncMock, PropertyMock, patch
|
|
|
|
from awesomeversion import AwesomeVersion
|
|
import pytest
|
|
|
|
from supervisor.const import BusEvent
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.docker.const import ContainerState
|
|
from supervisor.docker.manager import DockerAPI
|
|
from supervisor.docker.monitor import DockerContainerStateEvent
|
|
from supervisor.docker.multicast import DockerMulticast
|
|
from supervisor.exceptions import (
|
|
DockerContainerNotFoundError,
|
|
DockerContainerNotRunningError,
|
|
DockerError,
|
|
DockerStatsTimeoutError,
|
|
MulticastDisabledError,
|
|
MulticastError,
|
|
MulticastNotRunningError,
|
|
MulticastStatsTimeoutError,
|
|
MulticastUnknownError,
|
|
)
|
|
|
|
|
|
async def test_stats_not_running(coresys: CoreSys):
|
|
"""Test stats raises MulticastNotRunningError when the container isn't running."""
|
|
with (
|
|
patch.object(
|
|
DockerMulticast,
|
|
"stats",
|
|
AsyncMock(
|
|
side_effect=DockerContainerNotRunningError(name="hassio_multicast")
|
|
),
|
|
),
|
|
pytest.raises(MulticastNotRunningError),
|
|
):
|
|
await coresys.plugins.multicast.stats()
|
|
|
|
with (
|
|
patch.object(
|
|
DockerMulticast,
|
|
"stats",
|
|
AsyncMock(
|
|
side_effect=DockerContainerNotFoundError(name="hassio_multicast")
|
|
),
|
|
),
|
|
pytest.raises(MulticastNotRunningError),
|
|
):
|
|
await coresys.plugins.multicast.stats()
|
|
|
|
|
|
async def test_stats_timeout(coresys: CoreSys):
|
|
"""Test stats raises MulticastStatsTimeoutError on timeout."""
|
|
with (
|
|
patch.object(
|
|
DockerMulticast,
|
|
"stats",
|
|
AsyncMock(side_effect=DockerStatsTimeoutError(name="hassio_multicast")),
|
|
),
|
|
pytest.raises(MulticastStatsTimeoutError),
|
|
):
|
|
await coresys.plugins.multicast.stats()
|
|
|
|
|
|
async def test_stats_unknown_error(coresys: CoreSys):
|
|
"""Test stats raises MulticastUnknownError on an unexpected Docker error."""
|
|
with (
|
|
patch.object(
|
|
DockerMulticast, "stats", AsyncMock(side_effect=DockerError("boom"))
|
|
),
|
|
pytest.raises(MulticastUnknownError),
|
|
):
|
|
await coresys.plugins.multicast.stats()
|
|
|
|
|
|
async def test_enabled_default(coresys: CoreSys):
|
|
"""Test the multicast plugin is enabled by default."""
|
|
assert coresys.plugins.multicast.enabled is True
|
|
|
|
|
|
async def test_load_disabled(coresys: CoreSys):
|
|
"""Test load skips install and start but cleans up leftovers when disabled."""
|
|
coresys.plugins.multicast._data["enabled"] = False # pylint: disable=protected-access
|
|
coresys.plugins.multicast.version = AwesomeVersion("2024.01.0")
|
|
|
|
with (
|
|
patch.object(type(coresys.bus), "register_event") as register_event,
|
|
patch.object(DockerMulticast, "attach") as attach,
|
|
patch.object(DockerMulticast, "stop") as stop,
|
|
patch.object(DockerAPI, "remove_image") as remove_image,
|
|
patch.object(type(coresys.plugins.multicast), "install") as install,
|
|
patch.object(DockerMulticast, "run") as run,
|
|
patch.object(type(coresys.plugins.multicast), "save_data"),
|
|
):
|
|
await coresys.plugins.multicast.load()
|
|
|
|
register_event.assert_any_call(
|
|
BusEvent.DOCKER_CONTAINER_STATE_CHANGE,
|
|
coresys.plugins.multicast.watchdog_container,
|
|
)
|
|
attach.assert_not_called()
|
|
install.assert_not_called()
|
|
run.assert_not_called()
|
|
# Leftovers from an unfinished disable are removed
|
|
stop.assert_called_once()
|
|
remove_image.assert_called_once_with(
|
|
coresys.plugins.multicast.image, AwesomeVersion("2024.01.0")
|
|
)
|
|
assert coresys.plugins.multicast.version is None
|
|
|
|
|
|
async def test_load_disabled_cleanup_failure(coresys: CoreSys):
|
|
"""Test load tolerates a failing leftover cleanup when disabled."""
|
|
coresys.plugins.multicast._data["enabled"] = False # pylint: disable=protected-access
|
|
coresys.plugins.multicast.version = AwesomeVersion("2024.01.0")
|
|
|
|
with (
|
|
patch.object(DockerMulticast, "stop", side_effect=DockerError("boom")),
|
|
patch.object(DockerAPI, "remove_image") as remove_image,
|
|
):
|
|
await coresys.plugins.multicast.load()
|
|
|
|
remove_image.assert_not_called()
|
|
assert coresys.plugins.multicast.version == AwesomeVersion("2024.01.0")
|
|
|
|
|
|
async def test_watchdog_ignored_when_disabled(coresys: CoreSys):
|
|
"""Test the watchdog does not restart the plugin while disabled."""
|
|
coresys.plugins.multicast._data["enabled"] = False # pylint: disable=protected-access
|
|
|
|
with patch.object(
|
|
type(coresys.plugins.multicast), "_restart_after_problem"
|
|
) as restart_after_problem:
|
|
await coresys.plugins.multicast.watchdog_container(
|
|
DockerContainerStateEvent(
|
|
name="hassio_multicast",
|
|
state=ContainerState.FAILED,
|
|
id="abc123",
|
|
time=1,
|
|
)
|
|
)
|
|
restart_after_problem.assert_not_called()
|
|
|
|
|
|
async def test_disable(coresys: CoreSys):
|
|
"""Test disabling removes container and image and forgets the version."""
|
|
coresys.plugins.multicast.version = AwesomeVersion("2024.01.0")
|
|
|
|
with (
|
|
patch.object(DockerMulticast, "stop") as stop,
|
|
patch.object(DockerAPI, "remove_image") as remove_image,
|
|
patch.object(type(coresys.plugins.multicast), "save_data") as save_data,
|
|
):
|
|
await coresys.plugins.multicast.disable()
|
|
|
|
stop.assert_called_once()
|
|
remove_image.assert_called_once_with(
|
|
coresys.plugins.multicast.image, AwesomeVersion("2024.01.0")
|
|
)
|
|
assert save_data.called
|
|
assert coresys.plugins.multicast.enabled is False
|
|
assert coresys.plugins.multicast.version is None
|
|
assert coresys.plugins.multicast.need_update is False
|
|
|
|
# Disabling again only makes sure the container is gone
|
|
with (
|
|
patch.object(DockerMulticast, "stop") as stop,
|
|
patch.object(DockerAPI, "remove_image") as remove_image,
|
|
):
|
|
await coresys.plugins.multicast.disable()
|
|
stop.assert_called_once()
|
|
remove_image.assert_not_called()
|
|
|
|
|
|
async def test_disable_remove_failure_retried(coresys: CoreSys):
|
|
"""Test a failed removal persists the disabled state and is retried."""
|
|
coresys.plugins.multicast.version = AwesomeVersion("2024.01.0")
|
|
|
|
with (
|
|
patch.object(DockerMulticast, "stop"),
|
|
patch.object(DockerAPI, "remove_image", side_effect=DockerError("boom")),
|
|
patch.object(type(coresys.plugins.multicast), "save_data"),
|
|
pytest.raises(MulticastError),
|
|
):
|
|
await coresys.plugins.multicast.disable()
|
|
|
|
assert coresys.plugins.multicast.enabled is False
|
|
# Version is kept so the leftover image can be cleaned up on retry
|
|
assert coresys.plugins.multicast.version == AwesomeVersion("2024.01.0")
|
|
|
|
with (
|
|
patch.object(DockerMulticast, "stop"),
|
|
patch.object(DockerAPI, "remove_image") as remove_image,
|
|
patch.object(type(coresys.plugins.multicast), "save_data"),
|
|
):
|
|
await coresys.plugins.multicast.disable()
|
|
|
|
remove_image.assert_called_once()
|
|
assert coresys.plugins.multicast.version is None
|
|
|
|
|
|
@pytest.mark.usefixtures("supervisor_internet")
|
|
async def test_enable(coresys: CoreSys):
|
|
"""Test enabling installs and starts the plugin."""
|
|
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
|
coresys.plugins.multicast._data["enabled"] = False # pylint: disable=protected-access
|
|
coresys.plugins.multicast._data.pop("version", None) # pylint: disable=protected-access
|
|
|
|
with (
|
|
patch.object(
|
|
type(coresys.plugins.multicast),
|
|
"latest_version",
|
|
new=PropertyMock(return_value=AwesomeVersion("2024.01.0")),
|
|
),
|
|
patch.object(DockerMulticast, "install") as install,
|
|
patch.object(DockerMulticast, "run") as run,
|
|
patch.object(type(coresys.plugins.multicast), "save_data") as save_data,
|
|
):
|
|
await coresys.plugins.multicast.enable()
|
|
|
|
install.assert_called_once()
|
|
run.assert_called_once()
|
|
assert save_data.called
|
|
assert coresys.plugins.multicast.enabled is True
|
|
assert coresys.plugins.multicast.version == AwesomeVersion("2024.01.0")
|
|
|
|
# Enabling again is a no-op
|
|
with patch.object(DockerMulticast, "install") as install:
|
|
await coresys.plugins.multicast.enable()
|
|
install.assert_not_called()
|
|
|
|
|
|
async def test_actions_rejected_when_disabled(coresys: CoreSys):
|
|
"""Test start, restart and update refuse to run while disabled."""
|
|
coresys.plugins.multicast._data["enabled"] = False # pylint: disable=protected-access
|
|
|
|
with pytest.raises(MulticastDisabledError):
|
|
await coresys.plugins.multicast.start()
|
|
with pytest.raises(MulticastDisabledError):
|
|
await coresys.plugins.multicast.restart()
|
|
|
|
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
|
with (
|
|
patch.object(
|
|
type(coresys.plugins.multicast),
|
|
"latest_version",
|
|
new=PropertyMock(return_value=AwesomeVersion("2024.01.0")),
|
|
),
|
|
pytest.raises(MulticastDisabledError),
|
|
):
|
|
await coresys.plugins.multicast.update()
|
|
|
|
|
|
async def test_repair_skipped_when_disabled(coresys: CoreSys):
|
|
"""Test repair does nothing while disabled."""
|
|
coresys.plugins.multicast._data["enabled"] = False # pylint: disable=protected-access
|
|
|
|
with (
|
|
patch.object(DockerMulticast, "exists", return_value=False),
|
|
patch.object(DockerMulticast, "install") as install,
|
|
):
|
|
await coresys.plugins.multicast.repair()
|
|
|
|
install.assert_not_called()
|