supervisor/tests/conftest.py
Pascal Vizeli 40aa5c9caf
Write core state to /run/supervisor & monitor (#2027)
* Write core state to /run/supervisor & monitor

* Add watchdog

* Add log if they getting started

* fix stale comment

* Fix script

* avoid output

* fix tests

* fix lint

* Update rootfs/etc/services.d/watchdog/run

Co-authored-by: Franck Nijhof <git@frenck.dev>

Co-authored-by: Franck Nijhof <git@frenck.dev>
2020-09-08 10:09:41 +02:00

141 lines
4.1 KiB
Python

"""Common test functions."""
from unittest.mock import MagicMock, PropertyMock, patch
from uuid import uuid4
from aiohttp import web
from aiohttp.test_utils import TestClient
import pytest
from supervisor.api import RestAPI
from supervisor.bootstrap import initialize_coresys
from supervisor.coresys import CoreSys
from supervisor.dbus.const import DBUS_NAME_NM, DBUS_OBJECT_BASE
from supervisor.dbus.network import NetworkManager
from supervisor.dbus.network.interface import NetworkInterface
from supervisor.docker import DockerAPI
from supervisor.utils.gdbus import DBus
from tests.common import load_fixture, load_json_fixture
# pylint: disable=redefined-outer-name, protected-access
@pytest.fixture
def docker() -> DockerAPI:
"""Mock DockerAPI."""
images = [MagicMock(tags=["homeassistant/amd64-hassio-supervisor:latest"])]
with patch("docker.DockerClient", return_value=MagicMock()), patch(
"supervisor.docker.DockerAPI.images", return_value=MagicMock()
), patch("supervisor.docker.DockerAPI.containers", return_value=MagicMock()), patch(
"supervisor.docker.DockerAPI.api", return_value=MagicMock()
), patch(
"supervisor.docker.DockerAPI.images.list", return_value=images
):
docker_obj = DockerAPI()
yield docker_obj
@pytest.fixture
def dbus() -> DBus:
"""Mock DBUS."""
async def mock_get_properties(_, interface):
return load_json_fixture(f"{interface.replace('.', '_')}.json")
async def mock_send(_, command):
filetype = "xml" if "--xml" in command else "fixture"
fixture = f"{command[6].replace('/', '_')[1:]}.{filetype}"
return load_fixture(fixture)
with patch("supervisor.utils.gdbus.DBus._send", new=mock_send), patch(
"supervisor.dbus.interface.DBusInterface.is_connected",
return_value=True,
), patch("supervisor.utils.gdbus.DBus.get_properties", new=mock_get_properties):
dbus_obj = DBus(DBUS_NAME_NM, DBUS_OBJECT_BASE)
yield dbus_obj
@pytest.fixture
async def network_manager(dbus) -> NetworkManager:
"""Mock NetworkManager."""
async def dns_update():
pass
with patch("supervisor.dbus.network.NetworkManager.dns", return_value=MagicMock()):
nm_obj = NetworkManager()
nm_obj.dns.update = dns_update
nm_obj.dbus = dbus
await nm_obj.connect()
await nm_obj.update()
yield nm_obj
@pytest.fixture
async def coresys(loop, docker, dbus, network_manager, aiohttp_client) -> CoreSys:
"""Create a CoreSys Mock."""
with patch("supervisor.bootstrap.initialize_system_data"), patch(
"supervisor.bootstrap.setup_diagnostics"
), patch(
"supervisor.bootstrap.fetch_timezone",
return_value="Europe/Zurich",
), patch(
"aiohttp.ClientSession",
return_value=TestClient.session,
):
coresys_obj = await initialize_coresys()
# Mock save json
coresys_obj.ingress.save_data = MagicMock()
# Mock test client
coresys_obj.arch._default_arch = "amd64"
coresys_obj._machine = "qemux86-64"
coresys_obj._machine_id = uuid4()
# Mock host communication
coresys_obj._dbus = dbus
coresys_obj._dbus.network = network_manager
yield coresys_obj
@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):
"""Fixture for RestAPI client."""
api = RestAPI(coresys)
api.webapp = web.Application()
await api.load()
yield await aiohttp_client(api.webapp)
@pytest.fixture
async def network_interface(dbus):
"""Fixture for a network interface."""
interface = NetworkInterface()
await interface.connect(dbus, "/org/freedesktop/NetworkManager/ActiveConnection/1")
await interface.connection.update_information()
yield interface