Create addon boot failed issue for repair (#5397)

* Create addon boot failed issue for repair

* MDont make new objects for contains checks
This commit is contained in:
Mike Degatano 2024-11-07 13:39:15 -05:00 committed by GitHub
parent 473662e56d
commit e1e5d3a8f2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 381 additions and 23 deletions

View File

@ -81,7 +81,8 @@ from ..hardware.data import Device
from ..homeassistant.const import WSEvent, WSType from ..homeassistant.const import WSEvent, WSType
from ..jobs.const import JobExecutionLimit from ..jobs.const import JobExecutionLimit
from ..jobs.decorator import Job from ..jobs.decorator import Job
from ..resolution.const import UnhealthyReason from ..resolution.const import ContextType, IssueType, UnhealthyReason
from ..resolution.data import Issue
from ..store.addon import AddonStore from ..store.addon import AddonStore
from ..utils import check_port from ..utils import check_port
from ..utils.apparmor import adjust_profile from ..utils.apparmor import adjust_profile
@ -144,11 +145,19 @@ class Addon(AddonModel):
self._listeners: list[EventListener] = [] self._listeners: list[EventListener] = []
self._startup_event = asyncio.Event() self._startup_event = asyncio.Event()
self._startup_task: asyncio.Task | None = None self._startup_task: asyncio.Task | None = None
self._boot_failed_issue = Issue(
IssueType.BOOT_FAIL, ContextType.ADDON, reference=self.slug
)
def __repr__(self) -> str: def __repr__(self) -> str:
"""Return internal representation.""" """Return internal representation."""
return f"<Addon: {self.slug}>" return f"<Addon: {self.slug}>"
@property
def boot_failed_issue(self) -> Issue:
"""Get issue used if start on boot failed."""
return self._boot_failed_issue
@property @property
def state(self) -> AddonState: def state(self) -> AddonState:
"""Return state of the add-on.""" """Return state of the add-on."""
@ -166,6 +175,13 @@ class Addon(AddonModel):
if new_state == AddonState.STARTED or old_state == AddonState.STARTUP: if new_state == AddonState.STARTED or old_state == AddonState.STARTUP:
self._startup_event.set() self._startup_event.set()
# Dismiss boot failed issue if present and we started
if (
new_state == AddonState.STARTED
and self.boot_failed_issue in self.sys_resolution.issues
):
self.sys_resolution.dismiss_issue(self.boot_failed_issue)
self.sys_homeassistant.websocket.send_message( self.sys_homeassistant.websocket.send_message(
{ {
ATTR_TYPE: WSType.SUPERVISOR_EVENT, ATTR_TYPE: WSType.SUPERVISOR_EVENT,
@ -322,6 +338,13 @@ class Addon(AddonModel):
"""Store user boot options.""" """Store user boot options."""
self.persist[ATTR_BOOT] = value self.persist[ATTR_BOOT] = value
# Dismiss boot failed issue if present and boot at start disabled
if (
value == AddonBoot.MANUAL
and self._boot_failed_issue in self.sys_resolution.issues
):
self.sys_resolution.dismiss_issue(self._boot_failed_issue)
@property @property
def auto_update(self) -> bool: def auto_update(self) -> bool:
"""Return if auto update is enable.""" """Return if auto update is enable."""

View File

@ -7,24 +7,22 @@ import logging
import tarfile import tarfile
from typing import Union from typing import Union
from attr import evolve
from ..const import AddonBoot, AddonStartup, AddonState from ..const import AddonBoot, AddonStartup, AddonState
from ..coresys import CoreSys, CoreSysAttributes from ..coresys import CoreSys, CoreSysAttributes
from ..exceptions import ( from ..exceptions import (
AddonConfigurationError,
AddonsError, AddonsError,
AddonsJobError, AddonsJobError,
AddonsNotSupportedError, AddonsNotSupportedError,
CoreDNSError, CoreDNSError,
DockerAPIError,
DockerError, DockerError,
DockerNotFound,
HassioError, HassioError,
HomeAssistantAPIError, HomeAssistantAPIError,
) )
from ..jobs.decorator import Job, JobCondition from ..jobs.decorator import Job, JobCondition
from ..resolution.const import ContextType, IssueType, SuggestionType from ..resolution.const import ContextType, IssueType, SuggestionType
from ..store.addon import AddonStore from ..store.addon import AddonStore
from ..utils import check_exception_chain
from ..utils.sentry import capture_exception from ..utils.sentry import capture_exception
from .addon import Addon from .addon import Addon
from .const import ADDON_UPDATE_CONDITIONS from .const import ADDON_UPDATE_CONDITIONS
@ -118,15 +116,14 @@ class AddonManager(CoreSysAttributes):
try: try:
if start_task := await addon.start(): if start_task := await addon.start():
wait_boot.append(start_task) wait_boot.append(start_task)
except AddonsError as err:
# Check if there is an system/user issue
if check_exception_chain(
err, (DockerAPIError, DockerNotFound, AddonConfigurationError)
):
addon.boot = AddonBoot.MANUAL
addon.save_persist()
except HassioError: except HassioError:
pass # These are already handled self.sys_resolution.add_issue(
evolve(addon.boot_failed_issue),
suggestions=[
SuggestionType.EXECUTE_START,
SuggestionType.DISABLE_BOOT,
],
)
else: else:
continue continue
@ -135,6 +132,19 @@ class AddonManager(CoreSysAttributes):
# Ignore exceptions from waiting for addon startup, addon errors handled elsewhere # Ignore exceptions from waiting for addon startup, addon errors handled elsewhere
await asyncio.gather(*wait_boot, return_exceptions=True) await asyncio.gather(*wait_boot, return_exceptions=True)
# After waiting for startup, create an issue for boot addons that are error or unknown state
# Ignore stopped as single shot addons can be run at boot and this is successful exit
# Timeout waiting for startup is not a failure, addon is probably just slow
for addon in tasks:
if addon.state in {AddonState.ERROR, AddonState.UNKNOWN}:
self.sys_resolution.add_issue(
evolve(addon.boot_failed_issue),
suggestions=[
SuggestionType.EXECUTE_START,
SuggestionType.DISABLE_BOOT,
],
)
async def shutdown(self, stage: AddonStartup) -> None: async def shutdown(self, stage: AddonStartup) -> None:
"""Shutdown addons.""" """Shutdown addons."""
tasks: list[Addon] = [] tasks: list[Addon] = []

View File

@ -6,6 +6,8 @@ from dataclasses import dataclass
import logging import logging
from pathlib import PurePath from pathlib import PurePath
from attr import evolve
from ..const import ATTR_NAME from ..const import ATTR_NAME
from ..coresys import CoreSys, CoreSysAttributes from ..coresys import CoreSys, CoreSysAttributes
from ..dbus.const import UnitActiveState from ..dbus.const import UnitActiveState
@ -171,7 +173,7 @@ class MountManager(FileConfiguration, CoreSysAttributes):
capture_exception(errors[i]) capture_exception(errors[i])
self.sys_resolution.add_issue( self.sys_resolution.add_issue(
mounts[i].failed_issue, evolve(mounts[i].failed_issue),
suggestions=[ suggestions=[
SuggestionType.EXECUTE_RELOAD, SuggestionType.EXECUTE_RELOAD,
SuggestionType.EXECUTE_REMOVE, SuggestionType.EXECUTE_REMOVE,

View File

@ -68,6 +68,9 @@ class Mount(CoreSysAttributes, ABC):
self._data: MountData = data self._data: MountData = data
self._unit: SystemdUnit | None = None self._unit: SystemdUnit | None = None
self._state: UnitActiveState | None = None self._state: UnitActiveState | None = None
self._failed_issue = Issue(
IssueType.MOUNT_FAILED, ContextType.MOUNT, reference=self.name
)
@classmethod @classmethod
def from_dict(cls, coresys: CoreSys, data: MountData) -> "Mount": def from_dict(cls, coresys: CoreSys, data: MountData) -> "Mount":
@ -162,7 +165,7 @@ class Mount(CoreSysAttributes, ABC):
@property @property
def failed_issue(self) -> Issue: def failed_issue(self) -> Issue:
"""Get issue used if this mount has failed.""" """Get issue used if this mount has failed."""
return Issue(IssueType.MOUNT_FAILED, ContextType.MOUNT, reference=self.name) return self._failed_issue
async def is_mounted(self) -> bool: async def is_mounted(self) -> bool:
"""Return true if successfully mounted and available.""" """Return true if successfully mounted and available."""

View File

@ -71,6 +71,7 @@ class UnhealthyReason(StrEnum):
class IssueType(StrEnum): class IssueType(StrEnum):
"""Issue type.""" """Issue type."""
BOOT_FAIL = "boot_fail"
CORRUPT_DOCKER = "corrupt_docker" CORRUPT_DOCKER = "corrupt_docker"
CORRUPT_REPOSITORY = "corrupt_repository" CORRUPT_REPOSITORY = "corrupt_repository"
CORRUPT_FILESYSTEM = "corrupt_filesystem" CORRUPT_FILESYSTEM = "corrupt_filesystem"
@ -103,6 +104,7 @@ class SuggestionType(StrEnum):
ADOPT_DATA_DISK = "adopt_data_disk" ADOPT_DATA_DISK = "adopt_data_disk"
CLEAR_FULL_BACKUP = "clear_full_backup" CLEAR_FULL_BACKUP = "clear_full_backup"
CREATE_FULL_BACKUP = "create_full_backup" CREATE_FULL_BACKUP = "create_full_backup"
DISABLE_BOOT = "disable_boot"
EXECUTE_INTEGRITY = "execute_integrity" EXECUTE_INTEGRITY = "execute_integrity"
EXECUTE_REBOOT = "execute_reboot" EXECUTE_REBOOT = "execute_reboot"
EXECUTE_REBUILD = "execute_rebuild" EXECUTE_REBUILD = "execute_rebuild"
@ -110,6 +112,7 @@ class SuggestionType(StrEnum):
EXECUTE_REMOVE = "execute_remove" EXECUTE_REMOVE = "execute_remove"
EXECUTE_REPAIR = "execute_repair" EXECUTE_REPAIR = "execute_repair"
EXECUTE_RESET = "execute_reset" EXECUTE_RESET = "execute_reset"
EXECUTE_START = "execute_start"
EXECUTE_STOP = "execute_stop" EXECUTE_STOP = "execute_stop"
EXECUTE_UPDATE = "execute_update" EXECUTE_UPDATE = "execute_update"
REGISTRY_LOGIN = "registry_login" REGISTRY_LOGIN = "registry_login"

View File

@ -0,0 +1,48 @@
"""Helpers to fix addon by disabling boot."""
import logging
from ...const import AddonBoot
from ...coresys import CoreSys
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 FixupAddonDisableBoot(coresys)
class FixupAddonDisableBoot(FixupBase):
"""Storage class for fixup."""
async def process_fixup(self, reference: str | None = None) -> None:
"""Initialize the fixup class."""
if not (addon := self.sys_addons.get(reference, local_only=True)):
_LOGGER.info("Cannot change addon %s as it does not exist", reference)
return
# Disable boot on addon
addon.boot = AddonBoot.MANUAL
@property
def suggestion(self) -> SuggestionType:
"""Return a SuggestionType enum."""
return SuggestionType.DISABLE_BOOT
@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.BOOT_FAIL]
@property
def auto(self) -> bool:
"""Return if a fixup can be apply as auto fix."""
return False

View File

@ -0,0 +1,59 @@
"""Helpers to fix addon by starting it."""
import logging
from ...const import AddonState
from ...coresys import CoreSys
from ...exceptions import AddonsError, ResolutionFixupError
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 FixupAddonExecuteStart(coresys)
class FixupAddonExecuteStart(FixupBase):
"""Storage class for fixup."""
async def process_fixup(self, reference: str | None = None) -> None:
"""Initialize the fixup class."""
if not (addon := self.sys_addons.get(reference, local_only=True)):
_LOGGER.info("Cannot start addon %s as it does not exist", reference)
return
# Start addon
try:
start_task = await addon.start()
except AddonsError as err:
_LOGGER.error("Could not start %s due to %s", reference, err)
raise ResolutionFixupError() from None
# Wait for addon start. If it ends up in error or unknown state it's not fixed
await start_task
if addon.state in {AddonState.ERROR, AddonState.UNKNOWN}:
_LOGGER.error("Addon %s could not start successfully", reference)
raise ResolutionFixupError()
@property
def suggestion(self) -> SuggestionType:
"""Return a SuggestionType enum."""
return SuggestionType.EXECUTE_START
@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.BOOT_FAIL]
@property
def auto(self) -> bool:
"""Return if a fixup can be apply as auto fix."""
return False

View File

@ -14,7 +14,7 @@ from securetar import SecureTarFile
from supervisor.addons.addon import Addon from supervisor.addons.addon import Addon
from supervisor.addons.const import AddonBackupMode from supervisor.addons.const import AddonBackupMode
from supervisor.addons.model import AddonModel from supervisor.addons.model import AddonModel
from supervisor.const import AddonState, BusEvent from supervisor.const import AddonBoot, AddonState, BusEvent
from supervisor.coresys import CoreSys from supervisor.coresys import CoreSys
from supervisor.docker.addon import DockerAddon from supervisor.docker.addon import DockerAddon
from supervisor.docker.const import ContainerState from supervisor.docker.const import ContainerState
@ -24,6 +24,8 @@ from supervisor.ingress import Ingress
from supervisor.store.repository import Repository from supervisor.store.repository import Repository
from supervisor.utils.dt import utcnow from supervisor.utils.dt import utcnow
from .test_manager import BOOT_FAIL_ISSUE, BOOT_FAIL_SUGGESTIONS
from tests.common import get_fixture_path from tests.common import get_fixture_path
from tests.const import TEST_ADDON_SLUG from tests.const import TEST_ADDON_SLUG
@ -895,3 +897,32 @@ async def test_addon_manual_only_boot(coresys: CoreSys, install_addon_example: A
# However boot mode can change on update and user may have set auto before, ensure it is ignored # However boot mode can change on update and user may have set auto before, ensure it is ignored
install_addon_example.boot = "auto" install_addon_example.boot = "auto"
assert install_addon_example.boot == "manual" assert install_addon_example.boot == "manual"
async def test_addon_start_dismisses_boot_fail(
coresys: CoreSys, install_addon_ssh: Addon
):
"""Test a successful start dismisses the boot fail issue."""
install_addon_ssh.state = AddonState.ERROR
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE, [suggestion.type for suggestion in BOOT_FAIL_SUGGESTIONS]
)
install_addon_ssh.state = AddonState.STARTED
assert coresys.resolution.issues == []
assert coresys.resolution.suggestions == []
async def test_addon_disable_boot_dismisses_boot_fail(
coresys: CoreSys, install_addon_ssh: Addon
):
"""Test a disabling boot dismisses the boot fail issue."""
install_addon_ssh.boot = AddonBoot.AUTO
install_addon_ssh.state = AddonState.ERROR
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE, [suggestion.type for suggestion in BOOT_FAIL_SUGGESTIONS]
)
install_addon_ssh.boot = AddonBoot.MANUAL
assert coresys.resolution.issues == []
assert coresys.resolution.suggestions == []

View File

@ -1,6 +1,7 @@
"""Test addon manager.""" """Test addon manager."""
import asyncio import asyncio
from collections.abc import AsyncGenerator, Generator
from copy import deepcopy from copy import deepcopy
from pathlib import Path from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, Mock, PropertyMock, patch from unittest.mock import AsyncMock, MagicMock, Mock, PropertyMock, patch
@ -25,6 +26,8 @@ from supervisor.exceptions import (
DockerNotFound, DockerNotFound,
) )
from supervisor.plugins.dns import PluginDns from supervisor.plugins.dns import PluginDns
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
from supervisor.resolution.data import Issue, Suggestion
from supervisor.store.addon import AddonStore from supervisor.store.addon import AddonStore
from supervisor.store.repository import Repository from supervisor.store.repository import Repository
from supervisor.utils import check_exception_chain from supervisor.utils import check_exception_chain
@ -33,9 +36,21 @@ from supervisor.utils.common import write_json_file
from tests.common import load_json_fixture from tests.common import load_json_fixture
from tests.const import TEST_ADDON_SLUG from tests.const import TEST_ADDON_SLUG
BOOT_FAIL_ISSUE = Issue(
IssueType.BOOT_FAIL, ContextType.ADDON, reference=TEST_ADDON_SLUG
)
BOOT_FAIL_SUGGESTIONS = [
Suggestion(
SuggestionType.EXECUTE_START, ContextType.ADDON, reference=TEST_ADDON_SLUG
),
Suggestion(
SuggestionType.DISABLE_BOOT, ContextType.ADDON, reference=TEST_ADDON_SLUG
),
]
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
async def fixture_mock_arch_disk() -> None: async def fixture_mock_arch_disk() -> AsyncGenerator[None]:
"""Mock supported arch and disk space.""" """Mock supported arch and disk space."""
with ( with (
patch("shutil.disk_usage", return_value=(42, 42, 2 * (1024.0**3))), patch("shutil.disk_usage", return_value=(42, 42, 2 * (1024.0**3))),
@ -45,13 +60,15 @@ async def fixture_mock_arch_disk() -> None:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
async def fixture_remove_wait_boot(coresys: CoreSys) -> None: async def fixture_remove_wait_boot(coresys: CoreSys) -> AsyncGenerator[None]:
"""Remove default wait boot time for tests.""" """Remove default wait boot time for tests."""
coresys.config.wait_boot = 0 coresys.config.wait_boot = 0
@pytest.fixture(name="install_addon_example_image") @pytest.fixture(name="install_addon_example_image")
def fixture_install_addon_example_image(coresys: CoreSys, repository) -> Addon: def fixture_install_addon_example_image(
coresys: CoreSys, repository
) -> Generator[Addon]:
"""Install local_example add-on with image.""" """Install local_example add-on with image."""
store = coresys.addons.store["local_example_image"] store = coresys.addons.store["local_example_image"]
coresys.addons.data.install(store) coresys.addons.data.install(store)
@ -114,14 +131,17 @@ async def test_addon_boot_system_error(
): ):
"""Test system errors during addon boot.""" """Test system errors during addon boot."""
install_addon_ssh.boot = AddonBoot.AUTO install_addon_ssh.boot = AddonBoot.AUTO
assert coresys.resolution.issues == []
assert coresys.resolution.suggestions == []
with ( with (
patch.object(Addon, "write_options"), patch.object(Addon, "write_options"),
patch.object(DockerAddon, "run", side_effect=err), patch.object(DockerAddon, "run", side_effect=err),
): ):
await coresys.addons.boot(AddonStartup.APPLICATION) await coresys.addons.boot(AddonStartup.APPLICATION)
assert install_addon_ssh.boot == AddonBoot.MANUAL
capture_exception.assert_not_called() capture_exception.assert_not_called()
assert coresys.resolution.issues == [BOOT_FAIL_ISSUE]
assert coresys.resolution.suggestions == BOOT_FAIL_SUGGESTIONS
async def test_addon_boot_user_error( async def test_addon_boot_user_error(
@ -132,8 +152,9 @@ async def test_addon_boot_user_error(
with patch.object(Addon, "write_options", side_effect=AddonConfigurationError): with patch.object(Addon, "write_options", side_effect=AddonConfigurationError):
await coresys.addons.boot(AddonStartup.APPLICATION) await coresys.addons.boot(AddonStartup.APPLICATION)
assert install_addon_ssh.boot == AddonBoot.MANUAL
capture_exception.assert_not_called() capture_exception.assert_not_called()
assert coresys.resolution.issues == [BOOT_FAIL_ISSUE]
assert coresys.resolution.suggestions == BOOT_FAIL_SUGGESTIONS
async def test_addon_boot_other_error( async def test_addon_boot_other_error(
@ -148,8 +169,9 @@ async def test_addon_boot_other_error(
): ):
await coresys.addons.boot(AddonStartup.APPLICATION) await coresys.addons.boot(AddonStartup.APPLICATION)
assert install_addon_ssh.boot == AddonBoot.AUTO
capture_exception.assert_called_once_with(err) capture_exception.assert_called_once_with(err)
assert coresys.resolution.issues == [BOOT_FAIL_ISSUE]
assert coresys.resolution.suggestions == BOOT_FAIL_SUGGESTIONS
async def test_addon_shutdown_error( async def test_addon_shutdown_error(

View File

@ -0,0 +1,40 @@
"""Test fixup addon disable boot."""
from supervisor.addons.addon import Addon
from supervisor.const import AddonBoot
from supervisor.coresys import CoreSys
from supervisor.resolution.const import SuggestionType
from supervisor.resolution.fixups.addon_disable_boot import FixupAddonDisableBoot
from tests.addons.test_manager import BOOT_FAIL_ISSUE
async def test_fixup(coresys: CoreSys, install_addon_ssh: Addon):
"""Test fixup disables boot."""
install_addon_ssh.boot = AddonBoot.AUTO
addon_disable_boot = FixupAddonDisableBoot(coresys)
assert addon_disable_boot.auto is False
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE,
suggestions=[SuggestionType.DISABLE_BOOT],
)
await addon_disable_boot()
assert install_addon_ssh.boot == AddonBoot.MANUAL
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
async def test_fixup_no_addon(coresys: CoreSys):
"""Test fixup dismisses if addon is missing."""
addon_disable_boot = FixupAddonDisableBoot(coresys)
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE,
suggestions=[SuggestionType.DISABLE_BOOT],
)
await addon_disable_boot()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions

View File

@ -1,4 +1,4 @@
"""Test fixup core execute repair.""" """Test fixup addon execute repair."""
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch

View File

@ -0,0 +1,117 @@
"""Test fixup addon execute start."""
from unittest.mock import patch
import pytest
from supervisor.addons.addon import Addon
from supervisor.const import AddonState
from supervisor.coresys import CoreSys
from supervisor.docker.addon import DockerAddon
from supervisor.exceptions import DockerError
from supervisor.resolution.const import ContextType, SuggestionType
from supervisor.resolution.data import Suggestion
from supervisor.resolution.fixups.addon_execute_start import FixupAddonExecuteStart
from tests.addons.test_manager import BOOT_FAIL_ISSUE
EXECUTE_START_SUGGESTION = Suggestion(
SuggestionType.EXECUTE_START, ContextType.ADDON, reference="local_ssh"
)
@pytest.mark.parametrize(
"state", [AddonState.STARTED, AddonState.STARTUP, AddonState.STOPPED]
)
@pytest.mark.usefixtures("path_extern")
async def test_fixup(coresys: CoreSys, install_addon_ssh: Addon, state: AddonState):
"""Test fixup starts addon."""
install_addon_ssh.state = AddonState.UNKNOWN
addon_execute_start = FixupAddonExecuteStart(coresys)
assert addon_execute_start.auto is False
async def mock_start(*args, **kwargs):
install_addon_ssh.state = state
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE,
suggestions=[SuggestionType.EXECUTE_START],
)
with (
patch.object(DockerAddon, "run") as run,
patch.object(Addon, "_wait_for_startup", new=mock_start),
patch.object(Addon, "write_options"),
):
await addon_execute_start()
run.assert_called_once()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions
@pytest.mark.usefixtures("path_extern")
async def test_fixup_start_error(coresys: CoreSys, install_addon_ssh: Addon):
"""Test fixup fails on start addon failure."""
install_addon_ssh.state = AddonState.UNKNOWN
addon_execute_start = FixupAddonExecuteStart(coresys)
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE,
suggestions=[SuggestionType.EXECUTE_START],
)
with (
patch.object(DockerAddon, "run", side_effect=DockerError) as run,
patch.object(Addon, "write_options"),
):
await addon_execute_start()
run.assert_called_once()
assert BOOT_FAIL_ISSUE in coresys.resolution.issues
assert EXECUTE_START_SUGGESTION in coresys.resolution.suggestions
@pytest.mark.parametrize("state", [AddonState.ERROR, AddonState.UNKNOWN])
@pytest.mark.usefixtures("path_extern")
async def test_fixup_wait_start_failure(
coresys: CoreSys, install_addon_ssh: Addon, state: AddonState
):
"""Test fixup fails if addon does not complete startup."""
install_addon_ssh.state = AddonState.UNKNOWN
addon_execute_start = FixupAddonExecuteStart(coresys)
async def mock_start(*args, **kwargs):
install_addon_ssh.state = state
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE,
suggestions=[SuggestionType.EXECUTE_START],
)
with (
patch.object(DockerAddon, "run") as run,
patch.object(Addon, "_wait_for_startup", new=mock_start),
patch.object(Addon, "write_options"),
):
await addon_execute_start()
run.assert_called_once()
assert BOOT_FAIL_ISSUE in coresys.resolution.issues
assert EXECUTE_START_SUGGESTION in coresys.resolution.suggestions
async def test_fixup_no_addon(coresys: CoreSys):
"""Test fixup dismisses if addon is missing."""
addon_execute_start = FixupAddonExecuteStart(coresys)
coresys.resolution.add_issue(
BOOT_FAIL_ISSUE,
suggestions=[SuggestionType.EXECUTE_START],
)
with (
patch.object(DockerAddon, "run") as run,
patch.object(Addon, "write_options"),
):
await addon_execute_start()
run.assert_not_called()
assert not coresys.resolution.issues
assert not coresys.resolution.suggestions