Resolution: extend type and context (#2130)

* Resolution: extend type and context

* fix property

* add helper

* fix api

* fix tests

* Fix patch

* finish tests

* Update supervisor/resolution/const.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Update supervisor/resolution/const.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Fix type

* fix lint

* Update supervisor/api/resolution.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Update supervisor/resolution/__init__.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Update API & add more tests

* Update supervisor/api/resolution.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Update supervisor/resolution/__init__.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Update supervisor/resolution/__init__.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* fix black

* remove azure ci

* fix test

* fix tests

* fix tests

* fix tests p2

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>
This commit is contained in:
Pascal Vizeli 2020-10-16 12:22:32 +02:00 committed by GitHub
parent fe0e41adec
commit d119e99001
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 323 additions and 116 deletions

View File

@ -1,52 +0,0 @@
# https://dev.azure.com/home-assistant
trigger:
batch: true
branches:
include:
- master
- dev
pr:
- dev
variables:
- name: versionHadolint
value: "v1.16.3"
jobs:
- job: "Tox"
pool:
vmImage: "ubuntu-latest"
steps:
- script: |
sudo apt-get update
sudo apt-get install -y libpulse0 libudev1
displayName: "Install Host library"
- task: UsePythonVersion@0
displayName: "Use Python 3.8"
inputs:
versionSpec: "3.8"
- script: pip install tox
displayName: "Install Tox"
- script: tox
displayName: "Run Tox"
- job: "JQ"
pool:
vmImage: "ubuntu-latest"
steps:
- script: sudo apt-get install -y jq
displayName: "Install JQ"
- bash: |
shopt -s globstar
cat **/*.json | jq '.'
displayName: "Run JQ"
- job: "Hadolint"
pool:
vmImage: "ubuntu-latest"
steps:
- script: sudo docker pull hadolint/hadolint:$(versionHadolint)
displayName: "Install Hadolint"
- script: |
sudo docker run --rm -i \
-v $(pwd)/.hadolint.yaml:/.hadolint.yaml:ro \
hadolint/hadolint:$(versionHadolint) < Dockerfile
displayName: "Run Hadolint"

View File

@ -200,11 +200,18 @@ class RestAPI(CoreSysAttributes):
self.webapp.add_routes( self.webapp.add_routes(
[ [
web.get("/resolution", api_resolution.base), web.get("/resolution", api_resolution.base),
web.post("/resolution/{suggestion}", api_resolution.apply_suggestion),
web.post( web.post(
"/resolution/{suggestion}/dismiss", "/resolution/suggestion/{suggestion}",
api_resolution.apply_suggestion,
),
web.delete(
"/resolution/suggestion/{suggestion}",
api_resolution.dismiss_suggestion, api_resolution.dismiss_suggestion,
), ),
web.delete(
"/resolution/issue/{issue}",
api_resolution.dismiss_issue,
),
] ]
) )

View File

@ -2,11 +2,11 @@
from typing import Any, Dict from typing import Any, Dict
from aiohttp import web from aiohttp import web
import attr
from ..const import ATTR_ISSUES, ATTR_SUGGESTIONS, ATTR_UNSUPPORTED from ..const import ATTR_ISSUES, ATTR_SUGGESTIONS, ATTR_UNSUPPORTED
from ..coresys import CoreSysAttributes from ..coresys import CoreSysAttributes
from ..exceptions import APIError from ..exceptions import APIError, ResolutionNotFound
from ..resolution.const import Suggestion
from .utils import api_process from .utils import api_process
@ -18,24 +18,40 @@ class APIResoulution(CoreSysAttributes):
"""Return network information.""" """Return network information."""
return { return {
ATTR_UNSUPPORTED: self.sys_resolution.unsupported, ATTR_UNSUPPORTED: self.sys_resolution.unsupported,
ATTR_SUGGESTIONS: self.sys_resolution.suggestions, ATTR_SUGGESTIONS: [
ATTR_ISSUES: self.sys_resolution.issues, attr.asdict(suggestion)
for suggestion in self.sys_resolution.suggestions
],
ATTR_ISSUES: [attr.asdict(issue) for issue in self.sys_resolution.issues],
} }
@api_process @api_process
async def apply_suggestion(self, request: web.Request) -> None: async def apply_suggestion(self, request: web.Request) -> None:
"""Apply suggestion.""" """Apply suggestion."""
try: try:
suggestion = Suggestion(request.match_info.get("suggestion")) suggestion = self.sys_resolution.get_suggestion(
request.match_info.get("suggestion")
)
await self.sys_resolution.apply_suggestion(suggestion) await self.sys_resolution.apply_suggestion(suggestion)
except ValueError: except ResolutionNotFound:
raise APIError("Suggestion is not valid") from None raise APIError("The supplied UUID is not a valid suggestion") from None
@api_process @api_process
async def dismiss_suggestion(self, request: web.Request) -> None: async def dismiss_suggestion(self, request: web.Request) -> None:
"""Dismiss suggestion.""" """Dismiss suggestion."""
try: try:
suggestion = Suggestion(request.match_info.get("suggestion")) suggestion = self.sys_resolution.get_suggestion(
request.match_info.get("suggestion")
)
await self.sys_resolution.dismiss_suggestion(suggestion) await self.sys_resolution.dismiss_suggestion(suggestion)
except ValueError: except ResolutionNotFound:
raise APIError("Suggestion is not valid") from None raise APIError("The supplied UUID is not a valid suggestion") from None
@api_process
async def dismiss_issue(self, request: web.Request) -> None:
"""Dismiss issue."""
try:
issue = self.sys_resolution.get_issue(request.match_info.get("issue"))
await self.sys_resolution.dismiss_issue(issue)
except ResolutionNotFound:
raise APIError("The supplied UUID is not a valid issue") from None

View File

@ -22,7 +22,7 @@ from .exceptions import (
HomeAssistantError, HomeAssistantError,
SupervisorUpdateError, SupervisorUpdateError,
) )
from .resolution.const import UnsupportedReason from .resolution.const import ContextType, IssueType, UnsupportedReason
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -107,6 +107,9 @@ class Core(CoreSysAttributes):
self.sys_updater.channel, self.sys_updater.channel,
) )
elif self.sys_config.version != self.sys_supervisor.version: elif self.sys_config.version != self.sys_supervisor.version:
self.sys_resolution.create_issue(
IssueType.UPDATE_ROLLBACK, ContextType.SUPERVISOR
)
self.healthy = False self.healthy = False
_LOGGER.error( _LOGGER.error(
"Update '%s' of Supervisor '%s' failed!", "Update '%s' of Supervisor '%s' failed!",

View File

@ -270,3 +270,14 @@ class HardwareNotSupportedError(HassioNotSupportedError):
class PulseAudioError(HassioError): class PulseAudioError(HassioError):
"""Raise if an sound error is happening.""" """Raise if an sound error is happening."""
# Resolution
class ResolutionError(HassioError):
"""Raise if an error is happning on resoltuion."""
class ResolutionNotFound(ResolutionError):
"""Raise if suggestion/issue was not found."""

View File

@ -21,6 +21,7 @@ from ..exceptions import (
HomeAssistantError, HomeAssistantError,
HomeAssistantUpdateError, HomeAssistantUpdateError,
) )
from ..resolution.const import ContextType, IssueType
from ..utils import convert_to_ascii, process_lock from ..utils import convert_to_ascii, process_lock
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -191,6 +192,10 @@ class HomeAssistantCore(CoreSysAttributes):
# Update going wrong, revert it # Update going wrong, revert it
if self.error_state and rollback: if self.error_state and rollback:
_LOGGER.critical("HomeAssistant update failed -> rollback!") _LOGGER.critical("HomeAssistant update failed -> rollback!")
self.sys_resolution.create_issue(
IssueType.UPDATE_ROLLBACK, ContextType.CORE
)
# Make a copy of the current log file if it exsist # Make a copy of the current log file if it exsist
logfile = self.sys_config.path_homeassistant / "home-assistant.log" logfile = self.sys_config.path_homeassistant / "home-assistant.log"
if logfile.exists(): if logfile.exists():
@ -204,6 +209,7 @@ class HomeAssistantCore(CoreSysAttributes):
) )
await _update(rollback) await _update(rollback)
else: else:
self.sys_resolution.create_issue(IssueType.UPDATE_FAILED, ContextType.CORE)
raise HomeAssistantUpdateError() raise HomeAssistantUpdateError()
async def _start(self) -> None: async def _start(self) -> None:

View File

@ -1,10 +1,17 @@
"""Supervisor resolution center.""" """Supervisor resolution center."""
import logging import logging
from typing import List from typing import List, Optional
from ..coresys import CoreSys, CoreSysAttributes from ..coresys import CoreSys, CoreSysAttributes
from ..resolution.const import UnsupportedReason from ..exceptions import ResolutionError, ResolutionNotFound
from .const import SCHEDULED_HEALTHCHECK, IssueType, Suggestion from .const import (
SCHEDULED_HEALTHCHECK,
ContextType,
IssueType,
SuggestionType,
UnsupportedReason,
)
from .data import Issue, Suggestion
from .free_space import ResolutionStorage from .free_space import ResolutionStorage
from .notify import ResolutionNotify from .notify import ResolutionNotify
@ -19,9 +26,9 @@ class ResolutionManager(CoreSysAttributes):
self.coresys: CoreSys = coresys self.coresys: CoreSys = coresys
self._notify = ResolutionNotify(coresys) self._notify = ResolutionNotify(coresys)
self._storage = ResolutionStorage(coresys) self._storage = ResolutionStorage(coresys)
self._dismissed_suggestions: List[Suggestion] = []
self._suggestions: List[Suggestion] = [] self._suggestions: List[Suggestion] = []
self._issues: List[IssueType] = [] self._issues: List[Issue] = []
self._unsupported: List[UnsupportedReason] = [] self._unsupported: List[UnsupportedReason] = []
@property @property
@ -35,12 +42,12 @@ class ResolutionManager(CoreSysAttributes):
return self._notify return self._notify
@property @property
def issues(self) -> List[IssueType]: def issues(self) -> List[Issue]:
"""Return a list of issues.""" """Return a list of issues."""
return self._issues return self._issues
@issues.setter @issues.setter
def issues(self, issue: IssueType) -> None: def issues(self, issue: Issue) -> None:
"""Add issues.""" """Add issues."""
if issue not in self._issues: if issue not in self._issues:
self._issues.append(issue) self._issues.append(issue)
@ -48,7 +55,7 @@ class ResolutionManager(CoreSysAttributes):
@property @property
def suggestions(self) -> List[Suggestion]: def suggestions(self) -> List[Suggestion]:
"""Return a list of suggestions that can handled.""" """Return a list of suggestions that can handled."""
return [x for x in self._suggestions if x not in self._dismissed_suggestions] return self._suggestions
@suggestions.setter @suggestions.setter
def suggestions(self, suggestion: Suggestion) -> None: def suggestions(self, suggestion: Suggestion) -> None:
@ -67,6 +74,38 @@ class ResolutionManager(CoreSysAttributes):
if reason not in self._unsupported: if reason not in self._unsupported:
self._unsupported.append(reason) self._unsupported.append(reason)
def get_suggestion(self, uuid: str) -> Suggestion:
"""Return suggestion with uuid."""
for suggestion in self._suggestions:
if suggestion.uuid != uuid:
continue
return suggestion
raise ResolutionNotFound()
def get_issue(self, uuid: str) -> Issue:
"""Return issue with uuid."""
for issue in self._issues:
if issue.uuid != uuid:
continue
return issue
raise ResolutionNotFound()
def create_issue(
self,
issue: IssueType,
context: ContextType,
reference: Optional[str] = None,
suggestions: Optional[List[SuggestionType]] = None,
) -> None:
"""Create issues and suggestion."""
self.issues = Issue(issue, context, reference)
if not suggestions:
return
# Add suggestions
for suggestion in suggestions:
self.suggestions = Suggestion(suggestion, context, reference)
async def load(self): async def load(self):
"""Load the resoulution manager.""" """Load the resoulution manager."""
# Initial healthcheck when the manager is loaded # Initial healthcheck when the manager is loaded
@ -85,14 +124,14 @@ class ResolutionManager(CoreSysAttributes):
async def apply_suggestion(self, suggestion: Suggestion) -> None: async def apply_suggestion(self, suggestion: Suggestion) -> None:
"""Apply suggested action.""" """Apply suggested action."""
if suggestion not in self.suggestions: if suggestion not in self._suggestions:
_LOGGER.warning("Suggestion %s is not valid", suggestion) _LOGGER.warning("Suggestion %s is not valid", suggestion.uuid)
return raise ResolutionError()
if suggestion == Suggestion.CLEAR_FULL_SNAPSHOT: if suggestion.type == SuggestionType.CLEAR_FULL_SNAPSHOT:
self.storage.clean_full_snapshots() self.storage.clean_full_snapshots()
elif suggestion == Suggestion.CREATE_FULL_SNAPSHOT: elif suggestion.type == SuggestionType.CREATE_FULL_SNAPSHOT:
await self.sys_snapshots.do_snapshot_full() await self.sys_snapshots.do_snapshot_full()
self._suggestions.remove(suggestion) self._suggestions.remove(suggestion)
@ -100,9 +139,14 @@ class ResolutionManager(CoreSysAttributes):
async def dismiss_suggestion(self, suggestion: Suggestion) -> None: async def dismiss_suggestion(self, suggestion: Suggestion) -> None:
"""Dismiss suggested action.""" """Dismiss suggested action."""
if suggestion not in self.suggestions: if suggestion not in self._suggestions:
_LOGGER.warning("Suggestion %s is not valid", suggestion) _LOGGER.warning("The UUID %s is not valid suggestion", suggestion.uuid)
return raise ResolutionError()
self._suggestions.remove(suggestion)
if suggestion not in self._dismissed_suggestions: async def dismiss_issue(self, issue: Issue) -> None:
self._dismissed_suggestions.append(suggestion) """Dismiss suggested action."""
if issue not in self._issues:
_LOGGER.warning("The UUID %s is not a valid issue", issue.uuid)
raise ResolutionError()
self._issues.remove(issue)

View File

@ -7,6 +7,16 @@ MINIMUM_FREE_SPACE_THRESHOLD = 1
MINIMUM_FULL_SNAPSHOTS = 2 MINIMUM_FULL_SNAPSHOTS = 2
class ContextType(str, Enum):
"""Place where somethings was happening."""
SYSTEM = "system"
SUPERVISOR = "supervisor"
ADDON = "addon"
CORE = "core"
OS = "os"
class UnsupportedReason(str, Enum): class UnsupportedReason(str, Enum):
"""Reasons for unsupported status.""" """Reasons for unsupported status."""
@ -25,10 +35,15 @@ class IssueType(str, Enum):
"""Issue type.""" """Issue type."""
FREE_SPACE = "free_space" FREE_SPACE = "free_space"
CORRUPT_DOCKER = "corrupt_docker"
MISSING_IMAGE = "missing_image"
UPDATE_FAILED = "update_failed"
UPDATE_ROLLBACK = "update_rollback"
class Suggestion(str, Enum): class SuggestionType(str, Enum):
"""Sugestion.""" """Sugestion type."""
CLEAR_FULL_SNAPSHOT = "clear_full_snapshot" CLEAR_FULL_SNAPSHOT = "clear_full_snapshot"
CREATE_FULL_SNAPSHOT = "create_full_snapshot" CREATE_FULL_SNAPSHOT = "create_full_snapshot"
SYSTEM_REPAIR = "system_repair"

View File

@ -0,0 +1,27 @@
"""Data objects."""
from typing import Optional
from uuid import UUID, uuid4
import attr
from .const import ContextType, IssueType, SuggestionType
@attr.s(frozen=True, slots=True)
class Issue:
"""Represent an Issue."""
type: IssueType = attr.ib()
context: ContextType = attr.ib()
reference: Optional[str] = attr.ib(default=None)
uuid: UUID = attr.ib(factory=lambda: uuid4().hex, eq=False, init=False)
@attr.s(frozen=True, slots=True)
class Suggestion:
"""Represent an Suggestion."""
type: SuggestionType = attr.ib()
context: ContextType = attr.ib()
reference: Optional[str] = attr.ib(default=None)
uuid: UUID = attr.ib(factory=lambda: uuid4().hex, eq=False, init=False)

View File

@ -1,14 +1,17 @@
"""Helpers to check and fix issues with free space.""" """Helpers to check and fix issues with free space."""
import logging import logging
from typing import List
from ..const import SNAPSHOT_FULL from ..const import SNAPSHOT_FULL
from ..coresys import CoreSys, CoreSysAttributes from ..coresys import CoreSys, CoreSysAttributes
from .const import ( from .const import (
MINIMUM_FREE_SPACE_THRESHOLD, MINIMUM_FREE_SPACE_THRESHOLD,
MINIMUM_FULL_SNAPSHOTS, MINIMUM_FULL_SNAPSHOTS,
ContextType,
IssueType, IssueType,
Suggestion, SuggestionType,
) )
from .data import Suggestion
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -23,10 +26,14 @@ class ResolutionStorage(CoreSysAttributes):
def check_free_space(self) -> None: def check_free_space(self) -> None:
"""Check free space.""" """Check free space."""
if self.sys_host.info.free_space > MINIMUM_FREE_SPACE_THRESHOLD: if self.sys_host.info.free_space > MINIMUM_FREE_SPACE_THRESHOLD:
if len(self.sys_snapshots.list_snapshots) == 0:
# No snapshots, let's suggest the user to create one!
self.sys_resolution.suggestions = Suggestion(
SuggestionType.CREATE_FULL_SNAPSHOT, ContextType.SYSTEM
)
return return
self.sys_resolution.issues = IssueType.FREE_SPACE suggestions: List[SuggestionType] = []
if ( if (
len( len(
[ [
@ -37,11 +44,11 @@ class ResolutionStorage(CoreSysAttributes):
) )
>= MINIMUM_FULL_SNAPSHOTS >= MINIMUM_FULL_SNAPSHOTS
): ):
self.sys_resolution.suggestions = Suggestion.CLEAR_FULL_SNAPSHOT suggestions.append(SuggestionType.CLEAR_FULL_SNAPSHOT)
elif len(self.sys_snapshots.list_snapshots) == 0: self.sys_resolution.create_issue(
# No snapshots, let's suggest the user to create one! IssueType.FREE_SPACE, ContextType.SYSTEM, suggestions=suggestions
self.sys_resolution.suggestions = Suggestion.CREATE_FULL_SNAPSHOT )
def clean_full_snapshots(self): def clean_full_snapshots(self):
"""Clean out all old full snapshots, but keep the most recent.""" """Clean out all old full snapshots, but keep the most recent."""

View File

@ -28,11 +28,11 @@ class ResolutionNotify(CoreSysAttributes):
): ):
return return
issues = [] messages = []
for issue in self.sys_resolution.issues: for issue in self.sys_resolution.issues:
if issue == IssueType.FREE_SPACE: if issue.type == IssueType.FREE_SPACE:
issues.append( messages.append(
{ {
"title": "Available space is less than 1GB!", "title": "Available space is less than 1GB!",
"message": f"Available space is {self.sys_host.info.free_space}GB, see https://www.home-assistant.io/more-info/free-space for more information.", "message": f"Available space is {self.sys_host.info.free_space}GB, see https://www.home-assistant.io/more-info/free-space for more information.",
@ -40,12 +40,12 @@ class ResolutionNotify(CoreSysAttributes):
} }
) )
for issue in issues: for message in messages:
try: try:
async with self.sys_homeassistant.api.make_request( async with self.sys_homeassistant.api.make_request(
"post", "post",
"api/services/persistent_notification/create", "api/services/persistent_notification/create",
json=issue, json=message,
) as resp: ) as resp:
if resp.status in (200, 201): if resp.status in (200, 201):
_LOGGER.debug("Sucessfully created persistent_notification") _LOGGER.debug("Sucessfully created persistent_notification")

View File

@ -19,6 +19,7 @@ from .exceptions import (
SupervisorError, SupervisorError,
SupervisorUpdateError, SupervisorUpdateError,
) )
from .resolution.const import ContextType, IssueType
_LOGGER: logging.Logger = logging.getLogger(__name__) _LOGGER: logging.Logger = logging.getLogger(__name__)
@ -117,6 +118,9 @@ class Supervisor(CoreSysAttributes):
) )
except DockerError as err: except DockerError as err:
_LOGGER.error("Update of Supervisor failed!") _LOGGER.error("Update of Supervisor failed!")
self.sys_resolution.create_issue(
IssueType.UPDATE_FAILED, ContextType.SUPERVISOR
)
raise SupervisorUpdateError() from err raise SupervisorUpdateError() from err
else: else:
self.sys_config.version = version self.sys_config.version = version

View File

@ -1,47 +1,86 @@
"""Test Resolution API.""" """Test Resolution API."""
from unittest.mock import MagicMock, patch from unittest.mock import AsyncMock
import pytest import pytest
from supervisor.const import ATTR_ISSUES, ATTR_SUGGESTIONS, ATTR_UNSUPPORTED from supervisor.const import ATTR_ISSUES, ATTR_SUGGESTIONS, ATTR_UNSUPPORTED
from supervisor.coresys import CoreSys from supervisor.coresys import CoreSys
from supervisor.resolution.const import IssueType, Suggestion, UnsupportedReason from supervisor.exceptions import ResolutionError
from supervisor.resolution.const import (
ContextType,
IssueType,
SuggestionType,
UnsupportedReason,
)
from supervisor.resolution.data import Issue, Suggestion
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_api_resolution_base(coresys: CoreSys, api_client): async def test_api_resolution_base(coresys: CoreSys, api_client):
"""Test resolution manager api.""" """Test resolution manager api."""
coresys.resolution.unsupported = UnsupportedReason.OS coresys.resolution.unsupported = UnsupportedReason.OS
coresys.resolution.suggestions = Suggestion.CLEAR_FULL_SNAPSHOT coresys.resolution.suggestions = Suggestion(
coresys.resolution.issues = IssueType.FREE_SPACE SuggestionType.CLEAR_FULL_SNAPSHOT, ContextType.SYSTEM
)
coresys.resolution.create_issue(IssueType.FREE_SPACE, ContextType.SYSTEM)
resp = await api_client.get("/resolution") resp = await api_client.get("/resolution")
result = await resp.json() result = await resp.json()
assert UnsupportedReason.OS in result["data"][ATTR_UNSUPPORTED] assert UnsupportedReason.OS in result["data"][ATTR_UNSUPPORTED]
assert Suggestion.CLEAR_FULL_SNAPSHOT in result["data"][ATTR_SUGGESTIONS] assert (
assert IssueType.FREE_SPACE in result["data"][ATTR_ISSUES] SuggestionType.CLEAR_FULL_SNAPSHOT
== result["data"][ATTR_SUGGESTIONS][-1]["type"]
)
assert IssueType.FREE_SPACE == result["data"][ATTR_ISSUES][-1]["type"]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_api_resolution_dismiss_suggestion(coresys: CoreSys, api_client): async def test_api_resolution_dismiss_suggestion(coresys: CoreSys, api_client):
"""Test resolution manager suggestion apply api.""" """Test resolution manager suggestion apply api."""
coresys.resolution.suggestions = Suggestion.CLEAR_FULL_SNAPSHOT coresys.resolution.suggestions = clear_snapshot = Suggestion(
SuggestionType.CLEAR_FULL_SNAPSHOT, ContextType.SYSTEM
)
assert Suggestion.CLEAR_FULL_SNAPSHOT in coresys.resolution.suggestions assert SuggestionType.CLEAR_FULL_SNAPSHOT == coresys.resolution.suggestions[-1].type
await coresys.resolution.dismiss_suggestion(Suggestion.CLEAR_FULL_SNAPSHOT) await api_client.delete(f"/resolution/suggestion/{clear_snapshot.uuid}")
assert Suggestion.CLEAR_FULL_SNAPSHOT not in coresys.resolution.suggestions assert clear_snapshot not in coresys.resolution.suggestions
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_api_resolution_apply_suggestion(coresys: CoreSys, api_client): async def test_api_resolution_apply_suggestion(coresys: CoreSys, api_client):
"""Test resolution manager suggestion apply api.""" """Test resolution manager suggestion apply api."""
coresys.resolution.suggestions = Suggestion.CLEAR_FULL_SNAPSHOT coresys.resolution.suggestions = clear_snapshot = Suggestion(
coresys.resolution.suggestions = Suggestion.CREATE_FULL_SNAPSHOT SuggestionType.CLEAR_FULL_SNAPSHOT, ContextType.SYSTEM
)
coresys.resolution.suggestions = create_snapshot = Suggestion(
SuggestionType.CREATE_FULL_SNAPSHOT, ContextType.SYSTEM
)
with patch("supervisor.snapshots.SnapshotManager", return_value=MagicMock()): mock_snapshots = AsyncMock()
await coresys.resolution.apply_suggestion(Suggestion.CLEAR_FULL_SNAPSHOT) mock_health = AsyncMock()
await coresys.resolution.apply_suggestion(Suggestion.CREATE_FULL_SNAPSHOT) coresys.snapshots.do_snapshot_full = mock_snapshots
coresys.resolution.healthcheck = mock_health
assert Suggestion.CLEAR_FULL_SNAPSHOT not in coresys.resolution.suggestions await api_client.post(f"/resolution/suggestion/{clear_snapshot.uuid}")
assert Suggestion.CREATE_FULL_SNAPSHOT not in coresys.resolution.suggestions await api_client.post(f"/resolution/suggestion/{create_snapshot.uuid}")
await coresys.resolution.apply_suggestion(Suggestion.CLEAR_FULL_SNAPSHOT) assert clear_snapshot not in coresys.resolution.suggestions
assert create_snapshot not in coresys.resolution.suggestions
assert mock_snapshots.called
assert mock_health.called
with pytest.raises(ResolutionError):
await coresys.resolution.apply_suggestion(clear_snapshot)
@pytest.mark.asyncio
async def test_api_resolution_dismiss_issue(coresys: CoreSys, api_client):
"""Test resolution manager issue apply api."""
coresys.resolution.issues = updated_failed = Issue(
IssueType.UPDATE_FAILED, ContextType.SYSTEM
)
assert IssueType.UPDATE_FAILED == coresys.resolution.issues[-1].type
await api_client.delete(f"/resolution/issue/{updated_failed.uuid}")
assert updated_failed not in coresys.resolution.issues

View File

@ -1,5 +1,8 @@
"""Tests for resolution manager.""" """Tests for resolution manager."""
from pathlib import Path from pathlib import Path
from unittest.mock import AsyncMock
import pytest
from supervisor.const import ( from supervisor.const import (
ATTR_DATE, ATTR_DATE,
@ -9,7 +12,14 @@ from supervisor.const import (
SNAPSHOT_PARTIAL, SNAPSHOT_PARTIAL,
) )
from supervisor.coresys import CoreSys from supervisor.coresys import CoreSys
from supervisor.resolution.const import UnsupportedReason from supervisor.exceptions import ResolutionError
from supervisor.resolution.const import (
ContextType,
IssueType,
SuggestionType,
UnsupportedReason,
)
from supervisor.resolution.data import Issue, Suggestion
from supervisor.snapshots.snapshot import Snapshot from supervisor.snapshots.snapshot import Snapshot
from supervisor.utils.dt import utcnow from supervisor.utils.dt import utcnow
from supervisor.utils.tar import SecureTarFile from supervisor.utils.tar import SecureTarFile
@ -58,3 +68,73 @@ async def test_clear_snapshots(coresys: CoreSys, tmp_path):
) )
== 1 == 1
) )
@pytest.mark.asyncio
async def test_resolution_dismiss_suggestion(coresys: CoreSys):
"""Test resolution manager suggestion apply api."""
coresys.resolution.suggestions = clear_snapshot = Suggestion(
SuggestionType.CLEAR_FULL_SNAPSHOT, ContextType.SYSTEM
)
assert SuggestionType.CLEAR_FULL_SNAPSHOT == coresys.resolution.suggestions[-1].type
await coresys.resolution.dismiss_suggestion(clear_snapshot)
assert clear_snapshot not in coresys.resolution.suggestions
@pytest.mark.asyncio
async def test_resolution_apply_suggestion(coresys: CoreSys):
"""Test resolution manager suggestion apply api."""
coresys.resolution.suggestions = clear_snapshot = Suggestion(
SuggestionType.CLEAR_FULL_SNAPSHOT, ContextType.SYSTEM
)
coresys.resolution.suggestions = create_snapshot = Suggestion(
SuggestionType.CREATE_FULL_SNAPSHOT, ContextType.SYSTEM
)
mock_snapshots = AsyncMock()
mock_health = AsyncMock()
coresys.snapshots.do_snapshot_full = mock_snapshots
coresys.resolution.healthcheck = mock_health
await coresys.resolution.apply_suggestion(clear_snapshot)
await coresys.resolution.apply_suggestion(create_snapshot)
assert mock_snapshots.called
assert mock_health.called
assert clear_snapshot not in coresys.resolution.suggestions
assert create_snapshot not in coresys.resolution.suggestions
with pytest.raises(ResolutionError):
await coresys.resolution.apply_suggestion(clear_snapshot)
@pytest.mark.asyncio
async def test_resolution_dismiss_issue(coresys: CoreSys):
"""Test resolution manager issue apply api."""
coresys.resolution.issues = updated_failed = Issue(
IssueType.UPDATE_FAILED, ContextType.SYSTEM
)
assert IssueType.UPDATE_FAILED == coresys.resolution.issues[-1].type
await coresys.resolution.dismiss_issue(updated_failed)
assert updated_failed not in coresys.resolution.issues
@pytest.mark.asyncio
async def test_resolution_create_issue_suggestion(coresys: CoreSys):
"""Test resolution manager issue and suggestion."""
coresys.resolution.create_issue(
IssueType.UPDATE_ROLLBACK,
ContextType.CORE,
"slug",
[SuggestionType.SYSTEM_REPAIR],
)
assert IssueType.UPDATE_ROLLBACK == coresys.resolution.issues[-1].type
assert ContextType.CORE == coresys.resolution.issues[-1].context
assert coresys.resolution.issues[-1].reference == "slug"
assert SuggestionType.SYSTEM_REPAIR == coresys.resolution.suggestions[-1].type
assert ContextType.CORE == coresys.resolution.suggestions[-1].context