Make issue for problem with config for containers (#4317)

* Make issue for problem with config for containers

* Mount propagation in tests

* Fixes from rebase and feedback
This commit is contained in:
Mike Degatano 2023-05-30 13:25:38 -04:00 committed by GitHub
parent 0df19cee91
commit 841f68c175
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 922 additions and 25 deletions

View File

@ -111,6 +111,11 @@ class DockerInterface(CoreSysAttributes):
"""Return meta data of labels for container/image."""
return self.meta_config.get("Labels") or {}
@property
def meta_mounts(self) -> list[dict[str, Any]]:
"""Return meta data of mounts for container/image."""
return self._meta.get("Mounts", [])
@property
def image(self) -> str | None:
"""Return name of Docker image."""

View File

@ -10,6 +10,7 @@ import requests
from ..coresys import CoreSysAttributes
from ..exceptions import DockerError
from .const import PropagationMode
from .interface import DockerInterface
_LOGGER: logging.Logger = logging.getLogger(__name__)
@ -33,6 +34,15 @@ class DockerSupervisor(DockerInterface, CoreSysAttributes):
"""Return True if the container run with Privileged."""
return self.meta_host.get("Privileged", False)
@property
def host_mounts_available(self) -> bool:
"""Return True if container can see mounts on host within its data directory."""
return self._meta and any(
mount.get("Propagation") == PropagationMode.SLAVE.value
for mount in self.meta_mounts
if mount.get("Destination") == "/data"
)
def _attach(
self, version: AwesomeVersion, skip_state_event_if_down: bool = False
) -> None:

View File

@ -111,9 +111,14 @@ class HostManager(CoreSysAttributes):
if self.sys_dbus.udisks2.is_connected:
features.append(HostFeature.DISK)
# Support added in OS10. For supervised, assume they can if systemd is connected
if self.sys_dbus.systemd.is_connected and (
not self.sys_os.available or self.sys_os.version >= AwesomeVersion("10")
# Support added in OS10. Propagation mode changed on mount in 10.2 to support this
if (
self.sys_dbus.systemd.is_connected
and self.sys_supervisor.instance.host_mounts_available
and (
not self.sys_os.available
or self.sys_os.version >= AwesomeVersion("10.2")
)
):
features.append(HostFeature.MOUNT)

View File

@ -0,0 +1,102 @@
"""Helper to check if docker config for container needs an update."""
from ...const import CoreState
from ...coresys import CoreSys
from ...docker.const import PropagationMode
from ...docker.interface import DockerInterface
from ..const import ContextType, IssueType, SuggestionType
from ..data import Issue
from .base import CheckBase
def _check_container(container: DockerInterface) -> bool:
"""Return true if container has a config issue."""
return any(
mount.get("Propagation") != PropagationMode.SLAVE.value
for mount in container.meta_mounts
if mount.get("Destination") == "/media"
)
def setup(coresys: CoreSys) -> CheckBase:
"""Check setup function."""
return CheckDockerConfig(coresys)
class CheckDockerConfig(CheckBase):
"""CheckDockerConfig class for check."""
async def run_check(self) -> None:
"""Run check if not affected by issue."""
self._check_docker_config()
if self.current_issues:
self.sys_resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.SYSTEM,
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
async def approve_check(self, reference: str | None = None) -> bool:
"""Approve check if it is affected by issue."""
self._check_docker_config()
return bool(self.current_issues)
def _check_docker_config(self) -> None:
"""Check docker config and make issues."""
new_issues: set[Issue] = set()
if _check_container(self.sys_homeassistant.core.instance):
new_issues.add(Issue(IssueType.DOCKER_CONFIG, ContextType.CORE))
for addon in self.sys_addons.installed:
if _check_container(addon.instance):
new_issues.add(
Issue(
IssueType.DOCKER_CONFIG, ContextType.ADDON, reference=addon.slug
)
)
for plugin in self.sys_plugins.all_plugins:
if _check_container(plugin.instance):
new_issues.add(
Issue(
IssueType.DOCKER_CONFIG,
ContextType.PLUGIN,
reference=plugin.slug,
)
)
# Make an issue for each container with a bad config
for issue in new_issues - self.current_issues:
self.sys_resolution.add_issue(
issue, suggestions=[SuggestionType.EXECUTE_REBUILD]
)
# Dismiss issues when container config has been fixed
for issue in self.current_issues - new_issues:
self.sys_resolution.dismiss_issue(issue)
@property
def current_issues(self) -> set[Issue]:
"""List of current docker config issues, excluding the system one."""
return {
issue
for issue in self.sys_resolution.issues
if issue.type == IssueType.DOCKER_CONFIG and issue.context != self.context
}
@property
def issue(self) -> IssueType:
"""Return a IssueType enum."""
return IssueType.DOCKER_CONFIG
@property
def context(self) -> ContextType:
"""Return a ContextType enum."""
return ContextType.SYSTEM
@property
def states(self) -> list[CoreState]:
"""Return a list of valid states when this check can run."""
return [CoreState.RUNNING]

View File

@ -74,6 +74,7 @@ class IssueType(str, Enum):
DNS_LOOP = "dns_loop"
DNS_SERVER_FAILED = "dns_server_failed"
DNS_SERVER_IPV6_ERROR = "dns_server_ipv6_error"
DOCKER_CONFIG = "docker_config"
DOCKER_RATELIMIT = "docker_ratelimit"
FATAL_ERROR = "fatal_error"
FREE_SPACE = "free_space"
@ -97,6 +98,7 @@ class SuggestionType(str, Enum):
CREATE_FULL_BACKUP = "create_full_backup"
EXECUTE_INTEGRITY = "execute_integrity"
EXECUTE_REBOOT = "execute_reboot"
EXECUTE_REBUILD = "execute_rebuild"
EXECUTE_RELOAD = "execute_reload"
EXECUTE_REMOVE = "execute_remove"
EXECUTE_REPAIR = "execute_repair"

View File

@ -0,0 +1,59 @@
"""Helper to fix an issue with an addon by rebuilding its container."""
import logging
from ...coresys import CoreSys
from ...docker.const import ContainerState
from ..const import ContextType, IssueType, SuggestionType
from .base import FixupBase
_LOGGER: logging.Logger = logging.getLogger(__name__)
def setup(coresys: CoreSys) -> FixupBase:
"""Check setup function."""
return FixupAddonExecuteRebuild(coresys)
class FixupAddonExecuteRebuild(FixupBase):
"""Storage class for fixup."""
async def process_fixup(self, reference: str | None = None) -> None:
"""Rebuild the addon's container."""
addon = self.sys_addons.get(reference, local_only=True)
if not addon:
_LOGGER.info(
"Cannot rebuild addon %s as it is not installed, dismissing suggestion",
reference,
)
return
state = await addon.instance.current_state()
if state == ContainerState.UNKNOWN:
_LOGGER.info(
"Container for addon %s does not exist, it will be rebuilt when started next",
reference,
)
elif state == ContainerState.STOPPED:
_LOGGER.info(
"Addon %s is stopped, removing its container so it rebuilds when started next",
reference,
)
await addon.stop()
else:
await addon.restart()
@property
def suggestion(self) -> SuggestionType:
"""Return a SuggestionType enum."""
return SuggestionType.EXECUTE_REBUILD
@property
def context(self) -> ContextType:
"""Return a ContextType enum."""
return ContextType.ADDON
@property
def issues(self) -> list[IssueType]:
"""Return a IssueType enum list."""
return [IssueType.DOCKER_CONFIG]

View File

@ -0,0 +1,50 @@
"""Helper to fix an issue with core by rebuilding its container."""
import logging
from ...coresys import CoreSys
from ...docker.const import ContainerState
from ..const import ContextType, IssueType, SuggestionType
from .base import FixupBase
_LOGGER: logging.Logger = logging.getLogger(__name__)
def setup(coresys: CoreSys) -> FixupBase:
"""Check setup function."""
return FixupCoreExecuteRebuild(coresys)
class FixupCoreExecuteRebuild(FixupBase):
"""Storage class for fixup."""
async def process_fixup(self, reference: str | None = None) -> None:
"""Rebuild the core container."""
state = await self.sys_homeassistant.core.instance.current_state()
if state == ContainerState.UNKNOWN:
_LOGGER.info(
"Container for Home Assistant does not exist, it will be rebuilt when started next"
)
elif state == ContainerState.STOPPED:
_LOGGER.info(
"Home Assistant is stopped, removing its container so it rebuilds when started next"
)
await self.sys_homeassistant.core.instance.stop()
else:
await self.sys_homeassistant.core.rebuild()
@property
def suggestion(self) -> SuggestionType:
"""Return a SuggestionType enum."""
return SuggestionType.EXECUTE_REBUILD
@property
def context(self) -> ContextType:
"""Return a ContextType enum."""
return ContextType.CORE
@property
def issues(self) -> list[IssueType]:
"""Return a IssueType enum list."""
return [IssueType.DOCKER_CONFIG]

View File

@ -0,0 +1,49 @@
"""Helper to fix an issue with an plugin by rebuilding its container."""
from ...coresys import CoreSys
from ..const import ContextType, IssueType, SuggestionType
from .base import FixupBase
def setup(coresys: CoreSys) -> FixupBase:
"""Check setup function."""
return FixupPluginExecuteRebuild(coresys)
class FixupPluginExecuteRebuild(FixupBase):
"""Storage class for fixup."""
async def process_fixup(self, reference: str | None = None) -> None:
"""Rebuild the plugin's container."""
plugin = next(
(
plugin
for plugin in self.sys_plugins.all_plugins
if plugin.slug == reference
),
None,
)
if not plugin:
return
await plugin.rebuild()
@property
def suggestion(self) -> SuggestionType:
"""Return a SuggestionType enum."""
return SuggestionType.EXECUTE_REBUILD
@property
def context(self) -> ContextType:
"""Return a ContextType enum."""
return ContextType.PLUGIN
@property
def issues(self) -> list[IssueType]:
"""Return a IssueType enum list."""
return [IssueType.DOCKER_CONFIG]
@property
def auto(self) -> bool:
"""Return if a fixup can be apply as auto fix."""
return True

View File

@ -0,0 +1,52 @@
"""Helper to fix an issue with the system by rebuilding containers."""
import asyncio
from ...coresys import CoreSys
from ..const import ContextType, IssueType, SuggestionType
from ..data import Issue
from .base import FixupBase
def setup(coresys: CoreSys) -> FixupBase:
"""Check setup function."""
return FixupSystemExecuteRebuild(coresys)
class FixupSystemExecuteRebuild(FixupBase):
"""Storage class for fixup."""
async def process_fixup(self, reference: str | None = None) -> None:
"""Rebuild containers with docker config issues."""
await asyncio.gather(
*[
self.sys_resolution.apply_suggestion(suggestion)
for issue in self.current_issues
for suggestion in self.sys_resolution.suggestions_for_issue(issue)
if suggestion.type == SuggestionType.EXECUTE_REBUILD
]
)
@property
def current_issues(self) -> set[Issue]:
"""List of current docker config issues, excluding the system one."""
return {
issue
for issue in self.sys_resolution.issues
if issue.type == IssueType.DOCKER_CONFIG and issue.context != self.context
}
@property
def suggestion(self) -> SuggestionType:
"""Return a SuggestionType enum."""
return SuggestionType.EXECUTE_REBUILD
@property
def context(self) -> ContextType:
"""Return a ContextType enum."""
return ContextType.SYSTEM
@property
def issues(self) -> list[IssueType]:
"""Return a IssueType enum list."""
return [IssueType.DOCKER_CONFIG]

View File

@ -62,6 +62,7 @@ async def test_backup_to_location(
backup_dir: PurePath,
tmp_supervisor_data: Path,
path_extern,
mount_propagation,
):
"""Test making a backup to a specific location with default mount."""
await coresys.mounts.load()
@ -96,7 +97,11 @@ async def test_backup_to_location(
async def test_backup_to_default(
api_client: TestClient, coresys: CoreSys, tmp_supervisor_data, path_extern
api_client: TestClient,
coresys: CoreSys,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test making backup to default mount."""
await coresys.mounts.load()

View File

@ -17,7 +17,9 @@ from tests.dbus_service_mocks.systemd_unit import SystemdUnit as SystemdUnitServ
@pytest.fixture(name="mount")
async def fixture_mount(coresys: CoreSys, tmp_supervisor_data, path_extern) -> Mount:
async def fixture_mount(
coresys: CoreSys, tmp_supervisor_data, path_extern, mount_propagation
) -> Mount:
"""Add an initial mount and load mounts."""
mount = Mount.from_dict(
coresys,
@ -44,7 +46,11 @@ async def test_api_mounts_info(api_client: TestClient):
async def test_api_create_mount(
api_client: TestClient, coresys: CoreSys, tmp_supervisor_data, path_extern
api_client: TestClient,
coresys: CoreSys,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test creating a mount via API."""
resp = await api_client.post(
@ -99,6 +105,7 @@ async def test_api_create_dbus_error_mount_not_added(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test mount not added to list of mounts if a dbus error occurs."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -159,12 +166,38 @@ async def test_api_create_dbus_error_mount_not_added(
@pytest.mark.parametrize("os_available", ["9.5"], indirect=True)
async def test_api_create_mount_fails_not_supported_feature(
async def test_api_create_mount_fails_os_out_of_date(
api_client: TestClient,
coresys: CoreSys,
os_available,
mount_propagation,
):
"""Test creating a mount via API fails when mounting isn't supported due to OS version."""
resp = await api_client.post(
"/mounts",
json={
"name": "backup_test",
"type": "cifs",
"usage": "backup",
"server": "backup.local",
"share": "backups",
},
)
assert resp.status == 400
result = await resp.json()
assert result["result"] == "error"
assert (
result["message"]
== "'MountManager.create_mount' blocked from execution, mounting not supported on system"
)
async def test_api_create_mount_fails_missing_mount_propagation(
api_client: TestClient,
coresys: CoreSys,
os_available,
):
"""Test creating a mount via API fails when mounting isn't a supported feature on system.."""
"""Test creating a mount via API fails when mounting isn't supported due to container config."""
resp = await api_client.post(
"/mounts",
json={
@ -214,7 +247,9 @@ async def test_api_update_mount(api_client: TestClient, coresys: CoreSys, mount)
coresys.mounts.save_data.assert_called_once()
async def test_api_update_error_mount_missing(api_client: TestClient):
async def test_api_update_error_mount_missing(
api_client: TestClient, mount_propagation
):
"""Test update mount API errors when mount does not exist."""
resp = await api_client.put(
"/mounts/backup_test",
@ -237,6 +272,7 @@ async def test_api_update_dbus_error_mount_remains(
mount,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test mount remains in list with unsuccessful state if dbus error occurs during update."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -330,7 +366,9 @@ async def test_api_reload_mount(
]
async def test_api_reload_error_mount_missing(api_client: TestClient):
async def test_api_reload_error_mount_missing(
api_client: TestClient, mount_propagation
):
"""Test reload mount API errors when mount does not exist."""
resp = await api_client.post("/mounts/backup_test/reload")
assert resp.status == 400
@ -356,7 +394,9 @@ async def test_api_delete_mount(api_client: TestClient, coresys: CoreSys, mount)
coresys.mounts.save_data.assert_called_once()
async def test_api_delete_error_mount_missing(api_client: TestClient):
async def test_api_delete_error_mount_missing(
api_client: TestClient, mount_propagation
):
"""Test delete mount API errors when mount does not exist."""
resp = await api_client.delete("/mounts/backup_test")
assert resp.status == 400
@ -369,7 +409,11 @@ async def test_api_delete_error_mount_missing(api_client: TestClient):
async def test_api_create_backup_mount_sets_default(
api_client: TestClient, coresys: CoreSys, tmp_supervisor_data, path_extern
api_client: TestClient,
coresys: CoreSys,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test creating backup mounts sets default if not set."""
await coresys.mounts.load()
@ -488,6 +532,7 @@ async def test_backup_mounts_reload_backups(
coresys: CoreSys,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test actions on a backup mount reload backups."""
await coresys.mounts.load()

View File

@ -371,6 +371,7 @@ async def test_backup_media_with_mounts(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test backing up media folder with mounts."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -432,6 +433,7 @@ async def test_backup_share_with_mounts(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test backing up share folder with mounts."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -488,7 +490,9 @@ async def test_backup_share_with_mounts(
assert not mount_dir.exists()
async def test_full_backup_to_mount(coresys: CoreSys, tmp_supervisor_data, path_extern):
async def test_full_backup_to_mount(
coresys: CoreSys, tmp_supervisor_data, path_extern, mount_propagation
):
"""Test full backup to and restoring from a mount."""
(marker := coresys.config.path_homeassistant / "test.txt").touch()
@ -531,7 +535,10 @@ async def test_full_backup_to_mount(coresys: CoreSys, tmp_supervisor_data, path_
async def test_partial_backup_to_mount(
coresys: CoreSys, tmp_supervisor_data, path_extern
coresys: CoreSys,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test partial backup to and restoring from a mount."""
(marker := coresys.config.path_homeassistant / "test.txt").touch()
@ -584,7 +591,10 @@ async def test_partial_backup_to_mount(
async def test_backup_to_local_with_default(
coresys: CoreSys, tmp_supervisor_data, path_extern
coresys: CoreSys,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test making backup to local when a default mount is specified."""
# Add a default backup mount
@ -618,7 +628,9 @@ async def test_backup_to_local_with_default(
assert (coresys.config.path_backup / f"{backup.slug}.tar").exists()
async def test_backup_to_default(coresys: CoreSys, tmp_supervisor_data, path_extern):
async def test_backup_to_default(
coresys: CoreSys, tmp_supervisor_data, path_extern, mount_propagation
):
"""Test making backup to default mount."""
# Add a default backup mount
(mount_dir := coresys.config.path_mounts / "backup_test").mkdir()

View File

@ -63,7 +63,7 @@ from .dbus_service_mocks.network_manager import NetworkManager as NetworkManager
# pylint: disable=redefined-outer-name, protected-access
async def mock_async_return_true() -> bool:
async def mock_async_return_true(*args, **kwargs) -> bool:
"""Mock methods to return True."""
return True
@ -614,9 +614,30 @@ async def os_available(request: pytest.FixtureRequest) -> None:
version = (
AwesomeVersion(request.param)
if hasattr(request, "param")
else AwesomeVersion("10.0")
else AwesomeVersion("10.2")
)
with patch.object(
OSManager, "available", new=PropertyMock(return_value=True)
), patch.object(OSManager, "version", new=PropertyMock(return_value=version)):
yield
@pytest.fixture
async def mount_propagation(docker: DockerAPI, coresys: CoreSys) -> None:
"""Mock supervisor connected to container with propagation set."""
os.environ["SUPERVISOR_NAME"] = "hassio_supervisor"
docker.containers.get.return_value = supervisor = MagicMock()
supervisor.attrs = {
"Mounts": [
{
"Type": "bind",
"Source": "/mnt/data/supervisor",
"Destination": "/data",
"Mode": "rw",
"RW": True,
"Propagation": "slave",
}
]
}
await coresys.supervisor.load()
yield

View File

@ -51,7 +51,9 @@ SHARE_TEST_DATA = {
@pytest.fixture(name="mount")
async def fixture_mount(coresys: CoreSys, tmp_supervisor_data, path_extern) -> Mount:
async def fixture_mount(
coresys: CoreSys, tmp_supervisor_data, path_extern, mount_propagation
) -> Mount:
"""Add an initial mount and load mounts."""
mount = Mount.from_dict(coresys, MEDIA_TEST_DATA)
coresys.mounts._mounts = {"media_test": mount} # pylint: disable=protected-access
@ -64,6 +66,7 @@ async def test_load(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test mount manager loading."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -147,6 +150,7 @@ async def test_load_share_mount(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test mount manager loading with share mount."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -210,6 +214,7 @@ async def test_mount_failed_during_load(
dbus_session_bus: MessageBus,
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test mount failed during load."""
await mock_dbus_services(
@ -319,6 +324,7 @@ async def test_create_mount(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test creating a mount."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -356,7 +362,9 @@ async def test_create_mount(
async def test_update_mount(
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], mount: Mount
coresys: CoreSys,
all_dbus_services: dict[str, DBusServiceMock],
mount: Mount,
):
"""Test updating a mount."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -392,7 +400,9 @@ async def test_update_mount(
async def test_reload_mount(
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], mount: Mount
coresys: CoreSys,
all_dbus_services: dict[str, DBusServiceMock],
mount: Mount,
):
"""Test reloading a mount."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -430,7 +440,7 @@ async def test_remove_mount(
]
async def test_remove_reload_mount_missing(coresys: CoreSys):
async def test_remove_reload_mount_missing(coresys: CoreSys, mount_propagation):
"""Test removing or reloading a non existent mount errors."""
await coresys.mounts.load()
@ -441,7 +451,9 @@ async def test_remove_reload_mount_missing(coresys: CoreSys):
await coresys.mounts.reload_mount("does_not_exist")
async def test_save_data(coresys: CoreSys, tmp_supervisor_data: Path, path_extern):
async def test_save_data(
coresys: CoreSys, tmp_supervisor_data: Path, path_extern, mount_propagation
):
"""Test saving mount config data."""
# Replace mount manager with one that doesn't have save_data mocked
coresys._mounts = MountManager(coresys) # pylint: disable=protected-access
@ -487,6 +499,7 @@ async def test_create_mount_start_unit_failure(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test failure to start mount unit does not add mount to the list."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -517,6 +530,7 @@ async def test_create_mount_activation_failure(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test activation failure during create mount does not add mount to the list and unmounts new mount."""
systemd_service: SystemdService = all_dbus_services["systemd"]
@ -617,6 +631,7 @@ async def test_create_share_mount(
all_dbus_services: dict[str, DBusServiceMock],
tmp_supervisor_data,
path_extern,
mount_propagation,
):
"""Test creating a share mount."""
systemd_service: SystemdService = all_dbus_services["systemd"]

View File

@ -0,0 +1,141 @@
"""Test check Docker Config."""
from unittest.mock import MagicMock, patch
from supervisor.addons.addon import Addon
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.docker.interface import DockerInterface
from supervisor.docker.manager import DockerAPI
from supervisor.resolution.checks.docker_config import CheckDockerConfig
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.data import Issue, Suggestion
from tests.conftest import mock_async_return_true
def _make_mock_container_get(bad_config_names: list[str]):
"""Make mock of container get."""
def mock_container_get(name):
out = MagicMock()
out.status = "running"
out.attrs = {"State": {}, "Mounts": []}
if name in bad_config_names:
out.attrs["Mounts"].append(
{
"Type": "bind",
"Source": "/mnt/data/supervisor/media",
"Destination": "/media",
"Mode": "rw",
"RW": True,
"Propagation": "rprivate",
}
)
return out
return mock_container_get
async def test_base(coresys: CoreSys):
"""Test check basics."""
docker_config = CheckDockerConfig(coresys)
assert docker_config.slug == "docker_config"
assert docker_config.enabled
async def test_check(docker: DockerAPI, coresys: CoreSys, install_addon_ssh: Addon):
"""Test check reports issue when containers have incorrect config."""
docker.containers.get = _make_mock_container_get(
["homeassistant", "hassio_audio", "addon_local_ssh"]
)
with patch.object(DockerInterface, "is_running", new=mock_async_return_true):
await coresys.plugins.load()
await coresys.homeassistant.load()
await coresys.addons.load()
docker_config = CheckDockerConfig(coresys)
coresys.core.state = CoreState.RUNNING
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
# An issue and suggestion is added per container with a config issue
await docker_config.run_check()
assert len(coresys.resolution.issues) == 4
assert Issue(IssueType.DOCKER_CONFIG, ContextType.CORE) in coresys.resolution.issues
assert (
Issue(IssueType.DOCKER_CONFIG, ContextType.ADDON, reference="local_ssh")
in coresys.resolution.issues
)
assert (
Issue(IssueType.DOCKER_CONFIG, ContextType.PLUGIN, reference="audio")
in coresys.resolution.issues
)
assert (
Issue(IssueType.DOCKER_CONFIG, ContextType.SYSTEM) in coresys.resolution.issues
)
assert len(coresys.resolution.suggestions) == 4
assert (
Suggestion(SuggestionType.EXECUTE_REBUILD, ContextType.CORE)
in coresys.resolution.suggestions
)
assert (
Suggestion(
SuggestionType.EXECUTE_REBUILD, ContextType.PLUGIN, reference="audio"
)
in coresys.resolution.suggestions
)
assert (
Suggestion(
SuggestionType.EXECUTE_REBUILD, ContextType.ADDON, reference="local_ssh"
)
in coresys.resolution.suggestions
)
assert (
Suggestion(SuggestionType.EXECUTE_REBUILD, ContextType.SYSTEM)
in coresys.resolution.suggestions
)
assert await docker_config.approve_check()
# IF config issue is resolved, all issues are removed except the main one. Which will be removed if check isn't approved
docker.containers.get = _make_mock_container_get([])
with patch.object(DockerInterface, "is_running", new=mock_async_return_true):
await coresys.plugins.load()
await coresys.homeassistant.load()
await coresys.addons.load()
assert not await docker_config.approve_check()
assert len(coresys.resolution.issues) == 1
assert len(coresys.resolution.suggestions) == 1
assert (
Issue(IssueType.DOCKER_CONFIG, ContextType.SYSTEM) in coresys.resolution.issues
)
async def test_did_run(coresys: CoreSys):
"""Test that the check ran as expected."""
docker_config = CheckDockerConfig(coresys)
should_run = docker_config.states
should_not_run = [state for state in CoreState if state not in should_run]
assert len(should_run) != 0
assert len(should_not_run) != 0
with patch(
"supervisor.resolution.checks.docker_config.CheckDockerConfig.run_check",
return_value=None,
) as check:
for state in should_run:
coresys.core.state = state
await docker_config()
check.assert_called_once()
check.reset_mock()
for state in should_not_run:
coresys.core.state = state
await docker_config()
check.assert_not_called()
check.reset_mock()

View File

@ -0,0 +1,118 @@
"""Test fixup core execute rebuild."""
from unittest.mock import MagicMock, patch
from docker.errors import NotFound
import pytest
from supervisor.addons.addon import Addon
from supervisor.coresys import CoreSys
from supervisor.docker.interface import DockerInterface
from supervisor.docker.manager import DockerAPI
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.fixups.addon_execute_rebuild import FixupAddonExecuteRebuild
def make_mock_container_get(status: str):
"""Make mock of container get."""
out = MagicMock()
out.status = status
out.attrs = {"State": {"ExitCode": 0}, "Mounts": []}
def mock_container_get(name):
return out
return mock_container_get
async def test_fixup(docker: DockerAPI, coresys: CoreSys, install_addon_ssh: Addon):
"""Test fixup rebuilds addon's container."""
docker.containers.get = make_mock_container_get("running")
addon_execute_rebuild = FixupAddonExecuteRebuild(coresys)
assert addon_execute_rebuild.auto is False
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.ADDON,
reference="local_ssh",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(Addon, "restart") as restart:
await addon_execute_rebuild()
restart.assert_called_once()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
async def test_fixup_stopped_core(
docker: DockerAPI,
coresys: CoreSys,
install_addon_ssh: Addon,
caplog: pytest.LogCaptureFixture,
):
"""Test fixup just removes addon's container when it is stopped."""
caplog.clear()
docker.containers.get = make_mock_container_get("stopped")
addon_execute_rebuild = FixupAddonExecuteRebuild(coresys)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.ADDON,
reference="local_ssh",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(Addon, "restart") as restart:
await addon_execute_rebuild()
restart.assert_not_called()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
docker.containers.get("addon_local_ssh").remove.assert_called_once_with(force=True)
assert "Addon local_ssh is stopped" in caplog.text
async def test_fixup_unknown_core(
docker: DockerAPI,
coresys: CoreSys,
install_addon_ssh: Addon,
caplog: pytest.LogCaptureFixture,
):
"""Test fixup does nothing if addon's container has already been removed."""
caplog.clear()
docker.containers.get.side_effect = NotFound("")
addon_execute_rebuild = FixupAddonExecuteRebuild(coresys)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.ADDON,
reference="local_ssh",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(Addon, "restart") as restart, patch.object(
DockerInterface, "stop"
) as stop:
await addon_execute_rebuild()
restart.assert_not_called()
stop.assert_not_called()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
assert "Container for addon local_ssh does not exist" in caplog.text
async def test_fixup_addon_removed(coresys: CoreSys, caplog: pytest.LogCaptureFixture):
"""Test fixup does nothing if addon has been removed."""
caplog.clear()
addon_execute_rebuild = FixupAddonExecuteRebuild(coresys)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.ADDON,
reference="local_ssh",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
await addon_execute_rebuild()
assert "Cannot rebuild addon local_ssh as it is not installed" in caplog.text

View File

@ -0,0 +1,94 @@
"""Test fixup core execute rebuild."""
from unittest.mock import MagicMock, patch
from docker.errors import NotFound
import pytest
from supervisor.coresys import CoreSys
from supervisor.docker.interface import DockerInterface
from supervisor.docker.manager import DockerAPI
from supervisor.homeassistant.core import HomeAssistantCore
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.fixups.core_execute_rebuild import FixupCoreExecuteRebuild
def make_mock_container_get(status: str):
"""Make mock of container get."""
out = MagicMock()
out.status = status
out.attrs = {"State": {"ExitCode": 0}, "Mounts": []}
def mock_container_get(name):
return out
return mock_container_get
async def test_fixup(docker: DockerAPI, coresys: CoreSys):
"""Test fixup rebuilds core's container."""
docker.containers.get = make_mock_container_get("running")
core_execute_rebuild = FixupCoreExecuteRebuild(coresys)
assert core_execute_rebuild.auto is False
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.CORE,
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(HomeAssistantCore, "rebuild") as rebuild:
await core_execute_rebuild()
rebuild.assert_called_once()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
async def test_fixup_stopped_core(
docker: DockerAPI, coresys: CoreSys, caplog: pytest.LogCaptureFixture
):
"""Test fixup just removes HA's container when it is stopped."""
caplog.clear()
docker.containers.get = make_mock_container_get("stopped")
core_execute_rebuild = FixupCoreExecuteRebuild(coresys)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.CORE,
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(HomeAssistantCore, "rebuild") as rebuild:
await core_execute_rebuild()
rebuild.assert_not_called()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
docker.containers.get("homeassistant").remove.assert_called_once_with(force=True)
assert "Home Assistant is stopped" in caplog.text
async def test_fixup_unknown_core(
docker: DockerAPI, coresys: CoreSys, caplog: pytest.LogCaptureFixture
):
"""Test fixup does nothing if core's container has already been removed."""
caplog.clear()
docker.containers.get.side_effect = NotFound("")
core_execute_rebuild = FixupCoreExecuteRebuild(coresys)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.CORE,
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(HomeAssistantCore, "rebuild") as rebuild, patch.object(
DockerInterface, "stop"
) as stop:
await core_execute_rebuild()
rebuild.assert_not_called()
stop.assert_not_called()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
assert "Container for Home Assistant does not exist" in caplog.text

View File

@ -10,7 +10,10 @@ from tests.dbus_service_mocks.systemd import Systemd as SystemdService
async def test_fixup(
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], path_extern
coresys: CoreSys,
all_dbus_services: dict[str, DBusServiceMock],
path_extern,
mount_propagation,
):
"""Test fixup."""
systemd_service: SystemdService = all_dbus_services["systemd"]

View File

@ -10,7 +10,10 @@ from tests.dbus_service_mocks.systemd import Systemd as SystemdService
async def test_fixup(
coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], path_extern
coresys: CoreSys,
all_dbus_services: dict[str, DBusServiceMock],
path_extern,
mount_propagation,
):
"""Test fixup."""
systemd_service: SystemdService = all_dbus_services["systemd"]

View File

@ -0,0 +1,48 @@
"""Test fixup plugin execute rebuild."""
from unittest.mock import MagicMock, patch
import pytest
from supervisor.coresys import CoreSys
from supervisor.docker.manager import DockerAPI
from supervisor.plugins.audio import PluginAudio
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.fixups.plugin_execute_rebuild import (
FixupPluginExecuteRebuild,
)
def make_mock_container_get(status: str):
"""Make mock of container get."""
out = MagicMock()
out.status = status
out.attrs = {"State": {"ExitCode": 0}, "Mounts": []}
def mock_container_get(name):
return out
return mock_container_get
@pytest.mark.parametrize("status", ["running", "stopped"])
async def test_fixup(docker: DockerAPI, coresys: CoreSys, status: str):
"""Test fixup rebuilds plugin's container regardless of current state."""
docker.containers.get = make_mock_container_get(status)
plugin_execute_rebuild = FixupPluginExecuteRebuild(coresys)
assert plugin_execute_rebuild.auto is True
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.PLUGIN,
reference="audio",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(PluginAudio, "rebuild") as rebuild:
await plugin_execute_rebuild()
rebuild.assert_called_once()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions

View File

@ -0,0 +1,58 @@
"""Test fixup system execute rebuild."""
from unittest.mock import patch
from supervisor.coresys import CoreSys
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.fixups.addon_execute_rebuild import FixupAddonExecuteRebuild
from supervisor.resolution.fixups.core_execute_rebuild import FixupCoreExecuteRebuild
from supervisor.resolution.fixups.plugin_execute_rebuild import (
FixupPluginExecuteRebuild,
)
from supervisor.resolution.fixups.system_execute_rebuild import (
FixupSystemExecuteRebuild,
)
async def test_fixup(coresys: CoreSys):
"""Test fixup applies other rebuild fixups for docker config issues."""
system_execute_rebuild = FixupSystemExecuteRebuild(coresys)
assert system_execute_rebuild.auto is False
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.ADDON,
reference="local_ssh",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.CORE,
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.PLUGIN,
reference="audio",
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
coresys.resolution.create_issue(
IssueType.DOCKER_CONFIG,
ContextType.SYSTEM,
suggestions=[SuggestionType.EXECUTE_REBUILD],
)
with patch.object(
FixupAddonExecuteRebuild, "process_fixup"
) as addon_fixup, patch.object(
FixupCoreExecuteRebuild, "process_fixup"
) as core_fixup, patch.object(
FixupPluginExecuteRebuild, "process_fixup"
) as plugin_fixup:
await system_execute_rebuild()
addon_fixup.assert_called_once_with(reference="local_ssh")
core_fixup.assert_called_once()
plugin_fixup.assert_called_once_with(reference="audio")
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions