mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-07-09 18:26:30 +00:00
Create issue+suggestion when no recent backup (#3814)
* Automatic full backup option * Fix test for change in free space check * Suggestions only, no automation * Remove extra in backup config schema
This commit is contained in:
parent
fc646db95f
commit
bf48d48c51
@ -448,11 +448,13 @@ class RestAPI(CoreSysAttributes):
|
||||
self.webapp.add_routes(
|
||||
[
|
||||
web.get("/backups", api_backups.list),
|
||||
web.get("/backups/info", api_backups.info),
|
||||
web.post("/backups/options", api_backups.options),
|
||||
web.post("/backups/reload", api_backups.reload),
|
||||
web.post("/backups/new/full", api_backups.backup_full),
|
||||
web.post("/backups/new/partial", api_backups.backup_partial),
|
||||
web.post("/backups/new/upload", api_backups.upload),
|
||||
web.get("/backups/{slug}/info", api_backups.info),
|
||||
web.get("/backups/{slug}/info", api_backups.backup_info),
|
||||
web.delete("/backups/{slug}", api_backups.remove),
|
||||
web.post("/backups/{slug}/restore/full", api_backups.restore_full),
|
||||
web.post(
|
||||
|
@ -9,13 +9,14 @@ from aiohttp import web
|
||||
from aiohttp.hdrs import CONTENT_DISPOSITION
|
||||
import voluptuous as vol
|
||||
|
||||
from ..backups.validate import ALL_FOLDERS, FOLDER_HOMEASSISTANT
|
||||
from ..backups.validate import ALL_FOLDERS, FOLDER_HOMEASSISTANT, days_until_stale
|
||||
from ..const import (
|
||||
ATTR_ADDONS,
|
||||
ATTR_BACKUPS,
|
||||
ATTR_COMPRESSED,
|
||||
ATTR_CONTENT,
|
||||
ATTR_DATE,
|
||||
ATTR_DAYS_UNTIL_STALE,
|
||||
ATTR_FOLDERS,
|
||||
ATTR_HOMEASSISTANT,
|
||||
ATTR_NAME,
|
||||
@ -24,6 +25,7 @@ from ..const import (
|
||||
ATTR_REPOSITORIES,
|
||||
ATTR_SIZE,
|
||||
ATTR_SLUG,
|
||||
ATTR_SUPERVISOR_VERSION,
|
||||
ATTR_TYPE,
|
||||
ATTR_VERSION,
|
||||
)
|
||||
@ -68,6 +70,12 @@ SCHEMA_BACKUP_PARTIAL = SCHEMA_BACKUP_FULL.extend(
|
||||
}
|
||||
)
|
||||
|
||||
SCHEMA_OPTIONS = vol.Schema(
|
||||
{
|
||||
vol.Optional(ATTR_DAYS_UNTIL_STALE): days_until_stale,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class APIBackups(CoreSysAttributes):
|
||||
"""Handle RESTful API for backups functions."""
|
||||
@ -79,27 +87,30 @@ class APIBackups(CoreSysAttributes):
|
||||
raise APIError("Backup does not exist")
|
||||
return backup
|
||||
|
||||
def _list_backups(self):
|
||||
"""Return list of backups."""
|
||||
return [
|
||||
{
|
||||
ATTR_SLUG: backup.slug,
|
||||
ATTR_NAME: backup.name,
|
||||
ATTR_DATE: backup.date,
|
||||
ATTR_TYPE: backup.sys_type,
|
||||
ATTR_SIZE: backup.size,
|
||||
ATTR_PROTECTED: backup.protected,
|
||||
ATTR_COMPRESSED: backup.compressed,
|
||||
ATTR_CONTENT: {
|
||||
ATTR_HOMEASSISTANT: backup.homeassistant_version is not None,
|
||||
ATTR_ADDONS: backup.addon_list,
|
||||
ATTR_FOLDERS: backup.folders,
|
||||
},
|
||||
}
|
||||
for backup in self.sys_backups.list_backups
|
||||
]
|
||||
|
||||
@api_process
|
||||
async def list(self, request):
|
||||
"""Return backup list."""
|
||||
data_backups = []
|
||||
for backup in self.sys_backups.list_backups:
|
||||
data_backups.append(
|
||||
{
|
||||
ATTR_SLUG: backup.slug,
|
||||
ATTR_NAME: backup.name,
|
||||
ATTR_DATE: backup.date,
|
||||
ATTR_TYPE: backup.sys_type,
|
||||
ATTR_SIZE: backup.size,
|
||||
ATTR_PROTECTED: backup.protected,
|
||||
ATTR_COMPRESSED: backup.compressed,
|
||||
ATTR_CONTENT: {
|
||||
ATTR_HOMEASSISTANT: backup.homeassistant_version is not None,
|
||||
ATTR_ADDONS: backup.addon_list,
|
||||
ATTR_FOLDERS: backup.folders,
|
||||
},
|
||||
}
|
||||
)
|
||||
data_backups = self._list_backups()
|
||||
|
||||
if request.path == "/snapshots":
|
||||
# Kept for backwards compability
|
||||
@ -107,6 +118,24 @@ class APIBackups(CoreSysAttributes):
|
||||
|
||||
return {ATTR_BACKUPS: data_backups}
|
||||
|
||||
@api_process
|
||||
async def info(self, request):
|
||||
"""Return backup list and manager info."""
|
||||
return {
|
||||
ATTR_BACKUPS: self._list_backups(),
|
||||
ATTR_DAYS_UNTIL_STALE: self.sys_backups.days_until_stale,
|
||||
}
|
||||
|
||||
@api_process
|
||||
async def options(self, request):
|
||||
"""Set backup manager options."""
|
||||
body = await api_validate(SCHEMA_OPTIONS, request)
|
||||
|
||||
if ATTR_DAYS_UNTIL_STALE in body:
|
||||
self.sys_backups.days_until_stale = body[ATTR_DAYS_UNTIL_STALE]
|
||||
|
||||
self.sys_backups.save_data()
|
||||
|
||||
@api_process
|
||||
async def reload(self, request):
|
||||
"""Reload backup list."""
|
||||
@ -114,7 +143,7 @@ class APIBackups(CoreSysAttributes):
|
||||
return True
|
||||
|
||||
@api_process
|
||||
async def info(self, request):
|
||||
async def backup_info(self, request):
|
||||
"""Return backup info."""
|
||||
backup = self._extract_slug(request)
|
||||
|
||||
@ -137,6 +166,7 @@ class APIBackups(CoreSysAttributes):
|
||||
ATTR_SIZE: backup.size,
|
||||
ATTR_COMPRESSED: backup.compressed,
|
||||
ATTR_PROTECTED: backup.protected,
|
||||
ATTR_SUPERVISOR_VERSION: backup.supervisor_version,
|
||||
ATTR_HOMEASSISTANT: backup.homeassistant_version,
|
||||
ATTR_ADDONS: data_addons,
|
||||
ATTR_REPOSITORIES: backup.repositories,
|
||||
|
@ -1,5 +1,6 @@
|
||||
"""Representation of a backup file."""
|
||||
from base64 import b64decode, b64encode
|
||||
from datetime import timedelta
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
@ -40,6 +41,7 @@ from ..const import (
|
||||
from ..coresys import CoreSys, CoreSysAttributes
|
||||
from ..exceptions import AddonsError, BackupError
|
||||
from ..utils import remove_folder
|
||||
from ..utils.dt import parse_datetime, utcnow
|
||||
from ..utils.json import write_json_file
|
||||
from .const import BackupType
|
||||
from .utils import key_to_iv, password_to_key
|
||||
@ -164,6 +166,13 @@ class Backup(CoreSysAttributes):
|
||||
"""Return path to backup tarfile."""
|
||||
return self._tarfile
|
||||
|
||||
@property
|
||||
def is_current(self):
|
||||
"""Return true if backup is current, false if stale."""
|
||||
return parse_datetime(self.date) >= utcnow() - timedelta(
|
||||
days=self.sys_backups.days_until_stale
|
||||
)
|
||||
|
||||
def new(self, slug, name, date, sys_type, password=None, compressed=True):
|
||||
"""Initialize a new backup."""
|
||||
# Init metadata
|
||||
|
@ -6,24 +6,31 @@ import logging
|
||||
from pathlib import Path
|
||||
|
||||
from ..addons.addon import Addon
|
||||
from ..const import FOLDER_HOMEASSISTANT, CoreState
|
||||
from ..const import (
|
||||
ATTR_DAYS_UNTIL_STALE,
|
||||
FILE_HASSIO_BACKUPS,
|
||||
FOLDER_HOMEASSISTANT,
|
||||
CoreState,
|
||||
)
|
||||
from ..coresys import CoreSysAttributes
|
||||
from ..exceptions import AddonsError
|
||||
from ..jobs.decorator import Job, JobCondition
|
||||
from ..utils.common import FileConfiguration
|
||||
from ..utils.dt import utcnow
|
||||
from .backup import Backup
|
||||
from .const import BackupType
|
||||
from .utils import create_slug
|
||||
from .validate import ALL_FOLDERS
|
||||
from .validate import ALL_FOLDERS, SCHEMA_BACKUPS_CONFIG
|
||||
|
||||
_LOGGER: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BackupManager(CoreSysAttributes):
|
||||
class BackupManager(FileConfiguration, CoreSysAttributes):
|
||||
"""Manage backups."""
|
||||
|
||||
def __init__(self, coresys):
|
||||
"""Initialize a backup manager."""
|
||||
super().__init__(FILE_HASSIO_BACKUPS, SCHEMA_BACKUPS_CONFIG)
|
||||
self.coresys = coresys
|
||||
self._backups = {}
|
||||
self.lock = asyncio.Lock()
|
||||
@ -33,6 +40,16 @@ class BackupManager(CoreSysAttributes):
|
||||
"""Return a list of all backup objects."""
|
||||
return set(self._backups.values())
|
||||
|
||||
@property
|
||||
def days_until_stale(self) -> int:
|
||||
"""Get days until backup is considered stale."""
|
||||
return self._data[ATTR_DAYS_UNTIL_STALE]
|
||||
|
||||
@days_until_stale.setter
|
||||
def days_until_stale(self, value: int) -> None:
|
||||
"""Set days until backup is considered stale."""
|
||||
self._data[ATTR_DAYS_UNTIL_STALE] = value
|
||||
|
||||
def get(self, slug):
|
||||
"""Return backup object."""
|
||||
return self._backups.get(slug)
|
||||
|
@ -12,6 +12,7 @@ from ..const import (
|
||||
ATTR_COMPRESSED,
|
||||
ATTR_CRYPTO,
|
||||
ATTR_DATE,
|
||||
ATTR_DAYS_UNTIL_STALE,
|
||||
ATTR_DOCKER,
|
||||
ATTR_FOLDERS,
|
||||
ATTR_HOMEASSISTANT,
|
||||
@ -78,6 +79,8 @@ def v1_protected(protected: bool | str) -> bool:
|
||||
|
||||
|
||||
# pylint: disable=no-value-for-parameter
|
||||
days_until_stale = vol.All(vol.Coerce(int), vol.Range(min=1))
|
||||
|
||||
SCHEMA_BACKUP = vol.Schema(
|
||||
{
|
||||
vol.Optional(ATTR_VERSION, default=1): vol.All(vol.Coerce(int), vol.In((1, 2))),
|
||||
@ -127,3 +130,10 @@ SCHEMA_BACKUP = vol.Schema(
|
||||
},
|
||||
extra=vol.ALLOW_EXTRA,
|
||||
)
|
||||
|
||||
SCHEMA_BACKUPS_CONFIG = vol.Schema(
|
||||
{
|
||||
vol.Optional(ATTR_DAYS_UNTIL_STALE, default=30): days_until_stale,
|
||||
},
|
||||
extra=vol.REMOVE_EXTRA,
|
||||
)
|
||||
|
@ -17,6 +17,7 @@ SUPERVISOR_DATA = Path("/data")
|
||||
|
||||
FILE_HASSIO_ADDONS = Path(SUPERVISOR_DATA, "addons.json")
|
||||
FILE_HASSIO_AUTH = Path(SUPERVISOR_DATA, "auth.json")
|
||||
FILE_HASSIO_BACKUPS = Path(SUPERVISOR_DATA, "backups.json")
|
||||
FILE_HASSIO_CONFIG = Path(SUPERVISOR_DATA, "config.json")
|
||||
FILE_HASSIO_DISCOVERY = Path(SUPERVISOR_DATA, "discovery.json")
|
||||
FILE_HASSIO_DOCKER = Path(SUPERVISOR_DATA, "docker.json")
|
||||
@ -131,6 +132,7 @@ ATTR_CPU_PERCENT = "cpu_percent"
|
||||
ATTR_CRYPTO = "crypto"
|
||||
ATTR_DATA = "data"
|
||||
ATTR_DATE = "date"
|
||||
ATTR_DAYS_UNTIL_STALE = "days_until_stale"
|
||||
ATTR_DEBUG = "debug"
|
||||
ATTR_DEBUG_BLOCK = "debug_block"
|
||||
ATTR_DEFAULT = "default"
|
||||
|
49
supervisor/resolution/checks/backups.py
Normal file
49
supervisor/resolution/checks/backups.py
Normal file
@ -0,0 +1,49 @@
|
||||
"""Helpers to check if backed up."""
|
||||
from ...backups.const import BackupType
|
||||
from ...const import CoreState
|
||||
from ...coresys import CoreSys
|
||||
from ..const import ContextType, IssueType, SuggestionType
|
||||
from .base import CheckBase
|
||||
|
||||
|
||||
def setup(coresys: CoreSys) -> CheckBase:
|
||||
"""Check setup function."""
|
||||
return CheckBackups(coresys)
|
||||
|
||||
|
||||
class CheckBackups(CheckBase):
|
||||
"""CheckBackups class for check."""
|
||||
|
||||
async def run_check(self) -> None:
|
||||
"""Run check if not affected by issue."""
|
||||
if self.approve_check():
|
||||
self.sys_resolution.create_issue(
|
||||
IssueType.NO_CURRENT_BACKUP,
|
||||
ContextType.SYSTEM,
|
||||
suggestions=[SuggestionType.CREATE_FULL_BACKUP],
|
||||
)
|
||||
|
||||
async def approve_check(self, reference: str | None = None) -> bool:
|
||||
"""Approve check if it is affected by issue."""
|
||||
return 0 == len(
|
||||
[
|
||||
backup
|
||||
for backup in self.sys_backups.list_backups
|
||||
if backup.sys_type == BackupType.FULL and backup.is_current
|
||||
]
|
||||
)
|
||||
|
||||
@property
|
||||
def issue(self) -> IssueType:
|
||||
"""Return a IssueType enum."""
|
||||
return IssueType.NO_CURRENT_BACKUP
|
||||
|
||||
@property
|
||||
def context(self) -> ContextType:
|
||||
"""Return a ContextType enum."""
|
||||
return ContextType.SYSTEM
|
||||
|
||||
@property
|
||||
def states(self) -> list[CoreState]:
|
||||
"""Return a list of valid states when this check can run."""
|
||||
return [CoreState.RUNNING, CoreState.STARTUP]
|
@ -9,7 +9,6 @@ from ..const import (
|
||||
IssueType,
|
||||
SuggestionType,
|
||||
)
|
||||
from ..data import Suggestion
|
||||
from .base import CheckBase
|
||||
|
||||
|
||||
@ -24,23 +23,15 @@ class CheckFreeSpace(CheckBase):
|
||||
async def run_check(self) -> None:
|
||||
"""Run check if not affected by issue."""
|
||||
if self.sys_host.info.free_space > MINIMUM_FREE_SPACE_THRESHOLD:
|
||||
if len(self.sys_backups.list_backups) == 0:
|
||||
# No backups, let's suggest the user to create one!
|
||||
self.sys_resolution.suggestions = Suggestion(
|
||||
SuggestionType.CREATE_FULL_BACKUP, ContextType.SYSTEM
|
||||
)
|
||||
return
|
||||
|
||||
suggestions: list[SuggestionType] = []
|
||||
if (
|
||||
len(
|
||||
[
|
||||
x
|
||||
for x in self.sys_backups.list_backups
|
||||
if x.sys_type == BackupType.FULL
|
||||
]
|
||||
)
|
||||
>= MINIMUM_FULL_BACKUPS
|
||||
if MINIMUM_FULL_BACKUPS < len(
|
||||
[
|
||||
backup
|
||||
for backup in self.sys_backups.list_backups
|
||||
if backup.sys_type == BackupType.FULL
|
||||
]
|
||||
):
|
||||
suggestions.append(SuggestionType.CLEAR_FULL_BACKUP)
|
||||
|
||||
|
@ -75,6 +75,7 @@ class IssueType(str, Enum):
|
||||
FREE_SPACE = "free_space"
|
||||
IPV4_CONNECTION_PROBLEM = "ipv4_connection_problem"
|
||||
MISSING_IMAGE = "missing_image"
|
||||
NO_CURRENT_BACKUP = "no_current_backup"
|
||||
PWNED = "pwned"
|
||||
SECURITY = "security"
|
||||
TRUST = "trust"
|
||||
|
@ -23,11 +23,13 @@ class FixupSystemClearFullBackup(FixupBase):
|
||||
x for x in self.sys_backups.list_backups if x.sys_type == BackupType.FULL
|
||||
]
|
||||
|
||||
if len(full_backups) < MINIMUM_FULL_BACKUPS:
|
||||
if MINIMUM_FULL_BACKUPS >= len(full_backups):
|
||||
return
|
||||
|
||||
_LOGGER.info("Starting removal of old full backups")
|
||||
for backup in sorted(full_backups, key=lambda x: x.date)[:-1]:
|
||||
for backup in sorted(full_backups, key=lambda x: x.date)[
|
||||
: -1 * MINIMUM_FULL_BACKUPS
|
||||
]:
|
||||
self.sys_backups.remove(backup)
|
||||
|
||||
@property
|
||||
|
45
tests/api/test_backups.py
Normal file
45
tests/api/test_backups.py
Normal file
@ -0,0 +1,45 @@
|
||||
"""Test backups API."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from supervisor.backups.backup import Backup
|
||||
from supervisor.coresys import CoreSys
|
||||
|
||||
|
||||
async def test_info(api_client, coresys: CoreSys, mock_full_backup: Backup):
|
||||
"""Test info endpoint."""
|
||||
resp = await api_client.get("/backups/info")
|
||||
result = await resp.json()
|
||||
assert result["data"]["days_until_stale"] == 30
|
||||
assert len(result["data"]["backups"]) == 1
|
||||
assert result["data"]["backups"][0]["slug"] == "test"
|
||||
assert result["data"]["backups"][0]["content"]["homeassistant"] is True
|
||||
assert len(result["data"]["backups"][0]["content"]["addons"]) == 1
|
||||
assert result["data"]["backups"][0]["content"]["addons"][0] == "local_ssh"
|
||||
|
||||
|
||||
async def test_list(api_client, coresys: CoreSys, mock_full_backup: Backup):
|
||||
"""Test list endpoint."""
|
||||
resp = await api_client.get("/backups")
|
||||
result = await resp.json()
|
||||
assert len(result["data"]["backups"]) == 1
|
||||
assert result["data"]["backups"][0]["slug"] == "test"
|
||||
assert result["data"]["backups"][0]["content"]["homeassistant"] is True
|
||||
assert len(result["data"]["backups"][0]["content"]["addons"]) == 1
|
||||
assert result["data"]["backups"][0]["content"]["addons"][0] == "local_ssh"
|
||||
|
||||
|
||||
async def test_options(api_client, coresys: CoreSys):
|
||||
"""Test options endpoint."""
|
||||
assert coresys.backups.days_until_stale == 30
|
||||
|
||||
with patch.object(type(coresys.backups), "save_data") as save_data:
|
||||
await api_client.post(
|
||||
"/backups/options",
|
||||
json={
|
||||
"days_until_stale": 10,
|
||||
},
|
||||
)
|
||||
save_data.assert_called_once()
|
||||
|
||||
assert coresys.backups.days_until_stale == 10
|
@ -11,13 +11,30 @@ from aiohttp import web
|
||||
from awesomeversion import AwesomeVersion
|
||||
from dbus_next import introspection as intr
|
||||
import pytest
|
||||
from securetar import SecureTarFile
|
||||
|
||||
from supervisor import config as su_config
|
||||
from supervisor.addons.addon import Addon
|
||||
from supervisor.addons.validate import SCHEMA_ADDON_SYSTEM
|
||||
from supervisor.api import RestAPI
|
||||
from supervisor.backups.backup import Backup
|
||||
from supervisor.backups.const import BackupType
|
||||
from supervisor.backups.validate import ALL_FOLDERS
|
||||
from supervisor.bootstrap import initialize_coresys
|
||||
from supervisor.const import ATTR_ADDONS_CUSTOM_LIST, ATTR_REPOSITORIES, REQUEST_FROM
|
||||
from supervisor.const import (
|
||||
ATTR_ADDONS,
|
||||
ATTR_ADDONS_CUSTOM_LIST,
|
||||
ATTR_DATE,
|
||||
ATTR_FOLDERS,
|
||||
ATTR_HOMEASSISTANT,
|
||||
ATTR_NAME,
|
||||
ATTR_REPOSITORIES,
|
||||
ATTR_SIZE,
|
||||
ATTR_SLUG,
|
||||
ATTR_TYPE,
|
||||
ATTR_VERSION,
|
||||
REQUEST_FROM,
|
||||
)
|
||||
from supervisor.coresys import CoreSys
|
||||
from supervisor.dbus.agent import OSAgent
|
||||
from supervisor.dbus.const import DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED
|
||||
@ -32,6 +49,7 @@ from supervisor.docker.monitor import DockerMonitor
|
||||
from supervisor.store.addon import AddonStore
|
||||
from supervisor.store.repository import Repository
|
||||
from supervisor.utils.dbus import DBus
|
||||
from supervisor.utils.dt import utcnow
|
||||
|
||||
from .common import exists_fixture, load_fixture, load_json_fixture
|
||||
from .const import TEST_ADDON_SLUG
|
||||
@ -368,3 +386,50 @@ def install_addon_ssh(coresys: CoreSys, repository):
|
||||
addon = Addon(coresys, store.slug)
|
||||
coresys.addons.local[addon.slug] = addon
|
||||
yield addon
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def mock_full_backup(coresys: CoreSys, tmp_path):
|
||||
"""Mock a full backup."""
|
||||
mock_backup = Backup(coresys, Path(tmp_path, "test_backup"))
|
||||
mock_backup.new("test", "Test", utcnow().isoformat(), BackupType.FULL)
|
||||
mock_backup.repositories = ["https://github.com/awesome-developer/awesome-repo"]
|
||||
mock_backup.docker = {}
|
||||
mock_backup._data[ATTR_ADDONS] = [
|
||||
{
|
||||
ATTR_SLUG: "local_ssh",
|
||||
ATTR_NAME: "SSH",
|
||||
ATTR_VERSION: "1.0.0",
|
||||
ATTR_SIZE: 0,
|
||||
}
|
||||
]
|
||||
mock_backup._data[ATTR_FOLDERS] = ALL_FOLDERS
|
||||
mock_backup._data[ATTR_HOMEASSISTANT] = {
|
||||
ATTR_VERSION: AwesomeVersion("2022.8.0"),
|
||||
ATTR_SIZE: 0,
|
||||
}
|
||||
coresys.backups._backups = {"test": mock_backup}
|
||||
yield mock_backup
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def backups(
|
||||
coresys: CoreSys, tmp_path, request: pytest.FixtureRequest
|
||||
) -> list[Backup]:
|
||||
"""Create and return mock backups."""
|
||||
for i in range(request.param if hasattr(request, "param") else 5):
|
||||
slug = f"sn{i+1}"
|
||||
temp_tar = Path(tmp_path, f"{slug}.tar")
|
||||
with SecureTarFile(temp_tar, "w"):
|
||||
pass
|
||||
backup = Backup(coresys, temp_tar)
|
||||
backup._data = { # pylint: disable=protected-access
|
||||
ATTR_SLUG: slug,
|
||||
ATTR_DATE: utcnow().isoformat(),
|
||||
ATTR_TYPE: BackupType.PARTIAL
|
||||
if "1" == slug[-1] or "5" == slug[-1]
|
||||
else BackupType.FULL,
|
||||
}
|
||||
coresys.backups._backups[backup.slug] = backup
|
||||
|
||||
yield coresys.backups.list_backups
|
||||
|
@ -57,7 +57,7 @@ async def test_if_check_make_issue(coresys: CoreSys):
|
||||
assert coresys.resolution.issues[-1].type == IssueType.FREE_SPACE
|
||||
|
||||
|
||||
async def test_if_check_cleanup_issue(coresys: CoreSys):
|
||||
async def test_if_check_cleanup_issue(coresys: CoreSys, mock_full_backup):
|
||||
"""Test check for setup."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.security.content_trust = False
|
||||
|
@ -1,11 +1,32 @@
|
||||
"""Test evaluation base."""
|
||||
"""Test check free space fixup."""
|
||||
# pylint: disable=import-error,protected-access
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, PropertyMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from supervisor.backups.const import BackupType
|
||||
from supervisor.const import CoreState
|
||||
from supervisor.coresys import CoreSys
|
||||
from supervisor.resolution.checks.free_space import CheckFreeSpace
|
||||
from supervisor.resolution.const import IssueType
|
||||
from supervisor.resolution.const import IssueType, SuggestionType
|
||||
|
||||
|
||||
@pytest.fixture(name="suggestion")
|
||||
async def fixture_suggestion(
|
||||
coresys: CoreSys, request: pytest.FixtureRequest
|
||||
) -> SuggestionType | None:
|
||||
"""Set up test for suggestion."""
|
||||
if request.param == SuggestionType.CLEAR_FULL_BACKUP:
|
||||
backup = MagicMock()
|
||||
backup.sys_type = BackupType.FULL
|
||||
with patch.object(
|
||||
type(coresys.backups),
|
||||
"list_backups",
|
||||
new=PropertyMock(return_value=[backup, backup, backup]),
|
||||
):
|
||||
yield SuggestionType.CLEAR_FULL_BACKUP
|
||||
else:
|
||||
yield request.param
|
||||
|
||||
|
||||
async def test_base(coresys: CoreSys):
|
||||
@ -15,7 +36,12 @@ async def test_base(coresys: CoreSys):
|
||||
assert free_space.enabled
|
||||
|
||||
|
||||
async def test_check(coresys: CoreSys):
|
||||
@pytest.mark.parametrize(
|
||||
"suggestion",
|
||||
[None, SuggestionType.CLEAR_FULL_BACKUP],
|
||||
indirect=True,
|
||||
)
|
||||
async def test_check(coresys: CoreSys, suggestion: SuggestionType | None):
|
||||
"""Test check."""
|
||||
free_space = CheckFreeSpace(coresys)
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
@ -32,6 +58,11 @@ async def test_check(coresys: CoreSys):
|
||||
|
||||
assert coresys.resolution.issues[-1].type == IssueType.FREE_SPACE
|
||||
|
||||
if suggestion:
|
||||
assert coresys.resolution.suggestions[-1].type == suggestion
|
||||
else:
|
||||
assert len(coresys.resolution.suggestions) == 0
|
||||
|
||||
|
||||
async def test_approve(coresys: CoreSys):
|
||||
"""Test check."""
|
||||
|
@ -1,22 +1,16 @@
|
||||
"""Test evaluation base."""
|
||||
"""Test clear full backup fixup."""
|
||||
# pylint: disable=import-error,protected-access
|
||||
from pathlib import Path
|
||||
|
||||
from securetar import SecureTarFile
|
||||
|
||||
from supervisor.backups.backup import Backup
|
||||
from supervisor.backups.const import BackupType
|
||||
from supervisor.const import ATTR_DATE, ATTR_SLUG, ATTR_TYPE
|
||||
from supervisor.coresys import CoreSys
|
||||
from supervisor.resolution.const import ContextType, SuggestionType
|
||||
from supervisor.resolution.data import Suggestion
|
||||
from supervisor.resolution.fixups.system_clear_full_backup import (
|
||||
FixupSystemClearFullBackup,
|
||||
)
|
||||
from supervisor.utils.dt import utcnow
|
||||
|
||||
|
||||
async def test_fixup(coresys: CoreSys, tmp_path):
|
||||
async def test_fixup(coresys: CoreSys, backups: list[Backup]):
|
||||
"""Test fixup."""
|
||||
clear_full_backup = FixupSystemClearFullBackup(coresys)
|
||||
|
||||
@ -26,20 +20,6 @@ async def test_fixup(coresys: CoreSys, tmp_path):
|
||||
SuggestionType.CLEAR_FULL_BACKUP, ContextType.SYSTEM
|
||||
)
|
||||
|
||||
for slug in ["sn1", "sn2", "sn3", "sn4", "sn5"]:
|
||||
temp_tar = Path(tmp_path, f"{slug}.tar")
|
||||
with SecureTarFile(temp_tar, "w"):
|
||||
pass
|
||||
backup = Backup(coresys, temp_tar)
|
||||
backup._data = { # pylint: disable=protected-access
|
||||
ATTR_SLUG: slug,
|
||||
ATTR_DATE: utcnow().isoformat(),
|
||||
ATTR_TYPE: BackupType.PARTIAL
|
||||
if "1" in slug or "5" in slug
|
||||
else BackupType.FULL,
|
||||
}
|
||||
coresys.backups._backups[backup.slug] = backup
|
||||
|
||||
newest_full_backup = coresys.backups._backups["sn4"]
|
||||
|
||||
assert newest_full_backup in coresys.backups.list_backups
|
||||
@ -52,7 +32,7 @@ async def test_fixup(coresys: CoreSys, tmp_path):
|
||||
assert newest_full_backup in coresys.backups.list_backups
|
||||
assert (
|
||||
len([x for x in coresys.backups.list_backups if x.sys_type == BackupType.FULL])
|
||||
== 1
|
||||
== 2
|
||||
)
|
||||
|
||||
assert len(coresys.resolution.suggestions) == 0
|
||||
|
@ -1,4 +1,4 @@
|
||||
"""Test evaluation base."""
|
||||
"""Test create full backup fixup."""
|
||||
# pylint: disable=import-error,protected-access
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user