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

Remove class backup.backup.LocalBackup (#130919)

Browse Source
This commit is contained in:
Erik Montnemery
2024-11-19 01:32:51 +01:00
committed by GitHub
parent 1ea2ace356
commit 3cd69c32ec
5 changed files with 39 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
homeassistant/components/backup/backup.py
View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from dataclasses import asdict, dataclass
import json
from pathlib import Path
from tarfile import TarError
@@ -24,17 +23,6 @@ async def async_get_backup_agents(
return [CoreLocalBackupAgent(hass)]
@dataclass(slots=True)
class LocalBackup(BaseBackup):
"""Local backup class."""
path: Path
def as_dict(self) -> dict:
"""Return a dict representation of this backup."""
return {**asdict(self), "path": self.path.as_posix()}
class CoreLocalBackupAgent(LocalBackupAgent):
"""Local backup agent for Core and Container installations."""
@@ -45,7 +33,7 @@ class CoreLocalBackupAgent(LocalBackupAgent):
super().__init__()
self._hass = hass
self._backup_dir = Path(hass.config.path("backups"))
self._backups: dict[str, LocalBackup] = {}
self._backups: dict[str, BaseBackup] = {}
self._loaded_backups = False
async def load_backups(self) -> None:
@@ -55,17 +43,16 @@ class CoreLocalBackupAgent(LocalBackupAgent):
self._backups = backups
self._loaded_backups = True
def _read_backups(self) -> dict[str, LocalBackup]:
def _read_backups(self) -> dict[str, BaseBackup]:
"""Read backups from disk."""
backups: dict[str, LocalBackup] = {}
backups: dict[str, BaseBackup] = {}
for backup_path in self._backup_dir.glob("*.tar"):
try:
base_backup = read_backup(backup_path)
backup = LocalBackup(
backup = BaseBackup(
backup_id=base_backup.backup_id,
name=base_backup.name,
date=base_backup.date,
path=backup_path,
size=round(backup_path.stat().st_size / 1_048_576, 2),
protected=base_backup.protected,
)
@@ -92,11 +79,10 @@ class CoreLocalBackupAgent(LocalBackupAgent):
**kwargs: Any,
) -> None:
"""Upload a backup."""
self._backups[metadata.backup_id] = LocalBackup(
self._backups[metadata.backup_id] = BaseBackup(
backup_id=metadata.backup_id,
date=metadata.date,
name=metadata.name,
path=path,
protected=metadata.protected,
size=round(path.stat().st_size / 1_048_576, 2),
)
@@ -111,7 +97,7 @@ class CoreLocalBackupAgent(LocalBackupAgent):
self,
backup_id: str,
**kwargs: Any,
) -> LocalBackup | None:
) -> BaseBackup | None:
"""Return a backup."""
if not self._loaded_backups:
await self.load_backups()
@@ -119,14 +105,15 @@ class CoreLocalBackupAgent(LocalBackupAgent):
if not (backup := self._backups.get(backup_id)):
return None
if not await self._hass.async_add_executor_job(backup.path.exists):
backup_path = self.get_backup_path(backup_id)
if not await self._hass.async_add_executor_job(backup_path.exists):
LOGGER.debug(
(
"Removing tracked backup (%s) that does not exists on the expected"
" path %s"
),
backup.backup_id,
backup.path,
backup_path,
)
self._backups.pop(backup_id)
return None
@@ -139,9 +126,10 @@ class CoreLocalBackupAgent(LocalBackupAgent):
async def async_remove_backup(self, backup_id: str, **kwargs: Any) -> None:
"""Remove a backup."""
if (backup := await self.async_get_backup(backup_id)) is None:
if await self.async_get_backup(backup_id) is None:
return
await self._hass.async_add_executor_job(backup.path.unlink, True)
LOGGER.debug("Removed backup located at %s", backup.path)
backup_path = self.get_backup_path(backup_id)
await self._hass.async_add_executor_job(backup_path.unlink, True)
LOGGER.debug("Removed backup located at %s", backup_path)
self._backups.pop(backup_id)

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

@@ -13,7 +13,6 @@ from homeassistant.components.backup import (
BackupUploadMetadata,
BaseBackup,
)
from homeassistant.components.backup.backup import LocalBackup
from homeassistant.components.backup.const import DATA_MANAGER
from homeassistant.components.backup.manager import Backup
from homeassistant.core import HomeAssistant
@@ -32,14 +31,6 @@ TEST_BASE_BACKUP_ABC123 = BaseBackup(
size=0.0,
)
TEST_BACKUP_PATH_ABC123 = Path("abc123.tar")
TEST_LOCAL_BACKUP_ABC123 = LocalBackup(
date="1970-01-01T00:00:00.000Z",
backup_id="abc123",
name="Test",
path=Path("abc123.tar"),
protected=False,
size=0.0,
)
TEST_BASE_BACKUP_DEF456 = BaseBackup(
backup_id="def456",

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_ABC123, setup_backup_integration
from .common import TEST_BASE_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_ABC123,
return_value=TEST_BASE_BACKUP_ABC123,
),
patch("pathlib.Path.exists", return_value=True),
patch(

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

@@ -27,7 +27,6 @@ from .common import (
LOCAL_AGENT_ID,
TEST_BACKUP_PATH_ABC123,
TEST_BASE_BACKUP_ABC123,
TEST_LOCAL_BACKUP_ABC123,
BackupAgentTest,
)
@@ -150,14 +149,14 @@ async def test_load_backups(hass: HomeAssistant, snapshot: SnapshotAssertion) ->
patch(
"homeassistant.components.backup.util.json_loads_object",
return_value={
"date": TEST_LOCAL_BACKUP_ABC123.date,
"name": TEST_LOCAL_BACKUP_ABC123.name,
"slug": TEST_LOCAL_BACKUP_ABC123.backup_id,
"date": TEST_BASE_BACKUP_ABC123.date,
"name": TEST_BASE_BACKUP_ABC123.name,
"slug": TEST_BASE_BACKUP_ABC123.backup_id,
},
),
patch(
"pathlib.Path.stat",
return_value=MagicMock(st_size=TEST_LOCAL_BACKUP_ABC123.size),
return_value=MagicMock(st_size=TEST_BASE_BACKUP_ABC123.size),
),
):
await manager.backup_agents[LOCAL_AGENT_ID].load_backups()
@@ -201,13 +200,11 @@ async def test_removing_backup(
await manager.load_platforms()
local_agent = manager.backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {
TEST_LOCAL_BACKUP_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._backups = {TEST_BASE_BACKUP_ABC123.backup_id: TEST_BASE_BACKUP_ABC123}
local_agent._loaded_backups = True
with patch("pathlib.Path.exists", return_value=True):
await manager.async_remove_backup(TEST_LOCAL_BACKUP_ABC123.backup_id)
await manager.async_remove_backup(TEST_BASE_BACKUP_ABC123.backup_id)
assert "Removed backup located at" in caplog.text
@@ -236,21 +233,20 @@ 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_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._backups = {TEST_BASE_BACKUP_ABC123.backup_id: TEST_BASE_BACKUP_ABC123}
local_agent._loaded_backups = True
path = local_agent.get_backup_path(TEST_BASE_BACKUP_ABC123.backup_id)
with patch("pathlib.Path.exists", return_value=False):
backup, agent_errors = await manager.async_get_backup(
TEST_LOCAL_BACKUP_ABC123.backup_id
TEST_BASE_BACKUP_ABC123.backup_id
)
assert backup is None
assert agent_errors == {}
assert (
f"Removing tracked backup ({TEST_LOCAL_BACKUP_ABC123.backup_id}) that "
f"does not exists on the expected path {TEST_LOCAL_BACKUP_ABC123.path}"
f"Removing tracked backup ({TEST_BASE_BACKUP_ABC123.backup_id}) that "
f"does not exists on the expected path {path}"
) in caplog.text
@@ -526,9 +522,7 @@ async def test_async_trigger_restore(
await manager.load_platforms()
local_agent = manager.backup_agents[LOCAL_AGENT_ID]
local_agent._backups = {
TEST_LOCAL_BACKUP_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._backups = {TEST_BASE_BACKUP_ABC123.backup_id: TEST_BASE_BACKUP_ABC123}
local_agent._loaded_backups = True
with (
@@ -537,7 +531,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_ABC123.backup_id, agent_id=LOCAL_AGENT_ID, password=None
TEST_BASE_BACKUP_ABC123.backup_id, agent_id=LOCAL_AGENT_ID, password=None
)
assert (
mocked_write_text.call_args[0][0]
@@ -557,9 +551,7 @@ 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_ABC123.backup_id: TEST_LOCAL_BACKUP_ABC123
}
local_agent._backups = {TEST_BASE_BACKUP_ABC123.backup_id: TEST_BASE_BACKUP_ABC123}
local_agent._loaded_backups = True
with (
@@ -568,7 +560,7 @@ 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_ABC123.backup_id,
TEST_BASE_BACKUP_ABC123.backup_id,
agent_id=LOCAL_AGENT_ID,
password="abc123",
)
@@ -591,5 +583,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_ABC123.backup_id, agent_id=LOCAL_AGENT_ID, password=None
TEST_BASE_BACKUP_ABC123.backup_id, agent_id=LOCAL_AGENT_ID, password=None
)

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

@@ -19,7 +19,6 @@ from .common import (
LOCAL_AGENT_ID,
TEST_BASE_BACKUP_ABC123,
TEST_BASE_BACKUP_DEF456,
TEST_LOCAL_BACKUP_ABC123,
BackupAgentTest,
setup_backup_integration,
)
@@ -68,7 +67,7 @@ async def test_info(
await setup_backup_integration(
hass,
with_hassio=with_hassio,
backups={LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]} | remote_backups,
backups={LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123]} | remote_backups,
remote_agents=remote_agents,
)
@@ -90,7 +89,7 @@ async def test_info_with_errors(
) -> None:
"""Test getting backup info with one unavailable agent."""
await setup_backup_integration(
hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}
hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123]}
)
hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test")
@@ -106,13 +105,13 @@ async def test_info_with_errors(
("remote_agents", "backups"),
[
([], {}),
(["remote"], {LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}),
(["remote"], {LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_DEF456]}),
(
["remote"],
{
LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123],
LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123],
"test.remote": [TEST_BASE_BACKUP_ABC123],
},
),
@@ -159,7 +158,7 @@ async def test_details_with_errors(
) -> None:
"""Test getting backup info with one unavailable agent."""
await setup_backup_integration(
hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}
hass, with_hassio=False, backups={LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123]}
)
hass.data[DATA_MANAGER].backup_agents["domain.test"] = BackupAgentTest("test")
@@ -180,13 +179,13 @@ async def test_details_with_errors(
("remote_agents", "backups"),
[
([], {}),
(["remote"], {LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]}),
(["remote"], {LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_ABC123]}),
(["remote"], {"test.remote": [TEST_BASE_BACKUP_DEF456]}),
(
["remote"],
{
LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123],
LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123],
"test.remote": [TEST_BASE_BACKUP_ABC123],
},
),
@@ -325,7 +324,7 @@ async def test_generate_without_hassio(
"backups",
[
{},
{LOCAL_AGENT_ID: [TEST_LOCAL_BACKUP_ABC123]},
{LOCAL_AGENT_ID: [TEST_BASE_BACKUP_ABC123]},
],
)
@pytest.mark.parametrize(