mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-07-14 20:56:30 +00:00
De-duplicate Backup/Restore logic (#3311)
* De-duplicate Backup/Restore logic Create internal _do_backup()/_do_restore() method which de-duplicates some of the backup/restore logic previously part of full/partial backup/restore. * Add Backup/Restore test coverage
This commit is contained in:
Stefan Agner
GitHub
parent
cde45e2e7a
commit
eadc629cd9
@ -48,7 +48,7 @@ from ..exceptions import AddonsError
|
||||
from ..utils.json import write_json_file
|
||||
from ..utils.tar import SecureTarFile, atomic_contents_add, secure_path
|
||||
from .utils import key_to_iv, password_for_validating, password_to_key, remove_folder
|
||||
from .validate import ALL_FOLDERS, SCHEMA_BACKUP
|
||||
from .validate import SCHEMA_BACKUP
|
||||
|
||||
_LOGGER: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
@ -311,9 +311,8 @@ class Backup(CoreSysAttributes):
|
||||
finally:
|
||||
self._tmp.cleanup()
|
||||
|
||||
async def store_addons(self, addon_list: Optional[list[Addon]] = None):
|
||||
async def store_addons(self, addon_list: list[str]):
|
||||
"""Add a list of add-ons into backup."""
|
||||
addon_list: list[Addon] = addon_list or self.sys_addons.installed
|
||||
|
||||
async def _addon_save(addon: Addon):
|
||||
"""Task to store an add-on into backup."""
|
||||
@ -346,9 +345,8 @@ class Backup(CoreSysAttributes):
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.warning("Can't save Add-on %s: %s", addon.slug, err)
|
||||
|
||||
async def restore_addons(self, addon_list: Optional[list[str]] = None):
|
||||
async def restore_addons(self, addon_list: list[str]):
|
||||
"""Restore a list add-on from backup."""
|
||||
addon_list: list[str] = addon_list or self.addon_list
|
||||
|
||||
async def _addon_restore(addon_slug: str):
|
||||
"""Task to restore an add-on into backup."""
|
||||
@ -375,9 +373,8 @@ class Backup(CoreSysAttributes):
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.warning("Can't restore Add-on %s: %s", slug, err)
|
||||
|
||||
async def store_folders(self, folder_list: Optional[list[str]] = None):
|
||||
async def store_folders(self, folder_list: list[str]):
|
||||
"""Backup Supervisor data into backup."""
|
||||
folder_list: set[str] = set(folder_list or ALL_FOLDERS)
|
||||
|
||||
def _folder_save(name: str):
|
||||
"""Take backup of a folder."""
|
||||
@ -414,9 +411,8 @@ class Backup(CoreSysAttributes):
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.warning("Can't save folder %s: %s", folder, err)
|
||||
|
||||
async def restore_folders(self, folder_list: Optional[list[str]] = None):
|
||||
async def restore_folders(self, folder_list: list[str]):
|
||||
"""Backup Supervisor data into backup."""
|
||||
folder_list: set[str] = set(folder_list or self.folders)
|
||||
|
||||
def _folder_restore(name: str):
|
||||
"""Intenal function to restore a folder."""
|
||||
|
||||
@ -7,6 +7,9 @@ from typing import Awaitable
|
||||
from awesomeversion.awesomeversion import AwesomeVersion
|
||||
from awesomeversion.exceptions import AwesomeVersionCompareException
|
||||
|
||||
from supervisor.addons.addon import Addon
|
||||
from supervisor.backups.validate import ALL_FOLDERS
|
||||
|
||||
from ..const import FOLDER_HOMEASSISTANT, CoreState
|
||||
from ..coresys import CoreSysAttributes
|
||||
from ..exceptions import AddonsError
|
||||
@ -126,6 +129,38 @@ class BackupManager(CoreSysAttributes):
|
||||
self._backups[backup.slug] = backup
|
||||
return backup
|
||||
|
||||
async def _do_backup(
|
||||
self,
|
||||
backup: Backup,
|
||||
addon_list: list[Addon],
|
||||
folder_list: list[str],
|
||||
):
|
||||
try:
|
||||
self.sys_core.state = CoreState.FREEZE
|
||||
|
||||
async with backup:
|
||||
# Backup add-ons
|
||||
if addon_list:
|
||||
_LOGGER.info("Backing up %s store Add-ons", backup.slug)
|
||||
await backup.store_addons(addon_list)
|
||||
|
||||
# Backup folders
|
||||
if folder_list:
|
||||
_LOGGER.info("Backing up %s store folders", backup.slug)
|
||||
await backup.store_folders(folder_list)
|
||||
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Backup %s error", backup.slug)
|
||||
self.sys_capture_exception(err)
|
||||
return None
|
||||
|
||||
else:
|
||||
self._backups[backup.slug] = backup
|
||||
return backup
|
||||
|
||||
finally:
|
||||
self.sys_core.state = CoreState.RUNNING
|
||||
|
||||
@Job(conditions=[JobCondition.FREE_SPACE, JobCondition.RUNNING])
|
||||
async def do_backup_full(self, name="", password=None):
|
||||
"""Create a full backup."""
|
||||
@ -134,34 +169,16 @@ class BackupManager(CoreSysAttributes):
|
||||
return None
|
||||
|
||||
backup = self._create_backup(name, BackupType.FULL, password)
|
||||
|
||||
_LOGGER.info("Creating new full backup with slug %s", backup.slug)
|
||||
try:
|
||||
self.sys_core.state = CoreState.FREEZE
|
||||
await self.lock.acquire()
|
||||
|
||||
async with backup:
|
||||
# Backup add-ons
|
||||
_LOGGER.info("Backing up %s store Add-ons", backup.slug)
|
||||
await backup.store_addons()
|
||||
|
||||
# Backup folders
|
||||
_LOGGER.info("Backing up %s store folders", backup.slug)
|
||||
await backup.store_folders()
|
||||
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Backup %s error", backup.slug)
|
||||
self.sys_capture_exception(err)
|
||||
return None
|
||||
|
||||
else:
|
||||
async with self.lock:
|
||||
backup = await self._do_backup(
|
||||
backup, self.sys_addons.installed, ALL_FOLDERS
|
||||
)
|
||||
if backup:
|
||||
_LOGGER.info("Creating full backup with slug %s completed", backup.slug)
|
||||
self._backups[backup.slug] = backup
|
||||
return backup
|
||||
|
||||
finally:
|
||||
self.sys_core.state = CoreState.RUNNING
|
||||
self.lock.release()
|
||||
|
||||
@Job(conditions=[JobCondition.FREE_SPACE, JobCondition.RUNNING])
|
||||
async def do_backup_partial(
|
||||
self, name="", addons=None, folders=None, password=None, homeassistant=True
|
||||
@ -176,17 +193,11 @@ class BackupManager(CoreSysAttributes):
|
||||
|
||||
if len(addons) == 0 and len(folders) == 0 and not homeassistant:
|
||||
_LOGGER.error("Nothing to create backup for")
|
||||
return
|
||||
|
||||
backup = self._create_backup(name, BackupType.PARTIAL, password, homeassistant)
|
||||
|
||||
_LOGGER.info("Creating new partial backup with slug %s", backup.slug)
|
||||
try:
|
||||
self.sys_core.state = CoreState.FREEZE
|
||||
await self.lock.acquire()
|
||||
|
||||
async with backup:
|
||||
# Backup add-ons
|
||||
async with self.lock:
|
||||
addon_list = []
|
||||
for addon_slug in addons:
|
||||
addon = self.sys_addons.get(addon_slug)
|
||||
@ -195,28 +206,86 @@ class BackupManager(CoreSysAttributes):
|
||||
continue
|
||||
_LOGGER.warning("Add-on %s not found/installed", addon_slug)
|
||||
|
||||
if addon_list:
|
||||
_LOGGER.info("Backing up %s store Add-ons", backup.slug)
|
||||
await backup.store_addons(addon_list)
|
||||
|
||||
# Backup folders
|
||||
if folders:
|
||||
_LOGGER.info("Backing up %s store folders", backup.slug)
|
||||
await backup.store_folders(folders)
|
||||
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Backup %s error", backup.slug)
|
||||
self.sys_capture_exception(err)
|
||||
return None
|
||||
|
||||
else:
|
||||
_LOGGER.info("Creating partial backup with slug %s completed", backup.slug)
|
||||
self._backups[backup.slug] = backup
|
||||
backup = await self._do_backup(backup, addon_list, folders)
|
||||
if backup:
|
||||
_LOGGER.info(
|
||||
"Creating partial backup with slug %s completed", backup.slug
|
||||
)
|
||||
return backup
|
||||
|
||||
async def _do_restore(
|
||||
self,
|
||||
backup: Backup,
|
||||
addon_list: list[Addon],
|
||||
folder_list: list[str],
|
||||
homeassistant: bool,
|
||||
remove_other_addons: bool,
|
||||
):
|
||||
try:
|
||||
# Stop Home Assistant Core if we restore the version or config directory
|
||||
if FOLDER_HOMEASSISTANT in folder_list or homeassistant:
|
||||
await self.sys_homeassistant.core.stop()
|
||||
|
||||
async with backup:
|
||||
# Restore docker config
|
||||
_LOGGER.info("Restoring %s Docker config", backup.slug)
|
||||
backup.restore_dockerconfig()
|
||||
|
||||
if FOLDER_HOMEASSISTANT in folder_list:
|
||||
backup.restore_homeassistant()
|
||||
|
||||
# Process folders
|
||||
if folder_list:
|
||||
_LOGGER.info("Restoring %s folders", backup.slug)
|
||||
await backup.restore_folders(folder_list)
|
||||
|
||||
# Process Home-Assistant
|
||||
task_hass = None
|
||||
if homeassistant:
|
||||
_LOGGER.info("Restoring %s Home Assistant Core", backup.slug)
|
||||
task_hass = self._update_core_task(backup.homeassistant_version)
|
||||
|
||||
if addon_list:
|
||||
_LOGGER.info("Restoring %s Repositories", backup.slug)
|
||||
await backup.restore_repositories()
|
||||
|
||||
_LOGGER.info("Restoring %s Add-ons", backup.slug)
|
||||
await backup.restore_addons(addon_list)
|
||||
|
||||
# Delete delta add-ons
|
||||
if remove_other_addons:
|
||||
_LOGGER.info("Removing Add-ons not in the backup %s", backup.slug)
|
||||
for addon in self.sys_addons.installed:
|
||||
if addon.slug in backup.addon_list:
|
||||
continue
|
||||
|
||||
# Remove Add-on because it's not a part of the new env
|
||||
# Do it sequential avoid issue on slow IO
|
||||
try:
|
||||
await addon.uninstall()
|
||||
except AddonsError:
|
||||
_LOGGER.warning("Can't uninstall Add-on %s", addon.slug)
|
||||
|
||||
# Wait for Home Assistant Core update/downgrade
|
||||
if task_hass:
|
||||
_LOGGER.info("Restore %s wait for Home-Assistant", backup.slug)
|
||||
await task_hass
|
||||
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Restore %s error", backup.slug)
|
||||
self.sys_capture_exception(err)
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
finally:
|
||||
self.sys_core.state = CoreState.RUNNING
|
||||
self.lock.release()
|
||||
# Do we need start Home Assistant Core?
|
||||
if not await self.sys_homeassistant.core.is_running():
|
||||
await self.sys_homeassistant.core.start()
|
||||
|
||||
# Check If we can access to API / otherwise restart
|
||||
if not await self.sys_homeassistant.api.check_api_state():
|
||||
_LOGGER.warning("Need restart HomeAssistant for API")
|
||||
await self.sys_homeassistant.core.restart()
|
||||
|
||||
@Job(
|
||||
conditions=[
|
||||
@ -227,7 +296,7 @@ class BackupManager(CoreSysAttributes):
|
||||
JobCondition.RUNNING,
|
||||
]
|
||||
)
|
||||
async def do_restore_full(self, backup, password=None):
|
||||
async def do_restore_full(self, backup: Backup, password=None):
|
||||
"""Restore a backup."""
|
||||
if self.lock.locked():
|
||||
_LOGGER.error("A backup/restore process is already running")
|
||||
@ -242,65 +311,20 @@ class BackupManager(CoreSysAttributes):
|
||||
return False
|
||||
|
||||
_LOGGER.info("Full-Restore %s start", backup.slug)
|
||||
try:
|
||||
async with self.lock:
|
||||
self.sys_core.state = CoreState.FREEZE
|
||||
await self.lock.acquire()
|
||||
|
||||
async with backup:
|
||||
# Stop Home-Assistant / Add-ons
|
||||
await self.sys_core.shutdown()
|
||||
|
||||
# Restore folders
|
||||
_LOGGER.info("Restoring %s folders", backup.slug)
|
||||
await backup.restore_folders()
|
||||
success = await self._do_restore(
|
||||
backup, backup.addon_list, backup.folders, True, True
|
||||
)
|
||||
|
||||
# Restore docker config
|
||||
_LOGGER.info("Restoring %s Docker Config", backup.slug)
|
||||
backup.restore_dockerconfig()
|
||||
|
||||
# Start homeassistant restore
|
||||
_LOGGER.info("Restoring %s Home-Assistant", backup.slug)
|
||||
backup.restore_homeassistant()
|
||||
task_hass = self._update_core_task(backup.homeassistant_version)
|
||||
|
||||
# Restore repositories
|
||||
_LOGGER.info("Restoring %s Repositories", backup.slug)
|
||||
await backup.restore_repositories()
|
||||
|
||||
# Delete delta add-ons
|
||||
_LOGGER.info("Removing add-ons not in the backup %s", backup.slug)
|
||||
for addon in self.sys_addons.installed:
|
||||
if addon.slug in backup.addon_list:
|
||||
continue
|
||||
|
||||
# Remove Add-on because it's not a part of the new env
|
||||
# Do it sequential avoid issue on slow IO
|
||||
try:
|
||||
await addon.uninstall()
|
||||
except AddonsError:
|
||||
_LOGGER.warning("Can't uninstall Add-on %s", addon.slug)
|
||||
|
||||
# Restore add-ons
|
||||
_LOGGER.info("Restore %s old add-ons", backup.slug)
|
||||
await backup.restore_addons()
|
||||
|
||||
# finish homeassistant task
|
||||
_LOGGER.info("Restore %s wait until homeassistant ready", backup.slug)
|
||||
await task_hass
|
||||
await self.sys_homeassistant.core.start()
|
||||
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Restore %s error", backup.slug)
|
||||
self.sys_capture_exception(err)
|
||||
return False
|
||||
|
||||
else:
|
||||
_LOGGER.info("Full-Restore %s done", backup.slug)
|
||||
return True
|
||||
|
||||
finally:
|
||||
self.sys_core.state = CoreState.RUNNING
|
||||
self.lock.release()
|
||||
|
||||
if success:
|
||||
_LOGGER.info("Full-Restore %s done", backup.slug)
|
||||
|
||||
@Job(
|
||||
conditions=[
|
||||
@ -323,68 +347,21 @@ class BackupManager(CoreSysAttributes):
|
||||
_LOGGER.error("Invalid password for backup %s", backup.slug)
|
||||
return False
|
||||
|
||||
addons = addons or []
|
||||
folders = folders or []
|
||||
addon_list = addons or []
|
||||
folder_list = folders or []
|
||||
|
||||
_LOGGER.info("Partial-Restore %s start", backup.slug)
|
||||
try:
|
||||
async with self.lock:
|
||||
self.sys_core.state = CoreState.FREEZE
|
||||
await self.lock.acquire()
|
||||
|
||||
async with backup:
|
||||
# Restore docker config
|
||||
_LOGGER.info("Restoring %s Docker Config", backup.slug)
|
||||
backup.restore_dockerconfig()
|
||||
success = await self._do_restore(
|
||||
backup, addon_list, folder_list, homeassistant, False
|
||||
)
|
||||
|
||||
# Stop Home-Assistant for config restore
|
||||
if FOLDER_HOMEASSISTANT in folders:
|
||||
await self.sys_homeassistant.core.stop()
|
||||
backup.restore_homeassistant()
|
||||
|
||||
# Process folders
|
||||
if folders:
|
||||
_LOGGER.info("Restoring %s folders", backup.slug)
|
||||
await backup.restore_folders(folders)
|
||||
|
||||
# Process Home-Assistant
|
||||
task_hass = None
|
||||
if homeassistant:
|
||||
_LOGGER.info("Restoring %s Home-Assistant", backup.slug)
|
||||
task_hass = self._update_core_task(backup.homeassistant_version)
|
||||
|
||||
if addons:
|
||||
_LOGGER.info("Restoring %s Repositories", backup.slug)
|
||||
await backup.restore_repositories()
|
||||
|
||||
_LOGGER.info("Restoring %s old add-ons", backup.slug)
|
||||
await backup.restore_addons(addons)
|
||||
|
||||
# Make sure homeassistant run agen
|
||||
if task_hass:
|
||||
_LOGGER.info("Restore %s wait for Home-Assistant", backup.slug)
|
||||
await task_hass
|
||||
|
||||
# Do we need start HomeAssistant?
|
||||
if not await self.sys_homeassistant.core.is_running():
|
||||
await self.sys_homeassistant.core.start()
|
||||
|
||||
# Check If we can access to API / otherwise restart
|
||||
if not await self.sys_homeassistant.api.check_api_state():
|
||||
_LOGGER.warning("Need restart HomeAssistant for API")
|
||||
await self.sys_homeassistant.core.restart()
|
||||
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Restore %s error", backup.slug)
|
||||
self.sys_capture_exception(err)
|
||||
return False
|
||||
|
||||
else:
|
||||
_LOGGER.info("Partial-Restore %s done", backup.slug)
|
||||
return True
|
||||
|
||||
finally:
|
||||
self.sys_core.state = CoreState.RUNNING
|
||||
self.lock.release()
|
||||
|
||||
if success:
|
||||
_LOGGER.info("Partial-Restore %s done", backup.slug)
|
||||
|
||||
def _update_core_task(self, version: AwesomeVersion) -> Awaitable[None]:
|
||||
"""Process core update if needed and make awaitable object."""
|
||||
|
||||
1
tests/backups/__init__.py
Normal file
1
tests/backups/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Backup tests."""
|
||||
46
tests/backups/conftest.py
Normal file
46
tests/backups/conftest.py
Normal file
@ -0,0 +1,46 @@
|
||||
"""Mock test."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from supervisor.backups.const import BackupType
|
||||
from supervisor.backups.validate import ALL_FOLDERS
|
||||
|
||||
from tests.const import TEST_ADDON_SLUG
|
||||
|
||||
|
||||
@pytest.fixture(name="backup_mock")
|
||||
def fixture_backup_mock():
|
||||
"""Backup class mock."""
|
||||
with patch("supervisor.backups.manager.Backup") as backup_mock:
|
||||
backup_instance = MagicMock()
|
||||
backup_mock.return_value = backup_instance
|
||||
|
||||
backup_instance.store_addons = AsyncMock(return_value=None)
|
||||
backup_instance.store_folders = AsyncMock(return_value=None)
|
||||
backup_instance.restore_addons = AsyncMock(return_value=None)
|
||||
backup_instance.restore_folders = AsyncMock(return_value=None)
|
||||
backup_instance.restore_repositories = AsyncMock(return_value=None)
|
||||
|
||||
yield backup_mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def partial_backup_mock(backup_mock):
|
||||
"""Partial backup mock."""
|
||||
backup_instance = backup_mock.return_value
|
||||
backup_instance.sys_type = BackupType.PARTIAL
|
||||
backup_instance.folders = []
|
||||
backup_instance.addon_list = [TEST_ADDON_SLUG]
|
||||
yield backup_mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def full_backup_mock(backup_mock):
|
||||
"""Full backup mock."""
|
||||
backup_instance = backup_mock.return_value
|
||||
backup_instance.sys_type = BackupType.FULL
|
||||
backup_instance.folders = ALL_FOLDERS
|
||||
backup_instance.addon_list = [TEST_ADDON_SLUG]
|
||||
yield backup_mock
|
||||
205
tests/backups/test_manager.py
Normal file
205
tests/backups/test_manager.py
Normal file
@ -0,0 +1,205 @@
|
||||
"""Test BackupManager class."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from supervisor.backups.const import BackupType
|
||||
from supervisor.backups.manager import BackupManager
|
||||
from supervisor.const import FOLDER_HOMEASSISTANT, CoreState
|
||||
from supervisor.coresys import CoreSys
|
||||
|
||||
from tests.const import TEST_ADDON_SLUG
|
||||
|
||||
|
||||
async def test_do_backup_full(coresys: CoreSys, backup_mock, install_addon_ssh):
|
||||
"""Test creating Backup."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
# backup_mock fixture causes Backup() to be a MagicMock
|
||||
backup_instance: MagicMock = await manager.do_backup_full()
|
||||
|
||||
# Check Backup has been created without password
|
||||
assert backup_instance.new.call_args[0][3] == BackupType.FULL
|
||||
assert backup_instance.new.call_args[0][4] is None
|
||||
|
||||
backup_instance.store_homeassistant.assert_called_once()
|
||||
backup_instance.store_repositories.assert_called_once()
|
||||
backup_instance.store_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.store_addons.assert_called_once()
|
||||
assert install_addon_ssh in backup_instance.store_addons.call_args[0][0]
|
||||
|
||||
backup_instance.store_folders.assert_called_once()
|
||||
assert len(backup_instance.store_folders.call_args[0][0]) == 5
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
|
||||
|
||||
async def test_do_backup_partial_minimal(
|
||||
coresys: CoreSys, backup_mock, install_addon_ssh
|
||||
):
|
||||
"""Test creating minimal partial Backup."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
# backup_mock fixture causes Backup() to be a MagicMock
|
||||
backup_instance: MagicMock = await manager.do_backup_partial(homeassistant=False)
|
||||
|
||||
# Check Backup has been created without password
|
||||
assert backup_instance.new.call_args[0][3] == BackupType.PARTIAL
|
||||
assert backup_instance.new.call_args[0][4] is None
|
||||
|
||||
backup_instance.store_homeassistant.assert_not_called()
|
||||
backup_instance.store_repositories.assert_called_once()
|
||||
backup_instance.store_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.store_addons.assert_not_called()
|
||||
|
||||
backup_instance.store_folders.assert_not_called()
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
|
||||
|
||||
async def test_do_backup_partial_maximal(
|
||||
coresys: CoreSys, backup_mock, install_addon_ssh
|
||||
):
|
||||
"""Test creating maximal partial Backup."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
# backup_mock fixture causes Backup() to be a MagicMock
|
||||
backup_instance: MagicMock = await manager.do_backup_partial(
|
||||
addons=[TEST_ADDON_SLUG], folders=["/test"], homeassistant=True
|
||||
)
|
||||
|
||||
# Check Backup has been created without password
|
||||
assert backup_instance.new.call_args[0][3] == BackupType.PARTIAL
|
||||
assert backup_instance.new.call_args[0][4] is None
|
||||
|
||||
backup_instance.store_homeassistant.assert_called_once()
|
||||
backup_instance.store_repositories.assert_called_once()
|
||||
backup_instance.store_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.store_addons.assert_called_once()
|
||||
assert install_addon_ssh in backup_instance.store_addons.call_args[0][0]
|
||||
|
||||
backup_instance.store_folders.assert_called_once()
|
||||
assert len(backup_instance.store_folders.call_args[0][0]) == 1
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
|
||||
|
||||
async def test_do_restore_full(coresys: CoreSys, full_backup_mock, install_addon_ssh):
|
||||
"""Test restoring full Backup."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
coresys.homeassistant.core.start = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.stop = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.update = AsyncMock(return_value=None)
|
||||
install_addon_ssh.uninstall = AsyncMock(return_value=None)
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
backup_instance = full_backup_mock.return_value
|
||||
await manager.do_restore_full(backup_instance)
|
||||
|
||||
backup_instance.restore_homeassistant.assert_called_once()
|
||||
backup_instance.restore_repositories.assert_called_once()
|
||||
backup_instance.restore_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.restore_addons.assert_called_once()
|
||||
install_addon_ssh.uninstall.assert_not_called()
|
||||
|
||||
backup_instance.restore_folders.assert_called_once()
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
|
||||
|
||||
async def test_do_restore_full_different_addon(
|
||||
coresys: CoreSys, full_backup_mock, install_addon_ssh
|
||||
):
|
||||
"""Test restoring full Backup with different addons than installed."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
coresys.homeassistant.core.start = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.stop = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.update = AsyncMock(return_value=None)
|
||||
install_addon_ssh.uninstall = AsyncMock(return_value=None)
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
backup_instance = full_backup_mock.return_value
|
||||
backup_instance.addon_list = ["differentslug"]
|
||||
await manager.do_restore_full(backup_instance)
|
||||
|
||||
backup_instance.restore_homeassistant.assert_called_once()
|
||||
backup_instance.restore_repositories.assert_called_once()
|
||||
backup_instance.restore_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.restore_addons.assert_called_once()
|
||||
install_addon_ssh.uninstall.assert_called_once()
|
||||
|
||||
backup_instance.restore_folders.assert_called_once()
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
|
||||
|
||||
async def test_do_restore_partial_minimal(
|
||||
coresys: CoreSys, partial_backup_mock, install_addon_ssh
|
||||
):
|
||||
"""Test restoring partial Backup minimal."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
coresys.homeassistant.core.start = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.stop = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.update = AsyncMock(return_value=None)
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
backup_instance = partial_backup_mock.return_value
|
||||
await manager.do_restore_partial(backup_instance, homeassistant=False)
|
||||
|
||||
backup_instance.restore_homeassistant.assert_not_called()
|
||||
backup_instance.restore_repositories.assert_not_called()
|
||||
backup_instance.restore_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.restore_addons.assert_not_called()
|
||||
|
||||
backup_instance.restore_folders.assert_not_called()
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
|
||||
|
||||
async def test_do_restore_partial_maximal(coresys: CoreSys, partial_backup_mock):
|
||||
"""Test restoring partial Backup minimal."""
|
||||
coresys.core.state = CoreState.RUNNING
|
||||
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
|
||||
coresys.homeassistant.core.start = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.stop = AsyncMock(return_value=None)
|
||||
coresys.homeassistant.core.update = AsyncMock(return_value=None)
|
||||
|
||||
manager = BackupManager(coresys)
|
||||
|
||||
backup_instance = partial_backup_mock.return_value
|
||||
await manager.do_restore_partial(
|
||||
backup_instance,
|
||||
addons=[TEST_ADDON_SLUG],
|
||||
folders=[FOLDER_HOMEASSISTANT],
|
||||
homeassistant=True,
|
||||
)
|
||||
|
||||
backup_instance.restore_homeassistant.assert_called_once()
|
||||
backup_instance.restore_repositories.assert_called_once()
|
||||
backup_instance.restore_dockerconfig.assert_called_once()
|
||||
|
||||
backup_instance.restore_addons.assert_called_once()
|
||||
|
||||
backup_instance.restore_folders.assert_called_once()
|
||||
|
||||
assert coresys.core.state == CoreState.RUNNING
|
||||
@ -288,3 +288,4 @@ def install_addon_ssh(coresys: CoreSys, repository):
|
||||
coresys.addons.data.install(store)
|
||||
addon = Addon(coresys, store.slug)
|
||||
coresys.addons.local[addon.slug] = addon
|
||||
yield addon
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user