mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-11-04 00:19:36 +00:00
2da27937a5
* Update python to 3.12 * Fix tests and deprecations * Fix other references to 3.11 * build.json doesn't exist
712 lines
23 KiB
Python
712 lines
23 KiB
Python
"""Common test functions."""
|
|
from functools import partial
|
|
from inspect import unwrap
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
from unittest.mock import AsyncMock, MagicMock, Mock, PropertyMock, patch
|
|
from uuid import uuid4
|
|
|
|
from aiohttp import web
|
|
from aiohttp.test_utils import TestClient
|
|
from awesomeversion import AwesomeVersion
|
|
from dbus_fast import BusType
|
|
from dbus_fast.aio.message_bus import MessageBus
|
|
import pytest
|
|
from securetar import SecureTarFile
|
|
|
|
from supervisor import config as su_config
|
|
from supervisor.addons.addon import Addon
|
|
from supervisor.addons.validate import SCHEMA_ADDON_SYSTEM
|
|
from supervisor.api import RestAPI
|
|
from supervisor.backups.backup import Backup
|
|
from supervisor.backups.const import BackupType
|
|
from supervisor.backups.validate import ALL_FOLDERS
|
|
from supervisor.bootstrap import initialize_coresys
|
|
from supervisor.const import (
|
|
ATTR_ADDONS,
|
|
ATTR_ADDONS_CUSTOM_LIST,
|
|
ATTR_DATE,
|
|
ATTR_FOLDERS,
|
|
ATTR_HOMEASSISTANT,
|
|
ATTR_NAME,
|
|
ATTR_REPOSITORIES,
|
|
ATTR_SIZE,
|
|
ATTR_SLUG,
|
|
ATTR_TYPE,
|
|
ATTR_VERSION,
|
|
REQUEST_FROM,
|
|
)
|
|
from supervisor.coresys import CoreSys
|
|
from supervisor.dbus.network import NetworkManager
|
|
from supervisor.docker.manager import DockerAPI
|
|
from supervisor.docker.monitor import DockerMonitor
|
|
from supervisor.host.logs import LogsControl
|
|
from supervisor.os.manager import OSManager
|
|
from supervisor.store.addon import AddonStore
|
|
from supervisor.store.repository import Repository
|
|
from supervisor.utils.dt import utcnow
|
|
|
|
from .common import (
|
|
load_binary_fixture,
|
|
load_fixture,
|
|
load_json_fixture,
|
|
mock_dbus_services,
|
|
)
|
|
from .const import TEST_ADDON_SLUG
|
|
from .dbus_service_mocks.base import DBusServiceMock
|
|
from .dbus_service_mocks.network_connection_settings import (
|
|
ConnectionSettings as ConnectionSettingsService,
|
|
)
|
|
from .dbus_service_mocks.network_manager import NetworkManager as NetworkManagerService
|
|
|
|
# pylint: disable=redefined-outer-name, protected-access
|
|
|
|
|
|
@pytest.fixture
|
|
async def path_extern() -> None:
|
|
"""Set external path env for tests."""
|
|
os.environ["SUPERVISOR_SHARE"] = "/mnt/data/supervisor"
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def supervisor_name() -> None:
|
|
"""Set env for supervisor name."""
|
|
os.environ["SUPERVISOR_NAME"] = "hassio_supervisor"
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def docker() -> DockerAPI:
|
|
"""Mock DockerAPI."""
|
|
images = [MagicMock(tags=["ghcr.io/home-assistant/amd64-hassio-supervisor:latest"])]
|
|
|
|
with patch(
|
|
"supervisor.docker.manager.DockerClient", return_value=MagicMock()
|
|
), patch(
|
|
"supervisor.docker.manager.DockerAPI.images", return_value=MagicMock()
|
|
), patch(
|
|
"supervisor.docker.manager.DockerAPI.containers", return_value=MagicMock()
|
|
), patch(
|
|
"supervisor.docker.manager.DockerAPI.api", return_value=MagicMock()
|
|
), patch(
|
|
"supervisor.docker.manager.DockerAPI.images.list", return_value=images
|
|
), patch(
|
|
"supervisor.docker.manager.DockerAPI.info",
|
|
return_value=MagicMock(),
|
|
), patch(
|
|
"supervisor.docker.manager.DockerConfig",
|
|
return_value=MagicMock(),
|
|
), patch(
|
|
"supervisor.docker.manager.DockerAPI.unload"
|
|
):
|
|
docker_obj = DockerAPI(MagicMock())
|
|
with patch("supervisor.docker.monitor.DockerMonitor.load"):
|
|
await docker_obj.load()
|
|
|
|
docker_obj.info.logging = "journald"
|
|
docker_obj.info.storage = "overlay2"
|
|
docker_obj.info.version = "1.0.0"
|
|
|
|
docker_obj.config.registries = {}
|
|
|
|
yield docker_obj
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def dbus_session() -> None:
|
|
"""Start a dbus session."""
|
|
dbus_launch = subprocess.run(["dbus-launch"], stdout=subprocess.PIPE, check=False)
|
|
envs = dbus_launch.stdout.decode(encoding="utf-8").rstrip()
|
|
|
|
for env in envs.split("\n"):
|
|
name, value = env.split("=", 1)
|
|
os.environ[name] = value
|
|
|
|
|
|
@pytest.fixture
|
|
async def dbus_session_bus(dbus_session) -> MessageBus:
|
|
"""Return message bus connected to session dbus."""
|
|
bus = await MessageBus(bus_type=BusType.SESSION).connect()
|
|
yield bus
|
|
bus.disconnect()
|
|
|
|
|
|
@pytest.fixture
|
|
async def dbus_is_connected():
|
|
"""Mock DBusInterface.is_connected for tests."""
|
|
with patch(
|
|
"supervisor.dbus.interface.DBusInterface.is_connected",
|
|
return_value=True,
|
|
) as is_connected:
|
|
yield is_connected
|
|
|
|
|
|
@pytest.fixture(name="network_manager_services")
|
|
async def fixture_network_manager_services(
|
|
dbus_session_bus: MessageBus,
|
|
) -> dict[str, DBusServiceMock | dict[str, DBusServiceMock]]:
|
|
"""Mock all services network manager connects to."""
|
|
yield await mock_dbus_services(
|
|
{
|
|
"network_access_point": [
|
|
"/org/freedesktop/NetworkManager/AccessPoint/43099",
|
|
"/org/freedesktop/NetworkManager/AccessPoint/43100",
|
|
],
|
|
"network_active_connection": None,
|
|
"network_connection_settings": None,
|
|
"network_device_wireless": None,
|
|
"network_device": [
|
|
"/org/freedesktop/NetworkManager/Devices/1",
|
|
"/org/freedesktop/NetworkManager/Devices/3",
|
|
],
|
|
"network_dns_manager": None,
|
|
"network_ip4config": None,
|
|
"network_ip6config": None,
|
|
"network_manager": None,
|
|
"network_settings": None,
|
|
},
|
|
dbus_session_bus,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def network_manager(
|
|
network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
|
|
dbus_session_bus: MessageBus,
|
|
) -> NetworkManager:
|
|
"""Mock Network Manager."""
|
|
nm_obj = NetworkManager()
|
|
await nm_obj.connect(dbus_session_bus)
|
|
yield nm_obj
|
|
|
|
|
|
@pytest.fixture
|
|
async def network_manager_service(
|
|
network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]]
|
|
) -> NetworkManagerService:
|
|
"""Return Network Manager service mock."""
|
|
yield network_manager_services["network_manager"]
|
|
|
|
|
|
@pytest.fixture(name="connection_settings_service")
|
|
async def fixture_connection_settings_service(
|
|
network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]]
|
|
) -> ConnectionSettingsService:
|
|
"""Return mock connection settings service."""
|
|
yield network_manager_services["network_connection_settings"]
|
|
|
|
|
|
@pytest.fixture(name="udisks2_services")
|
|
async def fixture_udisks2_services(
|
|
dbus_session_bus: MessageBus,
|
|
) -> dict[str, DBusServiceMock | dict[str, DBusServiceMock]]:
|
|
"""Mock all services UDisks2 connects to."""
|
|
yield await mock_dbus_services(
|
|
{
|
|
"udisks2_block": [
|
|
"/org/freedesktop/UDisks2/block_devices/loop0",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p1",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p2",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p3",
|
|
"/org/freedesktop/UDisks2/block_devices/sda",
|
|
"/org/freedesktop/UDisks2/block_devices/sda1",
|
|
"/org/freedesktop/UDisks2/block_devices/sdb",
|
|
"/org/freedesktop/UDisks2/block_devices/sdb1",
|
|
"/org/freedesktop/UDisks2/block_devices/zram1",
|
|
],
|
|
"udisks2_drive": [
|
|
"/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291",
|
|
"/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6",
|
|
"/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56",
|
|
],
|
|
"udisks2_filesystem": [
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p1",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p3",
|
|
"/org/freedesktop/UDisks2/block_devices/sda1",
|
|
"/org/freedesktop/UDisks2/block_devices/sdb1",
|
|
"/org/freedesktop/UDisks2/block_devices/zram1",
|
|
],
|
|
"udisks2_loop": None,
|
|
"udisks2_manager": None,
|
|
"udisks2_partition_table": [
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1",
|
|
"/org/freedesktop/UDisks2/block_devices/sda",
|
|
"/org/freedesktop/UDisks2/block_devices/sdb",
|
|
],
|
|
"udisks2_partition": [
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p1",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p2",
|
|
"/org/freedesktop/UDisks2/block_devices/mmcblk1p3",
|
|
"/org/freedesktop/UDisks2/block_devices/sda1",
|
|
"/org/freedesktop/UDisks2/block_devices/sdb1",
|
|
],
|
|
},
|
|
dbus_session_bus,
|
|
)
|
|
|
|
|
|
@pytest.fixture(name="os_agent_services")
|
|
async def fixture_os_agent_services(
|
|
dbus_session_bus: MessageBus,
|
|
) -> dict[str, DBusServiceMock]:
|
|
"""Mock all services os agent connects to."""
|
|
yield await mock_dbus_services(
|
|
{
|
|
"os_agent": None,
|
|
"agent_apparmor": None,
|
|
"agent_cgroup": None,
|
|
"agent_datadisk": None,
|
|
"agent_system": None,
|
|
"agent_boards": None,
|
|
"agent_boards_yellow": None,
|
|
},
|
|
dbus_session_bus,
|
|
)
|
|
|
|
|
|
@pytest.fixture(name="all_dbus_services")
|
|
async def fixture_all_dbus_services(
|
|
dbus_session_bus: MessageBus,
|
|
network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
|
|
udisks2_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
|
|
os_agent_services: dict[str, DBusServiceMock],
|
|
) -> dict[str, DBusServiceMock | dict[str, DBusServiceMock]]:
|
|
"""Mock all dbus services supervisor uses."""
|
|
yield (
|
|
await mock_dbus_services(
|
|
{
|
|
"hostname": None,
|
|
"logind": None,
|
|
"rauc": None,
|
|
"resolved": None,
|
|
"systemd": None,
|
|
"systemd_unit": None,
|
|
"timedate": None,
|
|
},
|
|
dbus_session_bus,
|
|
)
|
|
) | network_manager_services | udisks2_services | os_agent_services
|
|
|
|
|
|
@pytest.fixture
|
|
async def coresys(
|
|
docker,
|
|
dbus_session_bus,
|
|
all_dbus_services,
|
|
aiohttp_client,
|
|
run_dir,
|
|
supervisor_name,
|
|
) -> CoreSys:
|
|
"""Create a CoreSys Mock."""
|
|
with patch("supervisor.bootstrap.initialize_system"), patch(
|
|
"supervisor.utils.sentry.sentry_sdk.init"
|
|
):
|
|
coresys_obj = await initialize_coresys()
|
|
|
|
# Mock save json
|
|
coresys_obj._ingress.save_data = MagicMock()
|
|
coresys_obj._auth.save_data = MagicMock()
|
|
coresys_obj._updater.save_data = MagicMock()
|
|
coresys_obj._config.save_data = MagicMock()
|
|
coresys_obj._jobs.save_data = MagicMock()
|
|
coresys_obj._resolution.save_data = MagicMock()
|
|
coresys_obj._addons.data.save_data = MagicMock()
|
|
coresys_obj._store.save_data = MagicMock()
|
|
coresys_obj._mounts.save_data = MagicMock()
|
|
|
|
# Mock test client
|
|
coresys_obj.arch._default_arch = "amd64"
|
|
coresys_obj.arch._supported_set = {"amd64"}
|
|
coresys_obj._machine = "qemux86-64"
|
|
coresys_obj._machine_id = uuid4()
|
|
|
|
# Mock host communication
|
|
with patch("supervisor.dbus.manager.MessageBus") as message_bus, patch(
|
|
"supervisor.dbus.manager.SOCKET_DBUS"
|
|
):
|
|
message_bus.return_value.connect = AsyncMock(return_value=dbus_session_bus)
|
|
await coresys_obj._dbus.load()
|
|
|
|
# Mock docker
|
|
coresys_obj._docker = docker
|
|
coresys_obj.docker._monitor = DockerMonitor(coresys_obj)
|
|
|
|
# Set internet state
|
|
coresys_obj.supervisor._connectivity = True
|
|
coresys_obj.host.network._connectivity = True
|
|
|
|
# Fix Paths
|
|
su_config.ADDONS_CORE = Path(
|
|
Path(__file__).parent.joinpath("fixtures"), "addons/core"
|
|
)
|
|
su_config.ADDONS_LOCAL = Path(
|
|
Path(__file__).parent.joinpath("fixtures"), "addons/local"
|
|
)
|
|
su_config.ADDONS_GIT = Path(
|
|
Path(__file__).parent.joinpath("fixtures"), "addons/git"
|
|
)
|
|
su_config.APPARMOR_DATA = Path(
|
|
Path(__file__).parent.joinpath("fixtures"), "apparmor"
|
|
)
|
|
|
|
# WebSocket
|
|
coresys_obj.homeassistant.api.get_api_state = AsyncMock(return_value="RUNNING")
|
|
coresys_obj.homeassistant._websocket._client = AsyncMock(
|
|
ha_version=AwesomeVersion("2021.2.4")
|
|
)
|
|
|
|
# Remove rate limiting decorator from fetch_data
|
|
coresys_obj.updater.fetch_data = partial(
|
|
unwrap(coresys_obj.updater.fetch_data), coresys_obj.updater
|
|
)
|
|
|
|
# Don't remove files/folders related to addons and stores
|
|
with patch("supervisor.store.git.GitRepo._remove"):
|
|
yield coresys_obj
|
|
|
|
await coresys_obj.dbus.unload()
|
|
await coresys_obj.websession.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def ha_ws_client(coresys: CoreSys) -> AsyncMock:
|
|
"""Return HA WS client mock for assertions."""
|
|
return coresys.homeassistant.websocket._client
|
|
|
|
|
|
@pytest.fixture
|
|
async def tmp_supervisor_data(coresys: CoreSys, tmp_path: Path) -> Path:
|
|
"""Patch supervisor data to be tmp_path."""
|
|
with patch.object(
|
|
su_config.CoreConfig, "path_supervisor", new=PropertyMock(return_value=tmp_path)
|
|
):
|
|
coresys.config.path_emergency.mkdir()
|
|
coresys.config.path_media.mkdir()
|
|
coresys.config.path_mounts.mkdir()
|
|
coresys.config.path_mounts_credentials.mkdir()
|
|
coresys.config.path_backup.mkdir()
|
|
coresys.config.path_tmp.mkdir()
|
|
coresys.config.path_homeassistant.mkdir()
|
|
coresys.config.path_audio.mkdir()
|
|
coresys.config.path_dns.mkdir()
|
|
coresys.config.path_share.mkdir()
|
|
coresys.config.path_addons_data.mkdir(parents=True)
|
|
coresys.config.path_addon_configs.mkdir(parents=True)
|
|
yield tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
async def journald_gateway() -> MagicMock:
|
|
"""Mock logs control."""
|
|
with patch("supervisor.host.logs.Path.is_socket", return_value=True), patch(
|
|
"supervisor.host.logs.ClientSession.get"
|
|
) as get:
|
|
get.return_value.__aenter__.return_value.text = AsyncMock(
|
|
return_value=load_fixture("logs_host.txt")
|
|
)
|
|
yield get
|
|
|
|
|
|
@pytest.fixture
|
|
def sys_machine():
|
|
"""Mock sys_machine."""
|
|
with patch("supervisor.coresys.CoreSys.machine", new_callable=PropertyMock) as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def sys_supervisor():
|
|
"""Mock sys_supervisor."""
|
|
with patch(
|
|
"supervisor.coresys.CoreSys.supervisor", new_callable=PropertyMock
|
|
) as mock:
|
|
mock.return_value = MagicMock()
|
|
yield MagicMock
|
|
|
|
|
|
@pytest.fixture
|
|
async def api_client(
|
|
aiohttp_client,
|
|
coresys: CoreSys,
|
|
request: pytest.FixtureRequest,
|
|
) -> TestClient:
|
|
"""Fixture for RestAPI client."""
|
|
|
|
request_from: str | None = getattr(request, "param", None)
|
|
|
|
@web.middleware
|
|
async def _security_middleware(request: web.Request, handler: web.RequestHandler):
|
|
"""Make request are from Core or specified add-on."""
|
|
if request_from:
|
|
request[REQUEST_FROM] = coresys.addons.get(request_from, local_only=True)
|
|
else:
|
|
request[REQUEST_FROM] = coresys.homeassistant
|
|
|
|
return await handler(request)
|
|
|
|
api = RestAPI(coresys)
|
|
api.webapp = web.Application(middlewares=[_security_middleware])
|
|
api.start = AsyncMock()
|
|
await api.load()
|
|
yield await aiohttp_client(api.webapp)
|
|
|
|
|
|
@pytest.fixture
|
|
def store_manager(coresys: CoreSys):
|
|
"""Fixture for the store manager."""
|
|
sm_obj = coresys.store
|
|
with patch("supervisor.store.data.StoreData.update", return_value=MagicMock()):
|
|
yield sm_obj
|
|
|
|
|
|
@pytest.fixture
|
|
def run_dir(tmp_path):
|
|
"""Fixture to inject hassio env."""
|
|
with patch("supervisor.core.RUN_SUPERVISOR_STATE") as mock_run:
|
|
tmp_state = Path(tmp_path, "supervisor")
|
|
mock_run.write_text = tmp_state.write_text
|
|
yield tmp_state
|
|
|
|
|
|
@pytest.fixture
|
|
def store_addon(coresys: CoreSys, tmp_path, repository):
|
|
"""Store add-on fixture."""
|
|
addon_obj = AddonStore(coresys, "test_store_addon")
|
|
|
|
coresys.addons.store[addon_obj.slug] = addon_obj
|
|
coresys.store.data.addons[addon_obj.slug] = SCHEMA_ADDON_SYSTEM(
|
|
load_json_fixture("add-on.json")
|
|
)
|
|
yield addon_obj
|
|
|
|
|
|
@pytest.fixture
|
|
async def repository(coresys: CoreSys):
|
|
"""Repository fixture."""
|
|
coresys.store._data[ATTR_REPOSITORIES].remove(
|
|
"https://github.com/hassio-addons/repository"
|
|
)
|
|
coresys.store._data[ATTR_REPOSITORIES].remove(
|
|
"https://github.com/esphome/home-assistant-addon"
|
|
)
|
|
coresys.config._data[ATTR_ADDONS_CUSTOM_LIST] = []
|
|
|
|
with patch(
|
|
"supervisor.store.validate.BUILTIN_REPOSITORIES", {"local", "core"}
|
|
), patch("supervisor.store.git.GitRepo.load", return_value=None):
|
|
await coresys.store.load()
|
|
|
|
repository_obj = Repository(
|
|
coresys, "https://github.com/awesome-developer/awesome-repo"
|
|
)
|
|
|
|
coresys.store.repositories[repository_obj.slug] = repository_obj
|
|
coresys.store._data[ATTR_REPOSITORIES].append(
|
|
"https://github.com/awesome-developer/awesome-repo"
|
|
)
|
|
|
|
yield repository_obj
|
|
|
|
|
|
@pytest.fixture
|
|
def install_addon_ssh(coresys: CoreSys, repository):
|
|
"""Install local_ssh add-on."""
|
|
store = coresys.addons.store[TEST_ADDON_SLUG]
|
|
coresys.addons.data.install(store)
|
|
coresys.addons.data._data = coresys.addons.data._schema(coresys.addons.data._data)
|
|
|
|
addon = Addon(coresys, store.slug)
|
|
coresys.addons.local[addon.slug] = addon
|
|
yield addon
|
|
|
|
|
|
@pytest.fixture
|
|
def install_addon_example(coresys: CoreSys, repository):
|
|
"""Install local_example add-on."""
|
|
store = coresys.addons.store["local_example"]
|
|
coresys.addons.data.install(store)
|
|
coresys.addons.data._data = coresys.addons.data._schema(coresys.addons.data._data)
|
|
|
|
addon = Addon(coresys, store.slug)
|
|
coresys.addons.local[addon.slug] = addon
|
|
yield addon
|
|
|
|
|
|
@pytest.fixture
|
|
async def mock_full_backup(coresys: CoreSys, tmp_path) -> Backup:
|
|
"""Mock a full backup."""
|
|
mock_backup = Backup(coresys, Path(tmp_path, "test_backup"))
|
|
mock_backup.new("test", "Test", utcnow().isoformat(), BackupType.FULL)
|
|
mock_backup.repositories = ["https://github.com/awesome-developer/awesome-repo"]
|
|
mock_backup.docker = {}
|
|
mock_backup._data[ATTR_ADDONS] = [
|
|
{
|
|
ATTR_SLUG: "local_ssh",
|
|
ATTR_NAME: "SSH",
|
|
ATTR_VERSION: "1.0.0",
|
|
ATTR_SIZE: 0,
|
|
}
|
|
]
|
|
mock_backup._data[ATTR_FOLDERS] = ALL_FOLDERS
|
|
mock_backup._data[ATTR_HOMEASSISTANT] = {
|
|
ATTR_VERSION: AwesomeVersion("2022.8.0"),
|
|
ATTR_SIZE: 0,
|
|
}
|
|
coresys.backups._backups = {"test": mock_backup}
|
|
yield mock_backup
|
|
|
|
|
|
@pytest.fixture
|
|
async def mock_partial_backup(coresys: CoreSys, tmp_path) -> Backup:
|
|
"""Mock a partial backup."""
|
|
mock_backup = Backup(coresys, Path(tmp_path, "test_backup"))
|
|
mock_backup.new("test", "Test", utcnow().isoformat(), BackupType.PARTIAL)
|
|
mock_backup.repositories = ["https://github.com/awesome-developer/awesome-repo"]
|
|
mock_backup.docker = {}
|
|
mock_backup._data[ATTR_ADDONS] = [
|
|
{
|
|
ATTR_SLUG: "local_ssh",
|
|
ATTR_NAME: "SSH",
|
|
ATTR_VERSION: "1.0.0",
|
|
ATTR_SIZE: 0,
|
|
}
|
|
]
|
|
mock_backup._data[ATTR_FOLDERS] = ALL_FOLDERS
|
|
mock_backup._data[ATTR_HOMEASSISTANT] = {
|
|
ATTR_VERSION: AwesomeVersion("2022.8.0"),
|
|
ATTR_SIZE: 0,
|
|
}
|
|
coresys.backups._backups = {"test": mock_backup}
|
|
yield mock_backup
|
|
|
|
|
|
@pytest.fixture
|
|
async def backups(
|
|
coresys: CoreSys, tmp_path, request: pytest.FixtureRequest
|
|
) -> list[Backup]:
|
|
"""Create and return mock backups."""
|
|
for i in range(request.param if hasattr(request, "param") else 5):
|
|
slug = f"sn{i + 1}"
|
|
temp_tar = Path(tmp_path, f"{slug}.tar")
|
|
with SecureTarFile(temp_tar, "w"):
|
|
pass
|
|
backup = Backup(coresys, temp_tar)
|
|
backup._data = { # pylint: disable=protected-access
|
|
ATTR_SLUG: slug,
|
|
ATTR_DATE: utcnow().isoformat(),
|
|
ATTR_TYPE: BackupType.PARTIAL
|
|
if "1" == slug[-1] or "5" == slug[-1]
|
|
else BackupType.FULL,
|
|
}
|
|
coresys.backups._backups[backup.slug] = backup
|
|
|
|
yield coresys.backups.list_backups
|
|
|
|
|
|
@pytest.fixture
|
|
async def journald_logs(coresys: CoreSys) -> MagicMock:
|
|
"""Mock journald logs and make it available."""
|
|
with patch.object(
|
|
LogsControl, "available", new=PropertyMock(return_value=True)
|
|
), patch.object(
|
|
LogsControl, "get_boot_ids", return_value=["aaa", "bbb", "ccc"]
|
|
), patch.object(
|
|
LogsControl,
|
|
"get_identifiers",
|
|
return_value=["hassio_supervisor", "hassos-config", "kernel"],
|
|
), patch.object(
|
|
LogsControl, "journald_logs", new=MagicMock()
|
|
) as logs:
|
|
await coresys.host.logs.load()
|
|
yield logs
|
|
|
|
|
|
@pytest.fixture
|
|
async def docker_logs(docker: DockerAPI, supervisor_name) -> MagicMock:
|
|
"""Mock log output for a container from docker."""
|
|
container_mock = MagicMock()
|
|
container_mock.logs.return_value = load_binary_fixture("logs_docker_container.txt")
|
|
docker.containers.get.return_value = container_mock
|
|
yield container_mock.logs
|
|
|
|
|
|
@pytest.fixture
|
|
async def capture_exception() -> Mock:
|
|
"""Mock capture exception method for testing."""
|
|
with patch("supervisor.utils.sentry.sentry_connected", return_value=True), patch(
|
|
"supervisor.utils.sentry.sentry_sdk.capture_exception"
|
|
) as capture_exception:
|
|
yield capture_exception
|
|
|
|
|
|
@pytest.fixture
|
|
async def capture_event() -> Mock:
|
|
"""Mock capture event for testing."""
|
|
with patch("supervisor.utils.sentry.sentry_connected", return_value=True), patch(
|
|
"supervisor.utils.sentry.sentry_sdk.capture_event"
|
|
) as capture_event:
|
|
yield capture_event
|
|
|
|
|
|
@pytest.fixture
|
|
async def os_available(request: pytest.FixtureRequest) -> None:
|
|
"""Mock os as available."""
|
|
version = (
|
|
AwesomeVersion(request.param)
|
|
if hasattr(request, "param")
|
|
else AwesomeVersion("10.2")
|
|
)
|
|
with patch.object(
|
|
OSManager, "available", new=PropertyMock(return_value=True)
|
|
), patch.object(OSManager, "version", new=PropertyMock(return_value=version)):
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def mount_propagation(docker: DockerAPI, coresys: CoreSys) -> None:
|
|
"""Mock supervisor connected to container with propagation set."""
|
|
docker.containers.get.return_value = supervisor = MagicMock()
|
|
supervisor.attrs = {
|
|
"Mounts": [
|
|
{
|
|
"Type": "bind",
|
|
"Source": "/mnt/data/supervisor",
|
|
"Destination": "/data",
|
|
"Mode": "rw",
|
|
"RW": True,
|
|
"Propagation": "slave",
|
|
}
|
|
]
|
|
}
|
|
await coresys.supervisor.load()
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
async def container(docker: DockerAPI) -> MagicMock:
|
|
"""Mock attrs and status for container on attach."""
|
|
docker.containers.get.return_value = addon = MagicMock()
|
|
docker.containers.create.return_value = addon
|
|
docker.images.pull.return_value = addon
|
|
docker.images.build.return_value = (addon, "")
|
|
addon.status = "stopped"
|
|
addon.attrs = {"State": {"ExitCode": 0}}
|
|
yield addon
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_amd64_arch_supported(coresys: CoreSys) -> None:
|
|
"""Mock amd64 arch as supported."""
|
|
with patch.object(coresys.arch, "_supported_set", {"amd64"}):
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_aarch64_arch_supported(coresys: CoreSys) -> None:
|
|
"""Mock aarch64 arch as supported."""
|
|
with patch.object(coresys.arch, "_supported_set", {"aarch64"}):
|
|
yield
|