Add tests

This commit is contained in:
Erik
2024-11-20 17:58:27 +01:00
parent d6b18ae3de
commit 339e75e0a4
3 changed files with 253 additions and 3 deletions

View File

@@ -50,7 +50,7 @@ class SupervisorLocalBackupAgent(LocalBackupAgent):
**kwargs: Any, **kwargs: Any,
) -> None: ) -> None:
"""Download a backup file.""" """Download a backup file."""
raise NotImplementedError raise NotImplementedError("Not yet supported by supervisor")
async def async_upload_backup( async def async_upload_backup(
self, self,
@@ -71,7 +71,7 @@ class SupervisorLocalBackupAgent(LocalBackupAgent):
date=backup.date.isoformat(), date=backup.date.isoformat(),
name=backup.name, name=backup.name,
protected=backup.protected, protected=backup.protected,
size=backup.size, size=int(backup.size * 2**20),
) )
for backup in await self._client.backups.list() for backup in await self._client.backups.list()
] ]
@@ -94,7 +94,7 @@ class SupervisorLocalBackupAgent(LocalBackupAgent):
async def async_delete_backup(self, backup_id: str, **kwargs: Any) -> None: async def async_delete_backup(self, backup_id: str, **kwargs: Any) -> None:
"""Remove a backup.""" """Remove a backup."""
raise NotImplementedError raise NotImplementedError("Not yet supported by supervisor")
class SupervisorBackupReaderWriter(BackupReaderWriter): class SupervisorBackupReaderWriter(BackupReaderWriter):

View File

@@ -523,6 +523,10 @@ def supervisor_client() -> Generator[AsyncMock]:
"homeassistant.components.hassio.addon_manager.get_supervisor_client", "homeassistant.components.hassio.addon_manager.get_supervisor_client",
return_value=supervisor_client, return_value=supervisor_client,
), ),
patch(
"homeassistant.components.hassio.backup.get_supervisor_client",
return_value=supervisor_client,
),
patch( patch(
"homeassistant.components.hassio.discovery.get_supervisor_client", "homeassistant.components.hassio.discovery.get_supervisor_client",
return_value=supervisor_client, return_value=supervisor_client,

View File

@@ -0,0 +1,246 @@
"""Test supervisor backup functionality."""
from collections.abc import AsyncGenerator, Generator
from datetime import datetime
from io import StringIO
import os
from unittest.mock import AsyncMock, patch
# from aiohasupervisor import backups as supervisor_backups
from aiohasupervisor.models import backups as supervisor_backups
import pytest
from homeassistant.components.backup import DOMAIN as BACKUP_DOMAIN, BaseBackup
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .test_init import MOCK_ENVIRON
from tests.typing import ClientSessionGenerator, WebSocketGenerator
TEST_BACKUP = supervisor_backups.Backup(
compressed=False,
content=supervisor_backups.BackupContent(
addons=["ssl"],
folders=["share"],
homeassistant=True,
),
date=datetime.fromisoformat("1970-01-01T00:00:00Z"),
location="local",
name="Test",
protected=False,
size=1.0,
slug="abc123",
type=supervisor_backups.BackupType.PARTIAL,
)
@pytest.fixture(autouse=True)
def fixture_supervisor_environ() -> Generator[None]:
"""Mock os environ for supervisor."""
with patch.dict(os.environ, MOCK_ENVIRON):
yield
@pytest.fixture(autouse=True)
async def setup_integration(
hass: HomeAssistant, supervisor_client: AsyncMock
) -> AsyncGenerator[None]:
"""Set up Kitchen Sink integration."""
with (
patch("homeassistant.components.backup.is_hassio", return_value=True),
patch("homeassistant.components.backup.backup.is_hassio", return_value=True),
):
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
await hass.async_block_till_done()
yield
@pytest.mark.usefixtures("hassio_client")
async def test_agent_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test backup agent info."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/agents/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agents": [{"agent_id": "hassio.local"}],
"syncing": False,
}
@pytest.mark.usefixtures("hassio_client")
async def test_agent_list_backups(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test agent list backups."""
client = await hass_ws_client(hass)
supervisor_client.backups.list.return_value = [TEST_BACKUP]
await client.send_json_auto_id({"type": "backup/agents/list_backups"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == [
{
"agent_id": "hassio.local",
"backup_id": "abc123",
"date": "1970-01-01T00:00:00+00:00",
"name": "Test",
"protected": False,
"size": 1048576,
}
]
@pytest.mark.usefixtures("hassio_client")
async def test_agent_download(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test agent download backup."""
client = await hass_ws_client(hass)
backup_id = "abc123"
await client.send_json_auto_id(
{
"type": "backup/agents/download",
"agent_id": "hassio.local",
"backup_id": backup_id,
}
)
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "backup_agents_download"
assert response["error"]["message"] == "Not yet supported by supervisor"
@pytest.mark.usefixtures("hassio_client")
async def test_agent_upload(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test agent upload backup."""
client = await hass_client()
backup_id = "test-backup"
test_backup = BaseBackup(
backup_id=backup_id,
date="1970-01-01T00:00:00.000Z",
name="Test",
protected=False,
size=0.0,
)
supervisor_client.backups.reload.assert_not_called()
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
) as fetch_backup,
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=test_backup,
),
patch("shutil.copy"),
):
fetch_backup.return_value = test_backup
resp = await client.post(
"/api/backup/upload?agent_id=hassio.local",
data={"file": StringIO("test")},
)
assert resp.status == 201
supervisor_client.backups.reload.assert_called_once()
@pytest.mark.usefixtures("hassio_client")
async def test_agent_delete_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test agent delete backup."""
client = await hass_ws_client(hass)
backup_id = "abc123"
await client.send_json_auto_id(
{
"type": "backup/remove",
"backup_id": backup_id,
}
)
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "unknown_error"
assert response["error"]["message"] == "Unknown error"
@pytest.mark.usefixtures("hassio_client")
async def test_reader_writer_create(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test generating a backup."""
client = await hass_ws_client(hass)
supervisor_client.backups.partial_backup.return_value.job_id = "abc123"
await client.send_json_auto_id(
{"type": "backup/generate", "agent_ids": ["hassio.local"], "name": "Test"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"backup_job_id": "abc123"}
supervisor_client.backups.partial_backup.assert_called_once_with(
supervisor_backups.PartialBackupOptions(
addons=None,
background=True,
compressed=True,
folders=None,
homeassistant_exclude_database=False,
homeassistant=True,
location=None,
name="Test",
password=None,
)
)
@pytest.mark.usefixtures("hassio_client")
async def test_reader_writer_restore(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
supervisor_client: AsyncMock,
) -> None:
"""Test generating a backup."""
client = await hass_ws_client(hass)
supervisor_client.backups.list.return_value = [TEST_BACKUP]
await client.send_json_auto_id(
{"type": "backup/restore", "agent_id": "hassio.local", "backup_id": "abc123"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] is None
supervisor_client.backups.partial_restore.assert_called_once_with(
"abc123",
supervisor_backups.PartialRestoreOptions(
addons=None,
background=True,
folders=None,
homeassistant=True,
password=None,
),
)