Create evaluation modules (#2220)

* Create evaluation modules

* Use sys_core
This commit is contained in:
Joakim Sørensen 2020-11-05 17:36:02 +01:00 committed by GitHub
parent 934e59596a
commit 9479672b88
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
33 changed files with 1125 additions and 139 deletions

View File

@ -31,24 +31,26 @@ setup(
zip_safe=False,
platforms="any",
packages=[
"supervisor",
"supervisor.docker",
"supervisor.addons",
"supervisor.api",
"supervisor.dbus",
"supervisor.dbus.payloads",
"supervisor.dbus.network",
"supervisor.discovery",
"supervisor.dbus.payloads",
"supervisor.dbus",
"supervisor.discovery.services",
"supervisor.services",
"supervisor.services.modules",
"supervisor.discovery",
"supervisor.docker",
"supervisor.homeassistant",
"supervisor.host",
"supervisor.misc",
"supervisor.utils",
"supervisor.plugins",
"supervisor.resolution.evaluations",
"supervisor.resolution",
"supervisor.services.modules",
"supervisor.services",
"supervisor.snapshots",
"supervisor.store",
"supervisor.utils",
"supervisor",
],
include_package_data=True,
)

View File

@ -73,7 +73,11 @@ class APIHost(CoreSysAttributes):
@api_process
def reload(self, request):
"""Reload host data."""
return asyncio.shield(self.sys_host.reload())
return asyncio.shield(
asyncio.wait(
[self.sys_host.reload(), self.sys_resolution.evaluate.evaluate_system()]
)
)
@api_process
async def services(self, request):

View File

@ -182,7 +182,11 @@ class APISupervisor(CoreSysAttributes):
"""Reload add-ons, configuration, etc."""
return asyncio.shield(
asyncio.wait(
[self.sys_updater.reload(), self.sys_homeassistant.secrets.reload()]
[
self.sys_updater.reload(),
self.sys_homeassistant.secrets.reload(),
self.sys_resolution.evaluate.evaluate_system(),
]
)
)

View File

@ -34,11 +34,6 @@ RUN_SUPERVISOR_STATE = Path("/run/supervisor")
DOCKER_NETWORK = "hassio"
DOCKER_NETWORK_MASK = ip_network("172.30.32.0/23")
DOCKER_NETWORK_RANGE = ip_network("172.30.33.0/24")
DOCKER_IMAGE_DENYLIST = [
"containrrr/watchtower",
"pyouroboros/ouroboros",
"v2tec/watchtower",
]
DNS_SUFFIX = "local.hass.io"
@ -356,8 +351,6 @@ ROLE_ALL = [ROLE_DEFAULT, ROLE_HOMEASSISTANT, ROLE_BACKUP, ROLE_MANAGER, ROLE_AD
CHAN_ID = "chan_id"
CHAN_TYPE = "chan_type"
SUPERVISED_SUPPORTED_OS = ["Debian GNU/Linux 10 (buster)"]
class AddonBoot(str, Enum):
"""Boot mode for the add-on."""

View File

@ -6,23 +6,15 @@ from typing import Awaitable, List, Optional
import async_timeout
from .const import (
RUN_SUPERVISOR_STATE,
SOCKET_DBUS,
SUPERVISED_SUPPORTED_OS,
AddonStartup,
CoreState,
HostFeature,
)
from .const import RUN_SUPERVISOR_STATE, AddonStartup, CoreState
from .coresys import CoreSys, CoreSysAttributes
from .exceptions import (
DockerError,
HassioError,
HomeAssistantCrashError,
HomeAssistantError,
SupervisorUpdateError,
)
from .resolution.const import ContextType, IssueType, UnsupportedReason
from .resolution.const import ContextType, IssueType
_LOGGER: logging.Logger = logging.getLogger(__name__)
@ -65,34 +57,8 @@ class Core(CoreSysAttributes):
# Load information from container
await self.sys_supervisor.load()
# If host docker is supported?
if not self.sys_docker.info.supported_version:
self.sys_resolution.unsupported = UnsupportedReason.DOCKER_VERSION
self.healthy = False
_LOGGER.error(
"Docker version '%s' is not supported by Supervisor!",
self.sys_docker.info.version,
)
elif self.sys_docker.info.inside_lxc:
self.sys_resolution.unsupported = UnsupportedReason.LXC
self.healthy = False
_LOGGER.error(
"Detected Docker running inside LXC. Running Home Assistant with the Supervisor on LXC is not supported!"
)
elif not self.sys_supervisor.instance.privileged:
self.sys_resolution.unsupported = UnsupportedReason.PRIVILEGED
self.healthy = False
_LOGGER.error("Supervisor does not run in Privileged mode.")
if self.sys_docker.info.check_requirements():
self.sys_resolution.unsupported = UnsupportedReason.DOCKER_CONFIGURATION
# Dbus available
if not SOCKET_DBUS.exists():
self.sys_resolution.unsupported = UnsupportedReason.DBUS
_LOGGER.error(
"D-Bus is required for Home Assistant. This system is not supported!"
)
# Evaluate the system
await self.sys_resolution.evaluate.evaluate_system()
# Check supervisor version/update
if self.sys_dev:
@ -156,38 +122,8 @@ class Core(CoreSysAttributes):
self.healthy = False
self.sys_capture_exception(err)
# Check supported OS
if not self.sys_hassos.available:
if self.sys_host.info.operating_system not in SUPERVISED_SUPPORTED_OS:
self.sys_resolution.unsupported = UnsupportedReason.OS
_LOGGER.error(
"Detected unsupported OS: %s",
self.sys_host.info.operating_system,
)
# Check Host features
if HostFeature.NETWORK not in self.sys_host.supported_features:
self.sys_resolution.unsupported = UnsupportedReason.NETWORK_MANAGER
_LOGGER.error("NetworkManager is not correctly configured")
if any(
feature not in self.sys_host.supported_features
for feature in (
HostFeature.HOSTNAME,
HostFeature.SERVICES,
HostFeature.SHUTDOWN,
HostFeature.REBOOT,
)
):
self.sys_resolution.unsupported = UnsupportedReason.SYSTEMD
_LOGGER.error("Systemd is not correctly working")
# Check if image names from denylist exist
try:
if await self.sys_run_in_executor(self.sys_docker.check_denylist_images):
self.sys_resolution.unsupported = UnsupportedReason.CONTAINER
self.healthy = False
except DockerError:
self.healthy = False
# Evaluate the system
await self.sys_resolution.evaluate.evaluate_system()
async def start(self):
"""Start Supervisor orchestration."""

View File

@ -13,7 +13,6 @@ import requests
from ..const import (
ATTR_REGISTRIES,
DNS_SUFFIX,
DOCKER_IMAGE_DENYLIST,
DOCKER_NETWORK,
FILE_HASSIO_DOCKER,
SOCKET_DOCKER,
@ -63,16 +62,6 @@ class DockerInfo:
"""Return True if the docker run inside lxc."""
return Path("/dev/lxd/sock").exists()
def check_requirements(self) -> None:
"""Show wrong configurations."""
if self.storage != "overlay2":
_LOGGER.error("Docker storage driver %s is not supported!", self.storage)
if self.logging != "journald":
_LOGGER.error("Docker logging driver %s is not supported!", self.logging)
return self.storage != "overlay2" or self.logging != "journald"
class DockerConfig(JsonConfig):
"""Home Assistant core object for Docker configuration."""
@ -317,29 +306,3 @@ class DockerAPI:
with suppress(docker.errors.DockerException, requests.RequestException):
network.disconnect(data.get("Name", cid), force=True)
def check_denylist_images(self) -> bool:
"""Return a boolean if the host has images in the denylist."""
denied_images = set()
try:
for image in self.images.list():
for tag in image.tags:
image_name = tag.split(":")[0]
if (
image_name in DOCKER_IMAGE_DENYLIST
and image_name not in denied_images
):
denied_images.add(image_name)
except (docker.errors.DockerException, requests.RequestException) as err:
_LOGGER.error("Corrupt docker overlayfs detect: %s", err)
raise DockerError() from err
if not denied_images:
return False
_LOGGER.error(
"Found images: '%s' which are not supported, remove these from the host!",
", ".join(denied_images),
)
return True

View File

@ -94,6 +94,8 @@ class HostManager(CoreSysAttributes):
with suppress(PulseAudioError):
await self.sound.update()
_LOGGER.info("Host information reload completed")
async def load(self):
"""Load host information."""
with suppress(HassioError):

View File

@ -12,6 +12,7 @@ from .const import (
UnsupportedReason,
)
from .data import Issue, Suggestion
from .evaluate import ResolutionEvaluation
from .free_space import ResolutionStorage
from .notify import ResolutionNotify
@ -24,6 +25,7 @@ class ResolutionManager(CoreSysAttributes):
def __init__(self, coresys: CoreSys):
"""Initialize Resolution manager."""
self.coresys: CoreSys = coresys
self._evaluate = ResolutionEvaluation(coresys)
self._notify = ResolutionNotify(coresys)
self._storage = ResolutionStorage(coresys)
@ -31,6 +33,11 @@ class ResolutionManager(CoreSysAttributes):
self._issues: List[Issue] = []
self._unsupported: List[UnsupportedReason] = []
@property
def evaluate(self) -> ResolutionEvaluation:
"""Return the ResolutionEvaluation class."""
return self._evaluate
@property
def storage(self) -> ResolutionStorage:
"""Return the ResolutionStorage class."""
@ -150,3 +157,10 @@ class ResolutionManager(CoreSysAttributes):
_LOGGER.warning("The UUID %s is not a valid issue", issue.uuid)
raise ResolutionError()
self._issues.remove(issue)
async def dismiss_unsupported(self, reason: Issue) -> None:
"""Dismiss a reason for unsupported."""
if reason not in self._unsupported:
_LOGGER.warning("The reason %s is not active", reason)
raise ResolutionError()
self._unsupported.remove(reason)

View File

@ -0,0 +1,59 @@
"""Helpers to evaluate the system."""
import logging
from ..coresys import CoreSys, CoreSysAttributes
from .const import UnsupportedReason
from .evaluations.container import EvaluateContainer
from .evaluations.dbus import EvaluateDbus
from .evaluations.docker_configuration import EvaluateDockerConfiguration
from .evaluations.docker_version import EvaluateDockerVersion
from .evaluations.lxc import EvaluateLxc
from .evaluations.network_manager import EvaluateNetworkManager
from .evaluations.operating_system import EvaluateOperatingSystem
from .evaluations.privileged import EvaluatePrivileged
from .evaluations.systemd import EvaluateSystemd
_LOGGER: logging.Logger = logging.getLogger(__name__)
UNHEALTHY = [
UnsupportedReason.CONTAINER,
UnsupportedReason.DOCKER_VERSION,
UnsupportedReason.LXC,
UnsupportedReason.PRIVILEGED,
]
class ResolutionEvaluation(CoreSysAttributes):
"""Evaluation class for resolution."""
def __init__(self, coresys: CoreSys) -> None:
"""Initialize the evaluation class."""
self.coresys = coresys
self._container = EvaluateContainer(coresys)
self._dbus = EvaluateDbus(coresys)
self._docker_configuration = EvaluateDockerConfiguration(coresys)
self._docker_version = EvaluateDockerVersion(coresys)
self._lxc = EvaluateLxc(coresys)
self._network_manager = EvaluateNetworkManager(coresys)
self._operating_system = EvaluateOperatingSystem(coresys)
self._privileged = EvaluatePrivileged(coresys)
self._systemd = EvaluateSystemd(coresys)
async def evaluate_system(self) -> None:
"""Evaluate the system."""
_LOGGER.info("Starting system evaluation with state %s", self.sys_core.state)
await self._container()
await self._dbus()
await self._docker_configuration()
await self._docker_version()
await self._lxc()
await self._network_manager()
await self._operating_system()
await self._privileged()
await self._systemd()
if any(reason in self.sys_resolution.unsupported for reason in UNHEALTHY):
self.sys_core.healthy = False
_LOGGER.info("System evaluation complete")

View File

@ -0,0 +1 @@
"""Initialize system."""

View File

@ -0,0 +1,53 @@
"""Baseclass for system evaluations."""
import logging
from typing import List
from ...const import CoreState
from ...coresys import CoreSys, CoreSysAttributes
from ..const import UnsupportedReason
_LOGGER: logging.Logger = logging.getLogger(__name__)
class EvaluateBase(CoreSysAttributes):
"""Baseclass for evaluation."""
def __init__(self, coresys: CoreSys) -> None:
"""Initialize the evaluation class."""
self.coresys = coresys
async def __call__(self) -> None:
"""Execute the evaluation."""
if self.sys_core.state not in self.states:
return
if await self.evaluate():
if self.reason not in self.sys_resolution.unsupported:
self.sys_resolution.unsupported = self.reason
_LOGGER.warning(
"%s (more-info: https://www.home-assistant.io/more-info/unsupported/%s)",
self.on_failure,
self.reason.value,
)
else:
if self.reason in self.sys_resolution.unsupported:
_LOGGER.info("Clearing %s as reason for unsupported", self.reason)
await self.sys_resolution.dismiss_unsupported(self.reason)
async def evaluate(self):
"""Run evaluation."""
raise NotImplementedError
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
raise NotImplementedError
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
raise NotImplementedError
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return []

View File

@ -0,0 +1,68 @@
"""Evaluation class for container."""
import logging
from typing import Any, List
from docker.errors import DockerException
from requests import RequestException
from ...const import CoreState
from ...coresys import CoreSys
from ..const import UnsupportedReason
from .base import EvaluateBase
_LOGGER: logging.Logger = logging.getLogger(__name__)
DOCKER_IMAGE_DENYLIST = [
"containrrr/watchtower",
"pyouroboros/ouroboros",
"v2tec/watchtower",
]
class EvaluateContainer(EvaluateBase):
"""Evaluate container."""
def __init__(self, coresys: CoreSys) -> None:
"""Initialize the evaluation class."""
super().__init__(coresys)
self.coresys = coresys
self._images = set()
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.CONTAINER
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return f"Found images: {self._images} which are not supported, remove these from the host!"
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.SETUP, CoreState.RUNNING]
async def evaluate(self) -> None:
"""Run evaluation."""
self._images.clear()
for image in await self.sys_run_in_executor(self._get_images):
for tag in image.tags:
image_name = tag.split(":")[0]
if (
image_name in DOCKER_IMAGE_DENYLIST
and image_name not in self._images
):
self._images.add(image_name)
return len(self._images) != 0
def _get_images(self) -> List[Any]:
"""Return a list of images."""
images = []
try:
images = self.sys_docker.images.list()
except (DockerException, RequestException) as err:
_LOGGER.error("Corrupt docker overlayfs detect: %s", err)
return images

View File

@ -0,0 +1,29 @@
"""Evaluation class for dbus."""
from typing import List
from ...const import SOCKET_DBUS, CoreState
from ..const import UnsupportedReason
from .base import EvaluateBase
class EvaluateDbus(EvaluateBase):
"""Evaluate dbus."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.DBUS
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return "D-Bus is required for Home Assistant."
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.INITIALIZE]
async def evaluate(self) -> None:
"""Run evaluation."""
return not SOCKET_DBUS.exists()

View File

@ -0,0 +1,44 @@
"""Evaluation class for docker configuration."""
import logging
from typing import List
from ...const import CoreState
from ..const import UnsupportedReason
from .base import EvaluateBase
EXPECTED_LOGGING = "journald"
EXPECTED_STORAGE = "overlay2"
_LOGGER: logging.Logger = logging.getLogger(__name__)
class EvaluateDockerConfiguration(EvaluateBase):
"""Evaluate Docker configuration."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.DOCKER_CONFIGURATION
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return "The configuration of Docker is not supported"
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.INITIALIZE]
async def evaluate(self):
"""Run evaluation."""
_storage = self.sys_docker.info.storage
_logging = self.sys_docker.info.logging
if _storage != EXPECTED_STORAGE:
_LOGGER.warning("Docker storage driver %s is not supported!", _storage)
if _logging != EXPECTED_LOGGING:
_LOGGER.warning("Docker logging driver %s is not supported!", _logging)
return _storage != EXPECTED_STORAGE or _logging != EXPECTED_LOGGING

View File

@ -0,0 +1,29 @@
"""Evaluation class for docker version."""
from typing import List
from ...const import CoreState
from ..const import UnsupportedReason
from .base import EvaluateBase
class EvaluateDockerVersion(EvaluateBase):
"""Evaluate Docker version."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.DOCKER_VERSION
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return f"Docker version '{self.sys_docker.info.version}' is not supported by the Supervisor!"
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.INITIALIZE]
async def evaluate(self):
"""Run evaluation."""
return not self.sys_docker.info.supported_version

View File

@ -0,0 +1,29 @@
"""Evaluation class for lxc."""
from typing import List
from ...const import CoreState
from ..const import UnsupportedReason
from .base import EvaluateBase
class EvaluateLxc(EvaluateBase):
"""Evaluate if running inside LXC."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.LXC
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return "Detected Docker running inside LXC."
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.INITIALIZE]
async def evaluate(self):
"""Run evaluation."""
return self.sys_docker.info.inside_lxc

