Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Don't mock out filesystem operations in backup tests (#162877)

Browse Source
This commit is contained in:
Erik Montnemery
2026-02-13 21:39:34 +01:00
committed by GitHub
parent b11a75d438
commit e16a8ed20e
25 changed files with 141 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
tests/components/backup/conftest.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
from asyncio import Future
from collections.abc import Generator
from pathlib import Path
import shutil
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
@@ -13,11 +14,31 @@ from homeassistant.components.backup import DOMAIN
from homeassistant.components.backup.manager import NewBackup, WrittenBackup
from homeassistant.core import HomeAssistant
from .common import TEST_BACKUP_PATH_ABC123, TEST_BACKUP_PATH_DEF456
from tests.common import get_fixture_path
@pytest.fixture
def available_backups() -> list[Path]:
"""Fixture to provide available backup files."""
return []
@pytest.fixture
def hass_config_dir(tmp_path: Path, available_backups: list[Path]) -> str:
"""Fixture to create a temporary config directory, populated with test files."""
shutil.copytree(
get_fixture_path("config_dir_contents", DOMAIN),
tmp_path,
symlinks=True,
dirs_exist_ok=True,
)
for backup in available_backups:
(get_fixture_path("test_backups", DOMAIN) / backup).copy_into(
tmp_path / "backups"
)
return tmp_path.as_posix()
@pytest.fixture(name="instance_id", autouse=True)
def instance_id_fixture(hass: HomeAssistant) -> Generator[None]:
"""Mock instance ID."""
@@ -38,74 +59,6 @@ def mocked_json_bytes_fixture() -> Generator[Mock]:
yield mocked_json_bytes
@pytest.fixture(name="mocked_tarfile")
def mocked_tarfile_fixture() -> Generator[Mock]:
"""Mock tarfile."""
with patch(
"homeassistant.components.backup.manager.SecureTarFile"
) as mocked_tarfile:
yield mocked_tarfile
@pytest.fixture(name="path_glob")
def path_glob_fixture(hass: HomeAssistant) -> Generator[MagicMock]:
"""Mock path glob."""
with patch(
"pathlib.Path.glob",
return_value=[
Path(hass.config.path()) / "backups" / TEST_BACKUP_PATH_ABC123,
Path(hass.config.path()) / "backups" / TEST_BACKUP_PATH_DEF456,
],
) as path_glob:
yield path_glob
CONFIG_DIR = {
"tests/testing_config": [
Path("test.txt"),
Path(".DS_Store"),
Path(".storage"),
Path("another_subdir"),
Path("backups"),
Path("tmp_backups"),
Path("tts"),
Path("home-assistant_v2.db"),
],
"/backups": [
Path("backups/backup.tar"),
Path("backups/not_backup"),
],
"/another_subdir": [
Path("another_subdir/.DS_Store"),
Path("another_subdir/backups"),
Path("another_subdir/tts"),
],
"another_subdir/backups": [
Path("another_subdir/backups/backup.tar"),
Path("another_subdir/backups/not_backup"),
],
"another_subdir/tts": [
Path("another_subdir/tts/voice.mp3"),
],
"/tmp_backups": [ # noqa: S108
Path("tmp_backups/forgotten_backup.tar"),
Path("tmp_backups/not_backup"),
],
"/tts": [
Path("tts/voice.mp3"),
],
}
CONFIG_DIR_DIRS = {
Path(".storage"),
Path("another_subdir"),
Path("another_subdir/backups"),
Path("another_subdir/tts"),
Path("backups"),
Path("tmp_backups"),
Path("tts"),
}
@pytest.fixture(name="create_backup")
def mock_create_backup() -> Generator[AsyncMock]:
"""Mock manager create backup."""
@@ -125,43 +78,15 @@ def mock_create_backup() -> Generator[AsyncMock]:
yield mock_create_backup
@pytest.fixture(name="mock_backup_generation")
def mock_backup_generation_fixture(
hass: HomeAssistant, mocked_json_bytes: Mock, mocked_tarfile: Mock
) -> Generator[None]:
"""Mock backup generator."""
@pytest.fixture(name="mock_ha_version")
def mock_ha_version_fixture(hass: HomeAssistant) -> Generator[None]:
"""Mock HA version.
with (
patch(
"pathlib.Path.iterdir",
lambda x: CONFIG_DIR.get(f"{x.parent.name}/{x.name}", []),
),
patch("pathlib.Path.stat", return_value=MagicMock(st_size=123)),
patch("pathlib.Path.is_file", lambda x: x not in CONFIG_DIR_DIRS),
patch("pathlib.Path.is_dir", lambda x: x in CONFIG_DIR_DIRS),
patch(
"pathlib.Path.exists",
lambda x: (
x
not in (
Path(hass.config.path("backups")),
Path(hass.config.path("tmp_backups")),
)
),
),
patch(
"pathlib.Path.is_symlink",
lambda _: False,
),
patch(
"pathlib.Path.mkdir",
MagicMock(),
),
patch(
"homeassistant.components.backup.manager.HAVERSION",
"2025.1.0",
),
):
The HA version is included in backup metadata. We mock it for the benefit
of tests that check the exact content of the metadata.
"""
with patch("homeassistant.components.backup.manager.HAVERSION", "2025.1.0"):
yield

