home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-08-11 16:45:19 +02:00
Code Issues Packages Projects Releases Wiki Activity

Improve backup websocket API tests (#130912)

Browse Source 
* Improve backup websocket API tests

* Add missing snapshot

* Fix tests leaving files behind
This commit is contained in:
Erik Montnemery
2024-11-19 00:30:43 +01:00
committed by GitHub
parent 78053b487c
commit 7b62ed2321
7 changed files with 943 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/backup/manager.py
View File

@@ -628,8 +628,6 @@ class BackupManager(BaseBackupManager[Backup]):
)
await agent.async_download_backup(backup_id, path=path)
path = local_agent.get_backup_path(backup_id)
def _write_restore_file() -> None:
"""Write the restore file."""
Path(self.hass.config.path(RESTORE_BACKUP_FILE)).write_text(

89
tests/components/backup/common.py
View File

@@ -4,11 +4,12 @@ from __future__ import annotations
from pathlib import Path
from typing import Any
from unittest.mock import patch
from unittest.mock import AsyncMock, Mock, patch
from homeassistant.components.backup import (
DOMAIN,
BackupAgent,
BackupAgentPlatformProtocol,
BackupUploadMetadata,
BaseBackup,
)
@@ -19,25 +20,19 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
from tests.common import MockPlatform, mock_platform
LOCAL_AGENT_ID = f"{DOMAIN}.local"
TEST_BASE_BACKUP = BaseBackup(
TEST_BASE_BACKUP_ABC123 = BaseBackup(
backup_id="abc123",
date="1970-01-01T00:00:00.000Z",
name="Test",
protected=False,
size=0.0,
)
TEST_BACKUP = Backup(
agent_ids=["backup.local"],
backup_id="abc123",
date="1970-01-01T00:00:00.000Z",
name="Test",
protected=False,
size=0.0,
)
TEST_BACKUP_PATH = Path("abc123.tar")
TEST_LOCAL_BACKUP = LocalBackup(
TEST_BACKUP_PATH_ABC123 = Path("abc123.tar")
TEST_LOCAL_BACKUP_ABC123 = LocalBackup(
date="1970-01-01T00:00:00.000Z",
backup_id="abc123",
name="Test",
@@ -46,18 +41,40 @@ TEST_LOCAL_BACKUP = LocalBackup(
size=0.0,
)
TEST_BASE_BACKUP_DEF456 = BaseBackup(
backup_id="def456",
date="1980-01-01T00:00:00.000Z",
name="Test 2",
protected=False,
size=1.0,
)
TEST_DOMAIN = "test"
class BackupAgentTest(BackupAgent):
"""Test backup agent."""
def __init__(self, name: str) -> None:
def __init__(self, name: str, backups: list[BaseBackup] | None = None) -> None:
"""Initialize the backup agent."""
self.name = name
if backups is None:
backups = [
BaseBackup(
backup_id="abc123",
date="1970-01-01T00:00:00Z",
name="Test",
protected=False,
size=13.37,
)
]
self._backups = {backup.backup_id: backup for backup in backups}
async def async_download_backup(
self,
backup_id: str,
*,
id: str,
path: Path,
**kwargs: Any,
) -> None:
@@ -74,48 +91,46 @@ class BackupAgentTest(BackupAgent):
async def async_list_backups(self, **kwargs: Any) -> list[BaseBackup]:
"""List backups."""
return [
BaseBackup(
backup_id="abc123",
date="1970-01-01T00:00:00Z",
name="Test",
protected=False,
size=13.37,
)
]
return list(self._backups.values())
async def async_get_backup(
self,
*,
backup_id: str,
**kwargs: Any,
) -> BaseBackup | None:
"""Return a backup."""
if backup_id != "abc123":
return None
return BaseBackup(
backup_id="abc123",
date="1970-01-01T00:00:00Z",
name="Test",
protected=False,
size=13.37,
)
return self._backups.get(backup_id)
async def setup_backup_integration(
hass: HomeAssistant,
with_hassio: bool = False,
configuration: ConfigType | None = None,
backups: list[Backup] | None = None,
*,
backups: dict[str, list[Backup]] | None = None,
remote_agents: list[str] | None = None,
) -> bool:
"""Set up the Backup integration."""
with patch("homeassistant.components.backup.is_hassio", return_value=with_hassio):
remote_agents = remote_agents or []
platform = Mock(
async_get_backup_agents=AsyncMock(
return_value=[BackupAgentTest(agent, []) for agent in remote_agents]
),
spec_set=BackupAgentPlatformProtocol,
)
mock_platform(hass, f"{TEST_DOMAIN}.backup", platform or MockPlatform())
assert await async_setup_component(hass, TEST_DOMAIN, {})
result = await async_setup_component(hass, DOMAIN, configuration or {})
if with_hassio or not backups:
return result
local_agent = hass.data[DATA_MANAGER].backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {backup.backup_id: backup for backup in backups}
local_agent._loaded_backups = True
for agent_id, agent_backups in backups.items():
agent = hass.data[DATA_MANAGER].backup_agents[agent_id]
agent._backups = {backups.backup_id: backups for backups in agent_backups}
if agent_id == LOCAL_AGENT_ID:
agent._loaded_backups = True
return result

15
tests/components/backup/snapshots/test_manager.ambr Normal file
View File

@@ -0,0 +1,15 @@
# serializer version: 1
# name: test_load_backups
dict({
'abc123': dict({
'agent_ids': list([
'backup.local',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
})
# ---

713
tests/components/backup/snapshots/test_websocket.ambr
View File

@@ -333,7 +333,7 @@
'type': 'result',
})
# ---
# name: test_details[with_hassio-with_backup_content]
# name: test_details[with_hassio-remote_agents0-backups0]
dict({
'error': dict({
'code': 'unknown_command',
@@ -344,7 +344,7 @@
'type': 'result',
})
# ---
# name: test_details[with_hassio-without_backup_content]
# name: test_details[with_hassio-remote_agents1-backups1]
dict({
'error': dict({
'code': 'unknown_command',
@@ -355,7 +355,52 @@
'type': 'result',
})
# ---
# name: test_details[without_hassio-with_backup_content]
# name: test_details[with_hassio-remote_agents2-backups2]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_details[with_hassio-remote_agents3-backups3]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_details[with_hassio-remote_agents4-backups4]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_details[without_hassio-remote_agents0-backups0]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backup': None,
}),
'success': True,
'type': 'result',
})
# ---
# name: test_details[without_hassio-remote_agents1-backups1]
dict({
'id': 1,
'result': dict({
@@ -376,7 +421,28 @@
'type': 'result',
})
# ---
# name: test_details[without_hassio-without_backup_content]
# name: test_details[without_hassio-remote_agents2-backups2]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backup': dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_details[without_hassio-remote_agents3-backups3]
dict({
'id': 1,
'result': dict({
@@ -388,6 +454,27 @@
'type': 'result',
})
# ---
# name: test_details[without_hassio-remote_agents4-backups4]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backup': dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_details_with_errors[BackupAgentUnreachableError]
dict({
'id': 1,
@@ -537,7 +624,7 @@
'type': 'result',
})
# ---
# name: test_info[with_hassio]
# name: test_info[with_hassio-remote_agents0-remote_backups0]
dict({
'error': dict({
'code': 'unknown_command',
@@ -548,7 +635,40 @@
'type': 'result',
})
# ---
# name: test_info[without_hassio]
# name: test_info[with_hassio-remote_agents1-remote_backups1]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_info[with_hassio-remote_agents2-remote_backups2]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_info[with_hassio-remote_agents3-remote_backups3]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_info[without_hassio-remote_agents0-remote_backups0]
dict({
'id': 1,
'result': dict({
@@ -572,6 +692,89 @@
'type': 'result',
})
# ---
# name: test_info[without_hassio-remote_agents1-remote_backups1]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'backup.local',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_info[without_hassio-remote_agents2-remote_backups2]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
'backup.local',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_info[without_hassio-remote_agents3-remote_backups3]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'def456',
'date': '1980-01-01T00:00:00.000Z',
'name': 'Test 2',
'protected': False,
'size': 1.0,
}),
dict({
'agent_ids': list([
'backup.local',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_info_with_errors[BackupAgentUnreachableError]
dict({
'id': 1,
@@ -608,7 +811,7 @@
'type': 'result',
})
# ---
# name: test_remove[with_hassio]
# name: test_remove[with_hassio-remote_agents0-backups0]
dict({
'error': dict({
'code': 'unknown_command',
@@ -619,15 +822,29 @@
'type': 'result',
})
# ---
# name: test_remove[without_hassio]
# name: test_remove[with_hassio-remote_agents0-backups0].1
dict({
'id': 1,
'result': None,
'success': True,
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 2,
'success': False,
'type': 'result',
})
# ---
# name: test_restore[with_hassio]
# name: test_remove[with_hassio-remote_agents0-backups0].2
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 3,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents1-backups1]
dict({
'error': dict({
'code': 'unknown_command',
@@ -638,7 +855,421 @@
'type': 'result',
})
# ---
# name: test_restore[without_hassio]
# name: test_remove[with_hassio-remote_agents1-backups1].1
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 2,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents1-backups1].2
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 3,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents2-backups2]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents2-backups2].1
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 2,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents2-backups2].2
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 3,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents3-backups3]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents3-backups3].1
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 2,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents3-backups3].2
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 3,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents4-backups4]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents4-backups4].1
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 2,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[with_hassio-remote_agents4-backups4].2
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 3,
'success': False,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents0-backups0]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents0-backups0].1
dict({
'id': 2,
'result': None,
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents0-backups0].2
dict({
'id': 3,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents1-backups1]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'backup.local',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents1-backups1].1
dict({
'id': 2,
'result': None,
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents1-backups1].2
dict({
'id': 3,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents2-backups2]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents2-backups2].1
dict({
'id': 2,
'result': None,
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents2-backups2].2
dict({
'id': 3,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents3-backups3]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'def456',
'date': '1980-01-01T00:00:00.000Z',
'name': 'Test 2',
'protected': False,
'size': 1.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents3-backups3].1
dict({
'id': 2,
'result': None,
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents3-backups3].2
dict({
'id': 3,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'def456',
'date': '1980-01-01T00:00:00.000Z',
'name': 'Test 2',
'protected': False,
'size': 1.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents4-backups4]
dict({
'id': 1,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
'backup.local',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents4-backups4].1
dict({
'id': 2,
'result': None,
'success': True,
'type': 'result',
})
# ---
# name: test_remove[without_hassio-remote_agents4-backups4].2
dict({
'id': 3,
'result': dict({
'agent_errors': dict({
}),
'backing_up': False,
'backups': list([
dict({
'agent_ids': list([
'test.remote',
]),
'backup_id': 'abc123',
'date': '1970-01-01T00:00:00.000Z',
'name': 'Test',
'protected': False,
'size': 0.0,
}),
]),
}),
'success': True,
'type': 'result',
})
# ---
# name: test_restore_local_agent[with_hassio-backups0]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_restore_local_agent[with_hassio-backups0].1
0
# ---
# name: test_restore_local_agent[with_hassio-backups1]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_restore_local_agent[with_hassio-backups1].1
0
# ---
# name: test_restore_local_agent[without_hassio-backups0]
dict({
'error': dict({
'code': 'home_assistant_error',
'message': 'Backup abc123 not found in agent backup.local',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_restore_local_agent[without_hassio-backups0].1
0
# ---
# name: test_restore_local_agent[without_hassio-backups1]
dict({
'id': 1,
'result': None,
@@ -646,3 +1277,59 @@
'type': 'result',
})
# ---
# name: test_restore_local_agent[without_hassio-backups1].1
1
# ---
# name: test_restore_remote_agent[with_hassio-remote_agents0-backups0]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_restore_remote_agent[with_hassio-remote_agents0-backups0].1
0
# ---
# name: test_restore_remote_agent[with_hassio-remote_agents1-backups1]
dict({
'error': dict({
'code': 'unknown_command',
'message': 'Unknown command.',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_restore_remote_agent[with_hassio-remote_agents1-backups1].1
0
# ---
# name: test_restore_remote_agent[without_hassio-remote_agents0-backups0]
dict({
'error': dict({
'code': 'home_assistant_error',
'message': 'Backup abc123 not found in agent test.remote',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---
# name: test_restore_remote_agent[without_hassio-remote_agents0-backups0].1
0
# ---
# name: test_restore_remote_agent[without_hassio-remote_agents1-backups1]
dict({
'id': 1,
'result': None,
'success': True,
'type': 'result',
})
# ---
# name: test_restore_remote_agent[without_hassio-remote_agents1-backups1].1
1
# ---

4
tests/components/backup/test_http.py
View File

@@ -9,7 +9,7 @@ import pytest
from homeassistant.core import HomeAssistant
from .common import TEST_LOCAL_BACKUP, setup_backup_integration
from .common import TEST_LOCAL_BACKUP_ABC123, setup_backup_integration
from tests.common import MockUser
from tests.typing import ClientSessionGenerator
@@ -27,7 +27,7 @@ async def test_downloading_backup(
with (
patch(
"homeassistant.components.backup.backup.CoreLocalBackupAgent.async_get_backup",
return_value=TEST_LOCAL_BACKUP,
return_value=TEST_LOCAL_BACKUP_ABC123,
),
patch("pathlib.Path.exists", return_value=True),
patch(

63
tests/components/backup/test_manager.py
View File

@@ -8,6 +8,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, mock_open, patc
import aiohttp
from multidict import CIMultiDict, CIMultiDictProxy
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.backup import (
DOMAIN,
@@ -23,10 +24,9 @@ from homeassistant.setup import async_setup_component
from .common import (
LOCAL_AGENT_ID,
TEST_BACKUP,
TEST_BACKUP_PATH,
TEST_BASE_BACKUP,
TEST_LOCAL_BACKUP,
TEST_BACKUP_PATH_ABC123,
TEST_BASE_BACKUP_ABC123,
TEST_LOCAL_BACKUP_ABC123,
BackupAgentTest,
)
@@ -120,7 +120,7 @@ async def test_constructor(hass: HomeAssistant) -> None:
assert manager.temp_backup_dir.as_posix() == hass.config.path("tmp_backups")
async def test_load_backups(hass: HomeAssistant) -> None:
async def test_load_backups(hass: HomeAssistant, snapshot: SnapshotAssertion) -> None:
"""Test loading backups."""
manager = BackupManager(hass)
@@ -128,24 +128,24 @@ async def test_load_backups(hass: HomeAssistant) -> None:
await manager.load_platforms()
with (
patch("pathlib.Path.glob", return_value=[TEST_BACKUP_PATH]),
patch("pathlib.Path.glob", return_value=[TEST_BACKUP_PATH_ABC123]),
patch("tarfile.open", return_value=MagicMock()),
patch(
"homeassistant.components.backup.util.json_loads_object",
return_value={
"date": TEST_LOCAL_BACKUP.date,
"name": TEST_LOCAL_BACKUP.name,
"slug": TEST_LOCAL_BACKUP.backup_id,
"date": TEST_LOCAL_BACKUP_ABC123.date,
"name": TEST_LOCAL_BACKUP_ABC123.name,
"slug": TEST_LOCAL_BACKUP_ABC123.backup_id,
},
),
patch(
"pathlib.Path.stat",
return_value=MagicMock(st_size=TEST_LOCAL_BACKUP.size),
return_value=MagicMock(st_size=TEST_LOCAL_BACKUP_ABC123.size),
),
):
await manager.backup_agents[LOCAL_AGENT_ID].load_backups()
backups, agent_errors = await manager.async_get_backups()
assert backups == {TEST_BACKUP.backup_id: TEST_BACKUP}
assert backups == snapshot
assert agent_errors == {}
@@ -160,12 +160,15 @@ async def test_load_backups_with_exception(
await manager.load_platforms()
with (
patch("pathlib.Path.glob", return_value=[TEST_BACKUP_PATH]),
patch("pathlib.Path.glob", return_value=[TEST_BACKUP_PATH_ABC123]),
patch("tarfile.open", side_effect=OSError("Test exception")),
):
await manager.backup_agents[LOCAL_AGENT_ID].load_backups()
backups, agent_errors = await manager.async_get_backups()
assert f"Unable to read backup {TEST_BACKUP_PATH}: Test exception" in caplog.text
assert (
f"Unable to read backup {TEST_BACKUP_PATH_ABC123}: Test exception"
in caplog.text
)
assert backups == {}
assert agent_errors == {}
@@ -181,11 +184,13 @@ async def test_removing_backup(
await manager.load_platforms()
local_agent = manager.backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {TEST_LOCAL_BACKUP.backup_id: TEST_LOCAL_BACKUP}
local_agent._backups = {
TEST_LOCAL_BACKUP_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._loaded_backups = True
with patch("pathlib.Path.exists", return_value=True):
await manager.async_remove_backup(TEST_LOCAL_BACKUP.backup_id)
await manager.async_remove_backup(TEST_LOCAL_BACKUP_ABC123.backup_id)
assert "Removed backup located at" in caplog.text
@@ -214,19 +219,21 @@ async def test_getting_backup_that_does_not_exist(
await manager.load_platforms()
local_agent = manager.backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {TEST_LOCAL_BACKUP.backup_id: TEST_LOCAL_BACKUP}
local_agent._backups = {
TEST_LOCAL_BACKUP_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._loaded_backups = True
with patch("pathlib.Path.exists", return_value=False):
backup, agent_errors = await manager.async_get_backup(
TEST_LOCAL_BACKUP.backup_id
TEST_LOCAL_BACKUP_ABC123.backup_id
)
assert backup is None
assert agent_errors == {}
assert (
f"Removing tracked backup ({TEST_LOCAL_BACKUP.backup_id}) that "
f"does not exists on the expected path {TEST_LOCAL_BACKUP.path}"
f"Removing tracked backup ({TEST_LOCAL_BACKUP_ABC123.backup_id}) that "
f"does not exists on the expected path {TEST_LOCAL_BACKUP_ABC123.path}"
) in caplog.text
@@ -424,7 +431,7 @@ async def test_async_receive_backup(
patch("shutil.copy") as copy_mock,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=TEST_BASE_BACKUP,
return_value=TEST_BASE_BACKUP_ABC123,
),
):
await manager.async_receive_backup(
@@ -457,7 +464,9 @@ async def test_async_trigger_restore(
await manager.load_platforms()
local_agent = manager.backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {TEST_LOCAL_BACKUP.backup_id: TEST_LOCAL_BACKUP}
local_agent._backups = {
TEST_LOCAL_BACKUP_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._loaded_backups = True
with (
@@ -466,7 +475,7 @@ async def test_async_trigger_restore(
patch("homeassistant.core.ServiceRegistry.async_call") as mocked_service_call,
):
await manager.async_restore_backup(
TEST_LOCAL_BACKUP.backup_id, agent_id=LOCAL_AGENT_ID, password=None
TEST_LOCAL_BACKUP_ABC123.backup_id, agent_id=LOCAL_AGENT_ID, password=None
)
assert (
mocked_write_text.call_args[0][0]
@@ -486,7 +495,9 @@ async def test_async_trigger_restore_with_password(
await manager.load_platforms()
local_agent = manager.backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {TEST_LOCAL_BACKUP.backup_id: TEST_LOCAL_BACKUP}
local_agent._backups = {
TEST_LOCAL_BACKUP_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._loaded_backups = True
with (
@@ -495,7 +506,9 @@ async def test_async_trigger_restore_with_password(
patch("homeassistant.core.ServiceRegistry.async_call") as mocked_service_call,
):
await manager.async_restore_backup(
TEST_LOCAL_BACKUP.backup_id, agent_id=LOCAL_AGENT_ID, password="abc123"
TEST_LOCAL_BACKUP_ABC123.backup_id,
agent_id=LOCAL_AGENT_ID,
password="abc123",
)
assert (
mocked_write_text.call_args[0][0]
@@ -516,5 +529,5 @@ async def test_async_trigger_restore_missing_backup(hass: HomeAssistant) -> None
with pytest.raises(HomeAssistantError, match="Backup abc123 not found"):
await manager.async_restore_backup(
TEST_LOCAL_BACKUP.backup_id, agent_id=LOCAL_AGENT_ID, password=None
TEST_LOCAL_BACKUP_ABC123.backup_id, agent_id=LOCAL_AGENT_ID, password=None
)

155
tests/components/backup/test_websocket.py
View File

@@ -15,8 +15,16 @@ from homeassistant.components.backup.manager import NewBackup
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .common import TEST_LOCAL_BACKUP, BackupAgentTest, setup_backup_integration
from .common import (
LOCAL_AGENT_ID,
TEST_BASE_BACKUP_ABC123,
TEST_BASE_BACKUP_DEF456,
TEST_LOCAL_BACKUP_ABC123,
BackupAgentTest,
setup_backup_integration,
)
from tests.common import async_mock_service
from tests.typing import WebSocketGenerator
@@ -32,6 +40,15 @@ def sync_access_token_proxy(
return request.getfixturevalue(access_token_fixture_name)
@pytest.mark.parametrize(
("remote_agents", "remote_backups"),
[
([], {}),
(["remote"], {}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_DEF456]}),
],
)
@pytest.mark.parametrize(
"with_hassio",
[
@@ -42,12 +59,17 @@ def sync_access_token_proxy(
async def test_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
remote_agents: list[str],
remote_backups: dict[str, list[BaseBackup]],
snapshot: SnapshotAssertion,
) -> None:
"""Test getting backup info."""
await setup_backup_integration(
hass, with_hassio=with_hassio, backups=[TEST_LOCAL_BACKUP]
hass,
with_hassio=with_hassio,
backups={LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]} | remote_backups,
remote_agents=remote_agents,
)
client = await hass_ws_client(hass)
@@ -67,7 +89,9 @@ async def test_info_with_errors(
snapshot: SnapshotAssertion,
) -> None:
"""Test getting backup info with one unavailable agent."""
await setup_backup_integration(hass, with_hassio=False, backups=[TEST_LOCAL_BACKUP])
await setup_backup_integration(
hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}
)
hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test")
client = await hass_ws_client(hass)
@@ -79,10 +103,19 @@ async def test_info_with_errors(
@pytest.mark.parametrize(
"backup_content",
("remote_agents", "backups"),
[
pytest.param([TEST_LOCAL_BACKUP], id="with_backup_content"),
pytest.param(None, id="without_backup_content"),
([], {}),
(["remote"], {LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_DEF456]}),
(
["remote"],
{
LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123],
"test.remote": [TEST_BASE_BACKUP_ABC123],
},
),
],
)
@pytest.mark.parametrize(
@@ -95,13 +128,14 @@ async def test_info_with_errors(
async def test_details(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
backup_content: BaseBackup | None,
remote_agents: list[str],
backups: dict[str, BaseBackup],
snapshot: SnapshotAssertion,
) -> None:
"""Test getting backup info."""
await setup_backup_integration(
hass, with_hassio=with_hassio, backups=backup_content
hass, with_hassio=with_hassio, backups=backups, remote_agents=remote_agents
)
client = await hass_ws_client(hass)
@@ -124,7 +158,9 @@ async def test_details_with_errors(
snapshot: SnapshotAssertion,
) -> None:
"""Test getting backup info with one unavailable agent."""
await setup_backup_integration(hass, with_hassio=False, backups=[TEST_LOCAL_BACKUP])
await setup_backup_integration(
hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}
)
hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test")
client = await hass_ws_client(hass)
@@ -140,6 +176,22 @@ async def test_details_with_errors(
assert await client.receive_json() == snapshot
@pytest.mark.parametrize(
("remote_agents", "backups"),
[
([], {}),
(["remote"], {LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_DEF456]}),
(
["remote"],
{
LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123],
"test.remote": [TEST_BASE_BACKUP_ABC123],
},
),
],
)
@pytest.mark.parametrize(
"with_hassio",
[
@@ -150,21 +202,28 @@ async def test_details_with_errors(
async def test_remove(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
remote_agents: list[str],
backups: dict[str, BaseBackup],
snapshot: SnapshotAssertion,
) -> None:
"""Test removing a backup file."""
await setup_backup_integration(hass, with_hassio=with_hassio)
await setup_backup_integration(
hass, with_hassio=with_hassio, backups=backups, remote_agents=remote_agents
)
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_remove_backup",
):
await client.send_json_auto_id({"type": "backup/info"})
assert await client.receive_json() == snapshot
await client.send_json_auto_id({"type": "backup/remove", "backup_id": "abc123"})
assert await client.receive_json() == snapshot
await client.send_json_auto_id({"type": "backup/info"})
assert await client.receive_json() == snapshot
@pytest.mark.parametrize(
"data",
@@ -262,6 +321,13 @@ async def test_generate_without_hassio(
)
@pytest.mark.parametrize(
"backups",
[
{},
{LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]},
],
)
@pytest.mark.parametrize(
"with_hassio",
[
@@ -269,20 +335,23 @@ async def test_generate_without_hassio(
pytest.param(False, id="without_hassio"),
],
)
async def test_restore(
async def test_restore_local_agent(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
with_hassio: bool,
backups: dict[str, BaseBackup],
snapshot: SnapshotAssertion,
) -> None:
"""Test calling the restore command."""
await setup_backup_integration(hass, with_hassio=with_hassio)
await setup_backup_integration(hass, with_hassio=with_hassio, backups=backups)
restart_calls = async_mock_service(hass, "homeassistant", "restart")
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch(
"homeassistant.components.backup.manager.BackupManager.async_restore_backup",
with (
patch("pathlib.Path.exists", return_value=True),
patch("pathlib.Path.write_text"),
):
await client.send_json_auto_id(
{
@@ -292,6 +361,50 @@ async def test_restore(
}
)
assert await client.receive_json() == snapshot
assert len(restart_calls) == snapshot
@pytest.mark.parametrize(
("remote_agents", "backups"),
[
(["remote"], {}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_ABC123]}),
],
)
@pytest.mark.parametrize(
"with_hassio",
[
pytest.param(True, id="with_hassio"),
pytest.param(False, id="without_hassio"),
],
)
async def test_restore_remote_agent(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
with_hassio: bool,
remote_agents: list[str],
backups: dict[str, BaseBackup],
snapshot: SnapshotAssertion,
) -> None:
"""Test calling the restore command."""
await setup_backup_integration(
hass, with_hassio=with_hassio, backups=backups, remote_agents=remote_agents
)
restart_calls = async_mock_service(hass, "homeassistant", "restart")
client = await hass_ws_client(hass)
await hass.async_block_till_done()
with patch("pathlib.Path.write_text"):
await client.send_json_auto_id(
{
"type": "backup/restore",
"backup_id": "abc123",
"agent_id": "test.remote",
}
)
assert await client.receive_json() == snapshot
assert len(restart_calls) == snapshot
@pytest.mark.parametrize(