View File

@ -0,0 +1,29 @@
"""Evaluation class for network manager."""
from typing import List
from ...const import CoreState, HostFeature
from ..const import UnsupportedReason
from .base import EvaluateBase
class EvaluateNetworkManager(EvaluateBase):
"""Evaluate network manager."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.NETWORK_MANAGER
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return "NetworkManager is not correctly configured"
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.SETUP, CoreState.RUNNING]
async def evaluate(self):
"""Run evaluation."""
return HostFeature.NETWORK not in self.sys_host.supported_features

View File

@ -0,0 +1,33 @@
"""Evaluation class for operating system."""
from typing import List
from ...const import CoreState
from ..const import UnsupportedReason
from .base import EvaluateBase
SUPPORTED_OS = ["Debian GNU/Linux 10 (buster)"]
class EvaluateOperatingSystem(EvaluateBase):
"""Evaluate the operating system."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.OS
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return f"Detected unsupported OS: {self.sys_host.info.operating_system}"
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.SETUP]
async def evaluate(self):
"""Run evaluation."""
if self.sys_hassos.available:
return False
return self.sys_host.info.operating_system not in SUPPORTED_OS

View File

@ -0,0 +1,29 @@
"""Evaluation class for privileged."""
from typing import List
from ...const import CoreState
from ..const import UnsupportedReason
from .base import EvaluateBase
class EvaluatePrivileged(EvaluateBase):
"""Evaluate Privileged mode."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.PRIVILEGED
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return "Supervisor does not run in Privileged mode."
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.INITIALIZE]
async def evaluate(self):
"""Run evaluation."""
return not self.sys_supervisor.instance.privileged

View File

@ -0,0 +1,37 @@
"""Evaluation class for systemd."""
from typing import List
from ...const import CoreState, HostFeature
from ..const import UnsupportedReason
from .base import EvaluateBase
class EvaluateSystemd(EvaluateBase):
"""Evaluate systemd."""
@property
def reason(self) -> UnsupportedReason:
"""Return a UnsupportedReason enum."""
return UnsupportedReason.SYSTEMD
@property
def on_failure(self) -> str:
"""Return a string that is printed when self.evaluate is False."""
return "Systemd is not correctly working"
@property
def states(self) -> List[CoreState]:
"""Return a list of valid states when this evaluation can run."""
return [CoreState.SETUP]
async def evaluate(self):
"""Run evaluation."""
return any(
feature not in self.sys_host.supported_features
for feature in (
HostFeature.HOSTNAME,
HostFeature.SERVICES,
HostFeature.SHUTDOWN,
HostFeature.REBOOT,
)
)

View File

@ -1,15 +0,0 @@
"""Test tags in denylist."""
from unittest.mock import MagicMock, patch
def test_has_images_in_denylist(docker):
"""Test tags in denylist exsist."""
images = [MagicMock(tags=["containrrr/watchtower:latest"])]
with patch("supervisor.docker.DockerAPI.images.list", return_value=images):
assert docker.check_denylist_images()
def test_no_images_in_denylist(docker):
"""Test tags in denylist does not exsist."""
assert not docker.check_denylist_images()

View File

@ -0,0 +1,21 @@
"""Test evaluation base."""
# pylint: disable=import-error
import pytest
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.base import EvaluateBase
async def test_evaluation_base(coresys: CoreSys):
"""Test evaluation base."""
base = EvaluateBase(coresys)
assert not base.states
with pytest.raises(NotImplementedError):
await base.evaluate()
with pytest.raises(NotImplementedError):
assert not base.on_failure
with pytest.raises(NotImplementedError):
assert not base.reason

View File

@ -0,0 +1,75 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import MagicMock, patch
from docker.errors import DockerException
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.container import (
DOCKER_IMAGE_DENYLIST,
EvaluateContainer,
)
def test_get_images(coresys: CoreSys):
"""Test getting images form docker."""
container = EvaluateContainer(coresys)
with patch(
"supervisor.resolution.evaluations.container.EvaluateContainer._get_images",
return_value=[MagicMock(tags=["test"])],
):
images = container._get_images()
assert images[0].tags[0] == "test"
with patch("supervisor.docker.DockerAPI.images.list", side_effect=DockerException):
images = container._get_images()
assert not images
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
container = EvaluateContainer(coresys)
coresys.core.state = CoreState.RUNNING
assert container.reason not in coresys.resolution.unsupported
with patch(
"supervisor.resolution.evaluations.container.EvaluateContainer._get_images",
return_value=[MagicMock(tags=[f"{DOCKER_IMAGE_DENYLIST[0]}:latest"])],
):
await container()
assert container.reason in coresys.resolution.unsupported
with patch(
"supervisor.resolution.evaluations.container.EvaluateContainer._get_images",
return_value=[MagicMock(tags=[])],
):
await container()
assert container.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
container = EvaluateContainer(coresys)
should_run = container.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.evaluations.container.EvaluateContainer.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await container()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await container()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,48 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.dbus import EvaluateDbus
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
dbus = EvaluateDbus(coresys)
coresys.core.state = CoreState.INITIALIZE
assert dbus.reason not in coresys.resolution.unsupported
with patch("pathlib.Path.exists", return_value=False):
await dbus()
assert dbus.reason in coresys.resolution.unsupported
with patch("pathlib.Path.exists", return_value=True):
await dbus()
assert dbus.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
dbus = EvaluateDbus(coresys)
should_run = dbus.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.evaluations.dbus.EvaluateDbus.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await dbus()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await dbus()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,61 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.docker_configuration import (
EXPECTED_LOGGING,
EXPECTED_STORAGE,
EvaluateDockerConfiguration,
)
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
docker_configuration = EvaluateDockerConfiguration(coresys)
coresys.core.state = CoreState.INITIALIZE
assert docker_configuration.reason not in coresys.resolution.unsupported
coresys.docker.info.storage = "unsupported"
coresys.docker.info.logging = EXPECTED_LOGGING
await docker_configuration()
assert docker_configuration.reason in coresys.resolution.unsupported
coresys.resolution.unsupported.clear()
coresys.docker.info.storage = EXPECTED_STORAGE
coresys.docker.info.logging = "unsupported"
await docker_configuration()
assert docker_configuration.reason in coresys.resolution.unsupported
coresys.resolution.unsupported.clear()
coresys.docker.info.storage = EXPECTED_STORAGE
coresys.docker.info.logging = EXPECTED_LOGGING
await docker_configuration()
assert docker_configuration.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
docker_configuration = EvaluateDockerConfiguration(coresys)
should_run = docker_configuration.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.evaluations.docker_configuration.EvaluateDockerConfiguration.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await docker_configuration()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await docker_configuration()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,48 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.docker_version import EvaluateDockerVersion
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
docker_version = EvaluateDockerVersion(coresys)
coresys.core.state = CoreState.INITIALIZE
assert docker_version.reason not in coresys.resolution.unsupported
coresys.docker.info.supported_version = False
await docker_version()
assert docker_version.reason in coresys.resolution.unsupported
coresys.docker.info.supported_version = True
await docker_version()
assert docker_version.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
docker_version = EvaluateDockerVersion(coresys)
should_run = docker_version.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.evaluations.docker_version.EvaluateDockerVersion.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await docker_version()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await docker_version()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,48 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.lxc import EvaluateLxc
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
lxc = EvaluateLxc(coresys)
coresys.core.state = CoreState.INITIALIZE
assert lxc.reason not in coresys.resolution.unsupported
coresys.docker.info.inside_lxc = True
await lxc()
assert lxc.reason in coresys.resolution.unsupported
coresys.docker.info.inside_lxc = False
await lxc()
assert lxc.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
lxc = EvaluateLxc(coresys)
should_run = lxc.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.evaluations.lxc.EvaluateLxc.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await lxc()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await lxc()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,48 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.network_manager import EvaluateNetworkManager
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
network_manager = EvaluateNetworkManager(coresys)
coresys.core.state = CoreState.RUNNING
assert network_manager.reason not in coresys.resolution.unsupported
coresys.dbus.network.is_connected = False
await network_manager()
assert network_manager.reason in coresys.resolution.unsupported
coresys.dbus.network.is_connected = True
await network_manager()
assert network_manager.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
network_manager = EvaluateNetworkManager(coresys)
should_run = network_manager.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.evaluations.network_manager.EvaluateNetworkManager.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await network_manager()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await network_manager()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,62 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import MagicMock, patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.operating_system import (
SUPPORTED_OS,
EvaluateOperatingSystem,
)
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
operating_system = EvaluateOperatingSystem(coresys)
coresys.core.state = CoreState.SETUP
assert operating_system.reason not in coresys.resolution.unsupported
with patch(
"supervisor.utils.gdbus.DBusCallWrapper",
return_value=MagicMock(hostname=MagicMock(operating_system="unsupported")),
):
await operating_system()
assert operating_system.reason in coresys.resolution.unsupported
coresys.hassos._available = True
await operating_system()
assert operating_system.reason not in coresys.resolution.unsupported
coresys.hassos._available = False
with patch(
"supervisor.utils.gdbus.DBusCallWrapper",
return_value=MagicMock(hostname=MagicMock(operating_system=SUPPORTED_OS[0])),
):
await operating_system()
assert operating_system.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
operating_system = EvaluateOperatingSystem(coresys)
should_run = operating_system.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.evaluations.operating_system.EvaluateOperatingSystem.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await operating_system()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await operating_system()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,48 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.privileged import EvaluatePrivileged
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
privileged = EvaluatePrivileged(coresys)
coresys.core.state = CoreState.INITIALIZE
assert privileged.reason not in coresys.resolution.unsupported
coresys.supervisor.instance._meta = {"HostConfig": {"Privileged": False}}
await privileged()
assert privileged.reason in coresys.resolution.unsupported
coresys.supervisor.instance._meta = {"HostConfig": {"Privileged": True}}
await privileged()
assert privileged.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
privileged = EvaluatePrivileged(coresys)
should_run = privileged.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.evaluations.privileged.EvaluatePrivileged.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await privileged()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await privileged()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,63 @@
"""Test evaluation base."""
# pylint: disable=import-error,protected-access
from unittest.mock import MagicMock, patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.evaluations.systemd import EvaluateSystemd
async def test_evaluation(coresys: CoreSys):
"""Test evaluation."""
systemd = EvaluateSystemd(coresys)
coresys.core.state = CoreState.SETUP
assert systemd.reason not in coresys.resolution.unsupported
with patch(
"supervisor.utils.gdbus.DBusCallWrapper",
return_value=MagicMock(systemd=MagicMock(is_connected=False)),
):
await systemd()
assert systemd.reason in coresys.resolution.unsupported
with patch(
"supervisor.utils.gdbus.DBusCallWrapper",
return_value=MagicMock(hostname=MagicMock(is_connected=False)),
):
await systemd()
assert systemd.reason in coresys.resolution.unsupported
with patch(
"supervisor.utils.gdbus.DBusCallWrapper",
return_value=MagicMock(
hostname=MagicMock(is_connected=True), systemd=MagicMock(is_connected=True)
),
):
await systemd()
assert systemd.reason not in coresys.resolution.unsupported
async def test_did_run(coresys: CoreSys):
"""Test that the evaluation ran as expected."""
systemd = EvaluateSystemd(coresys)
should_run = systemd.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.evaluations.systemd.EvaluateSystemd.evaluate",
return_value=None,
) as evaluate:
for state in should_run:
coresys.core.state = state
await systemd()
evaluate.assert_called_once()
evaluate.reset_mock()
for state in should_not_run:
coresys.core.state = state
await systemd()
evaluate.assert_not_called()
evaluate.reset_mock()

View File

@ -0,0 +1,103 @@
"""Test evaluation."""
# pylint: disable=import-error
from unittest.mock import patch
from supervisor.const import CoreState
from supervisor.coresys import CoreSys
from supervisor.resolution.const import UnsupportedReason
async def test_evaluation_initialize(coresys: CoreSys):
"""Test evaluation for initialize."""
coresys.core.state = CoreState.INITIALIZE
with patch(
"supervisor.resolution.evaluations.dbus.EvaluateDbus.evaluate",
return_value=False,
) as dbus, patch(
"supervisor.resolution.evaluations.lxc.EvaluateLxc.evaluate",
return_value=False,
) as lxc, patch(
"supervisor.resolution.evaluations.privileged.EvaluatePrivileged.evaluate",
return_value=False,
) as privileged, patch(
"supervisor.resolution.evaluations.docker_configuration.EvaluateDockerConfiguration.evaluate",
return_value=False,
) as docker_configuration, patch(
"supervisor.resolution.evaluations.docker_version.EvaluateDockerVersion.evaluate",
return_value=False,
) as docker_version:
await coresys.resolution.evaluate.evaluate_system()
dbus.assert_called_once()
lxc.assert_called_once()
privileged.assert_called_once()
docker_configuration.assert_called_once()
docker_version.assert_called_once()
async def test_evaluation_setup(coresys: CoreSys):
"""Test evaluation for setup."""
coresys.core.state = CoreState.SETUP
with patch(
"supervisor.resolution.evaluations.operating_system.EvaluateOperatingSystem.evaluate",
return_value=False,
) as operating_system, patch(
"supervisor.resolution.evaluations.container.EvaluateContainer.evaluate",
return_value=False,
) as container, patch(
"supervisor.resolution.evaluations.network_manager.EvaluateNetworkManager.evaluate",
return_value=False,
) as network_manager, patch(
"supervisor.resolution.evaluations.systemd.EvaluateSystemd.evaluate",
return_value=False,
) as systemd:
await coresys.resolution.evaluate.evaluate_system()
operating_system.assert_called_once()
container.assert_called_once()
network_manager.assert_called_once()
systemd.assert_called_once()
async def test_evaluation_running(coresys: CoreSys):
"""Test evaluation for running."""
coresys.core.state = CoreState.RUNNING
with patch(
"supervisor.resolution.evaluations.container.EvaluateContainer.evaluate",
return_value=False,
) as container, patch(
"supervisor.resolution.evaluations.network_manager.EvaluateNetworkManager.evaluate",
return_value=False,
) as network_manager:
await coresys.resolution.evaluate.evaluate_system()
container.assert_called_once()
network_manager.assert_called_once()
async def test_adding_and_removing_unsupported_reason(coresys: CoreSys):
"""Test adding and removing unsupported reason."""
coresys.core.state = CoreState.RUNNING
assert UnsupportedReason.NETWORK_MANAGER not in coresys.resolution.unsupported
with patch(
"supervisor.resolution.evaluations.network_manager.EvaluateNetworkManager.evaluate",
return_value=True,
):
await coresys.resolution.evaluate.evaluate_system()
assert UnsupportedReason.NETWORK_MANAGER in coresys.resolution.unsupported
assert not coresys.core.supported
with patch(
"supervisor.resolution.evaluations.network_manager.EvaluateNetworkManager.evaluate",
return_value=False,
):
await coresys.resolution.evaluate.evaluate_system()
assert UnsupportedReason.NETWORK_MANAGER not in coresys.resolution.unsupported
assert coresys.core.supported
async def test_set_unhealthy(coresys: CoreSys):
"""Test setting unhealthy."""
assert coresys.core.healthy
coresys.resolution.unsupported = UnsupportedReason.CONTAINER
await coresys.resolution.evaluate.evaluate_system()
assert not coresys.core.healthy

View File

@ -81,6 +81,9 @@ async def test_resolution_dismiss_suggestion(coresys: CoreSys):
await coresys.resolution.dismiss_suggestion(clear_snapshot)
assert clear_snapshot not in coresys.resolution.suggestions
with pytest.raises(ResolutionError):
await coresys.resolution.dismiss_suggestion(clear_snapshot)
@pytest.mark.asyncio
async def test_resolution_apply_suggestion(coresys: CoreSys):
@ -121,6 +124,9 @@ async def test_resolution_dismiss_issue(coresys: CoreSys):
await coresys.resolution.dismiss_issue(updated_failed)
assert updated_failed not in coresys.resolution.issues
with pytest.raises(ResolutionError):
await coresys.resolution.dismiss_issue(updated_failed)
@pytest.mark.asyncio
async def test_resolution_create_issue_suggestion(coresys: CoreSys):
@ -138,3 +144,15 @@ async def test_resolution_create_issue_suggestion(coresys: CoreSys):
assert SuggestionType.EXECUTE_REPAIR == coresys.resolution.suggestions[-1].type
assert ContextType.CORE == coresys.resolution.suggestions[-1].context
@pytest.mark.asyncio
async def test_resolution_dismiss_unsupported(coresys: CoreSys):
"""Test resolution manager dismiss unsupported reason."""
coresys.resolution.unsupported = UnsupportedReason.CONTAINER
await coresys.resolution.dismiss_unsupported(UnsupportedReason.CONTAINER)
assert UnsupportedReason.CONTAINER not in coresys.resolution.unsupported
with pytest.raises(ResolutionError):
await coresys.resolution.dismiss_unsupported(UnsupportedReason.CONTAINER)