0
tests/components/backup/fixtures/config_dir_contents/.storage/hacs.hacs Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/another_subdir/backups/backup.tar Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/another_subdir/backups/not_backup Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/another_subdir/tts/voice.mp3 Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/backups/not_backup Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/home-assistant_v2.db Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/test.txt Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/tmp_backups/forgotten_backup.tar Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/tmp_backups/not_backup Normal file
View File

0
tests/components/backup/fixtures/config_dir_contents/tts/voice.mp3 Normal file
View File

BIN
tests/components/backup/fixtures/test_backups/abc123.tar Normal file
View File

Binary file not shown.

BIN
tests/components/backup/fixtures/test_backups/backup_v2_compressed.tar Normal file
View File

Binary file not shown.

BIN
tests/components/backup/fixtures/test_backups/backup_v2_compressed_protected.tar Normal file
View File

Binary file not shown.

BIN
tests/components/backup/fixtures/test_backups/backup_v2_uncompressed.tar Normal file
View File

Binary file not shown.

BIN
tests/components/backup/fixtures/test_backups/backup_v2_uncompressed_protected.tar Normal file
View File

Binary file not shown.

BIN
tests/components/backup/fixtures/test_backups/custom_def456.tar Normal file
View File

Binary file not shown.

30
tests/components/backup/snapshots/test_backup.ambr
View File

@@ -1,5 +1,5 @@
# serializer version: 1
# name: test_delete_backup[found_backups0-abc123-1-unlink_path0]
# name: test_delete_backup[available_backups0-abc123-1-unlink_path0]
dict({
'id': 1,
'result': dict({
@@ -10,7 +10,7 @@
'type': 'result',
})
# ---
# name: test_delete_backup[found_backups1-def456-1-unlink_path1]
# name: test_delete_backup[available_backups1-def456-1-unlink_path1]
dict({
'id': 1,
'result': dict({
@@ -21,7 +21,7 @@
'type': 'result',
})
# ---
# name: test_delete_backup[found_backups2-abc123-0-None]
# name: test_delete_backup[available_backups2-abc123-0-None]
dict({
'id': 1,
'result': dict({
@@ -32,7 +32,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[mock_read_backup]
# name: test_load_backups[read_backup-available_backups0]
dict({
'id': 1,
'result': dict({
@@ -47,7 +47,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[mock_read_backup].1
# name: test_load_backups[read_backup-available_backups0].1
dict({
'id': 2,
'result': dict({
@@ -65,7 +65,7 @@
'agents': dict({
'backup.local': dict({
'protected': False,
'size': 0,
'size': 10240,
}),
}),
'backup_id': 'abc123',
@@ -96,7 +96,7 @@
'agents': dict({
'backup.local': dict({
'protected': False,
'size': 1,
'size': 10240,
}),
}),
'backup_id': 'def456',
@@ -133,7 +133,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect1]
# name: test_load_backups[side_effect1-available_backups0]
dict({
'id': 1,
'result': dict({
@@ -148,7 +148,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect1].1
# name: test_load_backups[side_effect1-available_backups0].1
dict({
'id': 2,
'result': dict({
@@ -167,7 +167,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect2]
# name: test_load_backups[side_effect2-available_backups0]
dict({
'id': 1,
'result': dict({
@@ -182,7 +182,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect2].1
# name: test_load_backups[side_effect2-available_backups0].1
dict({
'id': 2,
'result': dict({
@@ -201,7 +201,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect3]
# name: test_load_backups[side_effect3-available_backups0]
dict({
'id': 1,
'result': dict({
@@ -216,7 +216,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect3].1
# name: test_load_backups[side_effect3-available_backups0].1
dict({
'id': 2,
'result': dict({
@@ -235,7 +235,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect4]
# name: test_load_backups[side_effect4-available_backups0]
dict({
'id': 1,
'result': dict({
@@ -250,7 +250,7 @@
'type': 'result',
})
# ---
# name: test_load_backups[side_effect4].1
# name: test_load_backups[side_effect4-available_backups0].1
dict({
'id': 2,
'result': dict({

39
tests/components/backup/test_backup.py
View File

@@ -12,7 +12,7 @@ from unittest.mock import MagicMock, mock_open, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup import DOMAIN, AgentBackup
from homeassistant.components.backup import DOMAIN, backup
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@@ -25,30 +25,23 @@ from .common import (
from tests.typing import ClientSessionGenerator, WebSocketGenerator
def mock_read_backup(backup_path: Path) -> AgentBackup:
"""Mock read backup."""
mock_backups = {
"abc123": TEST_BACKUP_ABC123,
"custom_def456": TEST_BACKUP_DEF456,
}
return mock_backups[backup_path.stem]
real_read_backup = backup.read_backup
@pytest.fixture(name="read_backup")
def read_backup_fixture(path_glob: MagicMock) -> Generator[MagicMock]:
def read_backup_fixture() -> Generator[MagicMock]:
"""Mock read backup."""
with patch(
"homeassistant.components.backup.backup.read_backup",
side_effect=mock_read_backup,
) as read_backup:
with patch("homeassistant.components.backup.backup.read_backup") as read_backup:
yield read_backup
@pytest.mark.parametrize(
"available_backups", [[TEST_BACKUP_PATH_ABC123, TEST_BACKUP_PATH_DEF456]]
)
@pytest.mark.parametrize(
"side_effect",
[
mock_read_backup,
real_read_backup,
OSError("Boom"),
TarError("Boom"),
json.JSONDecodeError("Boom", "test", 1),
@@ -74,7 +67,11 @@ async def test_load_backups(
# load and list backups
await client.send_json_auto_id({"type": "backup/info"})
assert await client.receive_json() == snapshot
response = await client.receive_json()
response["result"]["backups"] = sorted(
response["result"]["backups"], key=lambda b: b["backup_id"]
)
assert response == snapshot
async def test_upload(
@@ -106,9 +103,8 @@ async def test_upload(
assert move_mock.mock_calls[0].args[1].name == "Test_1970-01-01_00.00_00000000.tar"
@pytest.mark.usefixtures("read_backup")
@pytest.mark.parametrize(
("found_backups", "backup_id", "unlink_calls", "unlink_path"),
("available_backups", "backup_id", "unlink_calls", "unlink_path"),
[
(
[TEST_BACKUP_PATH_ABC123, TEST_BACKUP_PATH_DEF456],
@@ -122,7 +118,7 @@ async def test_upload(
1,
TEST_BACKUP_PATH_DEF456,
),
(([], TEST_BACKUP_ABC123.backup_id, 0, None)),
([], TEST_BACKUP_ABC123.backup_id, 0, None),
],
)
async def test_delete_backup(
@@ -130,8 +126,6 @@ async def test_delete_backup(
caplog: pytest.LogCaptureFixture,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
path_glob: MagicMock,
found_backups: list[Path],
backup_id: str,
unlink_calls: int,
unlink_path: Path | None,
@@ -140,7 +134,6 @@ async def test_delete_backup(
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
client = await hass_ws_client(hass)
path_glob.return_value = found_backups
with (
patch("pathlib.Path.unlink", autospec=True) as unlink,
@@ -152,4 +145,4 @@ async def test_delete_backup(
assert unlink.call_count == unlink_calls
for call in unlink.mock_calls:
assert call.args[0] == unlink_path
assert Path(call.args[0].name) == unlink_path

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

@@ -2,7 +2,6 @@
from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup.const import DOMAIN
@@ -18,7 +17,6 @@ from tests.common import snapshot_platform
from tests.typing import WebSocketGenerator
@pytest.mark.usefixtures("mock_backup_generation")
async def test_event_entity(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@@ -34,7 +32,6 @@ async def test_event_entity(
await snapshot_platform(hass, entity_registry, snapshot, entry.entry_id)
@pytest.mark.usefixtures("mock_backup_generation")
async def test_event_entity_backup_completed(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@@ -66,7 +63,6 @@ async def test_event_entity_backup_completed(
assert state.attributes[ATTR_FAILED_REASON] is None
@pytest.mark.usefixtures("mock_backup_generation")
async def test_event_entity_backup_failed(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,

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

@@ -4,6 +4,7 @@ import asyncio
from collections.abc import AsyncIterator
from io import BytesIO, StringIO
import json
from pathlib import Path
import re
import tarfile
from typing import Any
@@ -23,7 +24,12 @@ from homeassistant.components.backup import (
from homeassistant.components.backup.const import DOMAIN
from homeassistant.core import HomeAssistant
from .common import TEST_BACKUP_ABC123, aiter_from_iter, setup_backup_integration
from .common import (
TEST_BACKUP_ABC123,
TEST_BACKUP_PATH_ABC123,
aiter_from_iter,
setup_backup_integration,
)
from tests.common import MockUser, get_fixture_path
from tests.typing import ClientSessionGenerator
@@ -43,6 +49,7 @@ PROTECTED_BACKUP = AgentBackup(
)
@pytest.mark.parametrize("available_backups", [[TEST_BACKUP_PATH_ABC123]])
async def test_downloading_local_backup(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
@@ -52,22 +59,8 @@ async def test_downloading_local_backup(
client = await hass_client()
with (
patch(
"homeassistant.components.backup.backup.CoreLocalBackupAgent.async_get_backup",
return_value=TEST_BACKUP_ABC123,
),
patch(
"homeassistant.components.backup.backup.CoreLocalBackupAgent.get_backup_path",
),
patch("pathlib.Path.exists", return_value=True),
patch(
"homeassistant.components.backup.http.FileResponse",
return_value=web.Response(text=""),
),
):
resp = await client.get("/api/backup/download/abc123?agent_id=backup.local")
assert resp.status == 200
resp = await client.get("/api/backup/download/abc123?agent_id=backup.local")
assert resp.status == 200
async def test_downloading_remote_backup(
@@ -87,27 +80,21 @@ async def test_downloading_remote_backup(
assert await resp.content.read() == b"backup data"
@pytest.mark.parametrize("available_backups", [[TEST_BACKUP_PATH_ABC123]])
async def test_downloading_local_encrypted_backup_file_not_found(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
) -> None:
"""Test downloading a local backup file."""
"""Test downloading a missing local backup file."""
await setup_backup_integration(hass)
client = await hass_client()
with (
patch(
"homeassistant.components.backup.backup.CoreLocalBackupAgent.async_get_backup",
return_value=TEST_BACKUP_ABC123,
),
patch(
"homeassistant.components.backup.backup.CoreLocalBackupAgent.get_backup_path",
),
):
resp = await client.get(
"/api/backup/download/abc123?agent_id=backup.local&password=blah"
)
assert resp.status == 404
Path(hass.config.path("backups/abc123.tar")).unlink()
resp = await client.get(
"/api/backup/download/abc123?agent_id=backup.local&password=blah"
)
assert resp.status == 404
@pytest.mark.usefixtures("mock_backups")
@@ -241,7 +228,7 @@ async def test_downloading_backup_not_found(
client = await hass_client()
resp = await client.get("/api/backup/download/abc123?agent_id=backup.local")
resp = await client.get("/api/backup/download/abc1234?agent_id=backup.local")
assert resp.status == 404

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

@@ -24,6 +24,7 @@ from unittest.mock import (
from freezegun.api import FrozenDateTimeFactory
import pytest
from securetar import SecureTarFile
from homeassistant.components.backup import (
DOMAIN,
@@ -70,6 +71,7 @@ from tests.typing import ClientSessionGenerator, WebSocketGenerator
_EXPECTED_FILES = [
"test.txt",
".storage",
".storage/hacs.hacs",
"another_subdir",
"another_subdir/backups",
"another_subdir/backups/backup.tar",
@@ -112,12 +114,10 @@ def mock_read_backup(backup_path: Path) -> AgentBackup:
return mock_backups[backup_path.stem]
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.usefixtures("mock_ha_version")
async def test_create_backup_service(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mocked_json_bytes: Mock,
mocked_tarfile: Mock,
) -> None:
"""Test create backup service."""
await setup_backup_integration(hass)
@@ -161,7 +161,7 @@ async def test_create_backup_service(
)
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.usefixtures("mock_ha_version")
@pytest.mark.parametrize(
("manager_kwargs", "expected_writer_kwargs"),
[
@@ -312,8 +312,6 @@ async def test_create_backup_service(
async def test_async_create_backup(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
mocked_json_bytes: Mock,
mocked_tarfile: Mock,
manager_kwargs: dict[str, Any],
expected_writer_kwargs: dict[str, Any],
) -> None:
@@ -342,7 +340,6 @@ async def test_async_create_backup(
assert create_backup.call_args == call(**expected_writer_kwargs)
@pytest.mark.usefixtures("mock_backup_generation")
async def test_create_backup_when_busy(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@@ -419,7 +416,7 @@ async def test_create_backup_wrong_parameters(
assert result["error"]["message"] == expected_error
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.usefixtures("mock_ha_version")
@pytest.mark.parametrize(
(
"agent_ids",
@@ -519,9 +516,7 @@ async def test_initiate_backup(
hass_ws_client: WebSocketGenerator,
freezer: FrozenDateTimeFactory,
mocked_json_bytes: Mock,
mocked_tarfile: Mock,
generate_backup_id: MagicMock,
path_glob: MagicMock,
params: dict[str, Any],
agent_ids: list[str],
backup_directory: str,
@@ -540,7 +535,6 @@ async def test_initiate_backup(
include_database = params.get("include_database", True)
password = params.get("password")
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -664,26 +658,31 @@ async def test_initiate_backup(
"with_automatic_settings": False,
}
outer_tar = mocked_tarfile.return_value
core_tar = outer_tar.create_inner_tar.return_value.__enter__.return_value
expected_files = [call(hass.config.path(), arcname="data", recursive=False)] + [
call(file, arcname=f"data/{file}", recursive=False)
for file in _EXPECTED_FILES_WITH_DATABASE[include_database]
]
assert core_tar.add.call_args_list == expected_files
expected_files = {
f"data/{file}" for file in _EXPECTED_FILES_WITH_DATABASE[include_database]
}
expected_files.add("data")
tar_file_path = str(mocked_tarfile.call_args_list[0][0][0])
backup_directory = hass.config.path(backup_directory)
assert tar_file_path == f"{backup_directory}/{expected_filename}"
with tarfile.TarFile(
hass.config.path(f"{backup_directory}/{expected_filename}"), mode="r"
) as outer_tar:
core_tar_io = outer_tar.extractfile("homeassistant.tar.gz")
assert core_tar_io is not None
with SecureTarFile(
fileobj=core_tar_io,
gzip=True,
key=password_to_key(password) if password is not None else None,
mode="r",
) as core_tar:
assert set(core_tar.getnames()) == expected_files
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.usefixtures("mock_ha_version")
@pytest.mark.parametrize("exception", [BackupAgentError("Boom!"), Exception("Boom!")])
async def test_initiate_backup_with_agent_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
generate_backup_id: MagicMock,
path_glob: MagicMock,
hass_storage: dict[str, Any],
exception: Exception,
) -> None:
@@ -781,8 +780,6 @@ async def test_initiate_backup_with_agent_error(
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -861,7 +858,7 @@ async def test_initiate_backup_with_agent_error(
new_expected_backup_data = {
"addons": [],
"agents": {"backup.local": {"protected": False, "size": 123}},
"agents": {"backup.local": {"protected": False, "size": 10240}},
"backup_id": "abc123",
"database_included": True,
"date": ANY,
@@ -911,7 +908,6 @@ async def test_initiate_backup_with_agent_error(
assert mock_agents["test.remote"].async_delete_backup.call_count == 1
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
("create_backup_command", "issues_after_create_backup"),
[
@@ -1332,7 +1328,6 @@ async def test_create_backup_failure_raises_issue(
assert issue.translation_placeholders == issue_data["translation_placeholders"]
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
"exception", [BackupReaderWriterError("Boom!"), BaseException("Boom!")]
)
@@ -1340,7 +1335,6 @@ async def test_initiate_backup_non_agent_upload_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
generate_backup_id: MagicMock,
path_glob: MagicMock,
hass_storage: dict[str, Any],
exception: Exception,
) -> None:
@@ -1350,8 +1344,6 @@ async def test_initiate_backup_non_agent_upload_error(
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -1425,7 +1417,6 @@ async def test_initiate_backup_non_agent_upload_error(
assert DOMAIN not in hass_storage
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
"exception", [BackupReaderWriterError("Boom!"), Exception("Boom!")]
)
@@ -1433,7 +1424,6 @@ async def test_initiate_backup_with_task_error(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
generate_backup_id: MagicMock,
path_glob: MagicMock,
create_backup: AsyncMock,
exception: Exception,
) -> None:
@@ -1447,8 +1437,6 @@ async def test_initiate_backup_with_task_error(
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -1503,7 +1491,6 @@ async def test_initiate_backup_with_task_error(
assert backup_id == generate_backup_id.return_value
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
(
"open_call_count",
@@ -1526,7 +1513,6 @@ async def test_initiate_backup_file_error_upload_to_agents(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
generate_backup_id: MagicMock,
path_glob: MagicMock,
open_call_count: int,
open_exception: Exception | None,
read_call_count: int,
@@ -1543,8 +1529,6 @@ async def test_initiate_backup_file_error_upload_to_agents(
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -1629,7 +1613,6 @@ async def test_initiate_backup_file_error_upload_to_agents(
assert unlink_mock.call_count == unlink_call_count
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
(
"mkdir_call_count",
@@ -1650,7 +1633,6 @@ async def test_initiate_backup_file_error_create_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
generate_backup_id: MagicMock,
path_glob: MagicMock,
caplog: pytest.LogCaptureFixture,
mkdir_call_count: int,
mkdir_exception: Exception | None,
@@ -1667,8 +1649,6 @@ async def test_initiate_backup_file_error_create_backup(
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -1879,7 +1859,6 @@ async def test_exception_platform_pre(hass: HomeAssistant) -> None:
),
],
)
@pytest.mark.usefixtures("mock_backup_generation")
async def test_exception_platform_post(
hass: HomeAssistant,
unhandled_error: Exception | None,
@@ -2004,7 +1983,6 @@ async def test_receive_backup(
assert unlink_mock.call_count == temp_file_unlink_call_count
@pytest.mark.usefixtures("mock_backup_generation")
async def test_receive_backup_busy_manager(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
@@ -2068,13 +2046,11 @@ async def test_receive_backup_busy_manager(
await hass.async_block_till_done()
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize("exception", [BackupAgentError("Boom!"), Exception("Boom!")])
async def test_receive_backup_agent_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
path_glob: MagicMock,
hass_storage: dict[str, Any],
exception: Exception,
) -> None:
@@ -2172,8 +2148,6 @@ async def test_receive_backup_agent_error(
client = await hass_client()
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -2294,13 +2268,11 @@ async def test_receive_backup_agent_error(
assert mock_agents["test.remote"].async_delete_backup.call_count == 0
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize("exception", [asyncio.CancelledError("Boom!")])
async def test_receive_backup_non_agent_upload_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
path_glob: MagicMock,
hass_storage: dict[str, Any],
exception: Exception,
) -> None:
@@ -2310,8 +2282,6 @@ async def test_receive_backup_non_agent_upload_error(
client = await hass_client()
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -2388,7 +2358,6 @@ async def test_receive_backup_non_agent_upload_error(
assert unlink_mock.call_count == 0
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
(
"open_call_count",
@@ -2408,7 +2377,6 @@ async def test_receive_backup_file_write_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
path_glob: MagicMock,
open_call_count: int,
open_exception: Exception | None,
write_call_count: int,
@@ -2422,8 +2390,6 @@ async def test_receive_backup_file_write_error(
client = await hass_client()
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -2495,7 +2461,6 @@ async def test_receive_backup_file_write_error(
assert open_mock.return_value.close.call_count == close_call_count
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
"exception",
[
@@ -2509,7 +2474,6 @@ async def test_receive_backup_read_tar_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
path_glob: MagicMock,
exception: Exception,
) -> None:
"""Test read tar error during backup receive."""
@@ -2518,8 +2482,6 @@ async def test_receive_backup_read_tar_error(
client = await hass_client()
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -2590,7 +2552,6 @@ async def test_receive_backup_read_tar_error(
assert read_backup.call_count == 1
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
(
"open_call_count",
@@ -2664,7 +2625,6 @@ async def test_receive_backup_file_read_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
path_glob: MagicMock,
open_call_count: int,
open_exception: list[Exception | None],
read_call_count: int,
@@ -2683,8 +2643,6 @@ async def test_receive_backup_file_read_error(
client = await hass_client()
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
@@ -2771,7 +2729,9 @@ async def test_receive_backup_file_read_error(
assert unlink_mock.call_count == unlink_call_count
@pytest.mark.usefixtures("path_glob")
@pytest.mark.parametrize(
"available_backups", [[TEST_BACKUP_PATH_ABC123, TEST_BACKUP_PATH_DEF456]]
)
@pytest.mark.parametrize(
(
"agent_id",
@@ -2921,7 +2881,7 @@ async def test_restore_backup(
assert mocked_service_call.called
@pytest.mark.usefixtures("path_glob")
@pytest.mark.parametrize("available_backups", [[TEST_BACKUP_PATH_ABC123]])
@pytest.mark.parametrize(
("agent_id", "dir"), [(LOCAL_AGENT_ID, "backups"), ("test.remote", "tmp_backups")]
)
@@ -3001,7 +2961,7 @@ async def test_restore_backup_wrong_password(
mocked_service_call.assert_not_called()
@pytest.mark.usefixtures("path_glob")
@pytest.mark.parametrize("available_backups", [[TEST_BACKUP_PATH_ABC123]])
@pytest.mark.parametrize(
("parameters", "expected_error", "expected_reason"),
[
@@ -3093,7 +3053,6 @@ async def test_restore_backup_wrong_parameters(
mocked_service_call.assert_not_called()
@pytest.mark.usefixtures("mock_backup_generation")
async def test_restore_backup_when_busy(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@@ -3123,7 +3082,6 @@ async def test_restore_backup_when_busy(
assert result["error"]["message"] == "Backup manager busy: create_backup"
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
("exception", "error_code", "error_message", "expected_reason"),
[
@@ -3208,7 +3166,6 @@ async def test_restore_backup_agent_error(
assert mocked_service_call.call_count == 0
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
(
"open_call_count",
@@ -3353,6 +3310,7 @@ async def test_restore_backup_file_error(
assert mocked_service_call.call_count == 0
@pytest.mark.usefixtures("mock_ha_version")
@pytest.mark.parametrize(
("commands", "agent_ids", "password", "protected_backup", "inner_tar_key"),
[
@@ -3477,13 +3435,10 @@ async def test_restore_backup_file_error(
),
],
)
@pytest.mark.usefixtures("mock_backup_generation")
async def test_initiate_backup_per_agent_encryption(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
generate_backup_id: MagicMock,
mocked_tarfile: Mock,
path_glob: MagicMock,
commands: dict[str, Any],
agent_ids: list[str],
password: str | None,
@@ -3495,8 +3450,6 @@ async def test_initiate_backup_per_agent_encryption(
ws_client = await hass_ws_client(hass)
path_glob.return_value = []
await ws_client.send_json_auto_id({"type": "backup/info"})
result = await ws_client.receive_json()
assert result["success"] is True
@@ -3526,6 +3479,7 @@ async def test_initiate_backup_per_agent_encryption(
with (
patch("pathlib.Path.open", mock_open(read_data=b"test")),
patch("securetar.SecureTarFile.create_inner_tar") as mock_create_inner_tar,
):
await ws_client.send_json_auto_id(
{
@@ -3550,9 +3504,7 @@ async def test_initiate_backup_per_agent_encryption(
await hass.async_block_till_done()
mocked_tarfile.return_value.create_inner_tar.assert_called_once_with(
ANY, gzip=True, key=inner_tar_key
)
mock_create_inner_tar.assert_called_once_with(ANY, gzip=True, key=inner_tar_key)
result = await ws_client.receive_json()
assert result["event"] == {

2
tests/components/backup/test_sensors.py
View File

@@ -4,7 +4,6 @@ from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.backup import store
@@ -19,7 +18,6 @@ from tests.common import async_fire_time_changed, snapshot_platform
from tests.typing import WebSocketGenerator
@pytest.mark.usefixtures("mock_backup_generation")
async def test_sensors(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,

38
tests/components/backup/test_util.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
import asyncio
from collections.abc import AsyncIterator
import dataclasses
from pathlib import Path
import tarfile
from unittest.mock import Mock, patch
@@ -129,22 +130,39 @@ def test_read_backup(backup_json_content: bytes, expected_backup: AgentBackup) -
assert backup == expected_backup
@pytest.mark.parametrize("password", [None, "hunter2"])
def test_validate_password(password: str | None) -> None:
@pytest.mark.parametrize(
("backup", "password", "validation_result"),
[
# Backup not protected, no password provided -> validation passes
(Path("backup_v2_compressed.tar"), None, True),
(Path("backup_v2_uncompressed.tar"), None, True),
# Backup not protected, password provided -> validation fails
(Path("backup_v2_compressed.tar"), "hunter2", False),
(Path("backup_v2_uncompressed.tar"), "hunter2", False),
# Backup protected, correct password provided -> validation passes
(Path("backup_v2_compressed_protected.tar"), "hunter2", True),
(Path("backup_v2_uncompressed_protected.tar"), "hunter2", True),
# Backup protected, no password provided -> validation fails
(Path("backup_v2_compressed_protected.tar"), None, False),
(Path("backup_v2_uncompressed_protected.tar"), None, False),
# Backup protected, wrong password provided -> validation fails
(Path("backup_v2_compressed_protected.tar"), "wrong_password", False),
(Path("backup_v2_uncompressed_protected.tar"), "wrong_password", False),
],
)
def test_validate_password(
password: str | None, backup: Path, validation_result: bool
) -> None:
"""Test validating a password."""
mock_path = Mock()
test_backups = get_fixture_path("test_backups", DOMAIN)
with (
patch("homeassistant.components.backup.util.tarfile.open"),
patch("homeassistant.components.backup.util.SecureTarFile"),
):
assert validate_password(mock_path, password) is True
assert validate_password(test_backups / backup, password) == validation_result
@pytest.mark.parametrize("password", [None, "hunter2"])
@pytest.mark.parametrize("secure_tar_side_effect", [tarfile.ReadError, Exception])
def test_validate_password_wrong_password(
password: str | None, secure_tar_side_effect: Exception
def test_validate_password_with_error(
password: str | None, secure_tar_side_effect: type[Exception]
) -> None:
"""Test validating a password."""
mock_path = Mock()

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

@@ -403,6 +403,7 @@ async def test_agent_delete_backup(
assert mock_agents["test.remote"].async_delete_backup.call_args == call("abc123")
@pytest.mark.usefixtures("mock_ha_version")
@pytest.mark.parametrize(
"data",
[
@@ -411,7 +412,6 @@ async def test_agent_delete_backup(
{"password": "abc123"},
],
)
@pytest.mark.usefixtures("mock_backup_generation")
async def test_generate(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
@@ -478,7 +478,6 @@ async def test_generate_wrong_parameters(
}
@pytest.mark.usefixtures("mock_backup_generation")
@pytest.mark.parametrize(
("params", "expected_extra_call_params"),
[