Dbus refactor and tests (#3854)

* Dbus refactor and tests

* Link to timedate introspection
This commit is contained in:
Mike Degatano 2022-09-12 11:59:56 -04:00 committed by GitHub
parent 611128c014
commit 4f272ad4fd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
52 changed files with 920 additions and 321 deletions

View File

@ -120,25 +120,25 @@ class APIHost(CoreSysAttributes):
def service_start(self, request):
"""Start a service."""
unit = request.match_info.get(SERVICE)
return asyncio.shield(self.sys_host.services.start(unit))
return [asyncio.shield(self.sys_host.services.start(unit))]
@api_process
def service_stop(self, request):
"""Stop a service."""
unit = request.match_info.get(SERVICE)
return asyncio.shield(self.sys_host.services.stop(unit))
return [asyncio.shield(self.sys_host.services.stop(unit))]
@api_process
def service_reload(self, request):
"""Reload a service."""
unit = request.match_info.get(SERVICE)
return asyncio.shield(self.sys_host.services.reload(unit))
return [asyncio.shield(self.sys_host.services.reload(unit))]
@api_process
def service_restart(self, request):
"""Restart a service."""
unit = request.match_info.get(SERVICE)
return asyncio.shield(self.sys_host.services.restart(unit))
return [asyncio.shield(self.sys_host.services.restart(unit))]
@api_process_raw(CONTENT_TYPE_BINARY)
def logs(self, request: web.Request) -> Awaitable[bytes]:

View File

@ -22,7 +22,10 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
class Hostname(DBusInterface):
"""Handle D-Bus interface for hostname/system."""
"""Handle D-Bus interface for hostname/system.
https://www.freedesktop.org/software/systemd/man/org.freedesktop.hostname1.html
"""
name = DBUS_NAME_HOSTNAME
@ -78,12 +81,9 @@ class Hostname(DBusInterface):
return self.properties[DBUS_ATTR_STATIC_OPERATING_SYSTEM_CPE_NAME]
@dbus_connected
def set_static_hostname(self, hostname: str):
"""Change local hostname.
Return a coroutine.
"""
return self.dbus.SetStaticHostname(hostname, False)
async def set_static_hostname(self, hostname: str) -> None:
"""Change local hostname."""
await self.dbus.SetStaticHostname(hostname, False)
@dbus_connected
async def update(self):

View File

@ -11,7 +11,10 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
class Logind(DBusInterface):
"""Logind function handler."""
"""Logind function handler.
https://www.freedesktop.org/software/systemd/man/org.freedesktop.login1.html
"""
name = DBUS_NAME_LOGIND
@ -25,17 +28,11 @@ class Logind(DBusInterface):
_LOGGER.info("No systemd-logind support on the host.")
@dbus_connected
def reboot(self):
"""Reboot host computer.
Return a coroutine.
"""
return self.dbus.Manager.Reboot(False)
async def reboot(self) -> None:
"""Reboot host computer."""
await self.dbus.Manager.Reboot(False)
@dbus_connected
def power_off(self):
"""Power off host computer.
Return a coroutine.
"""
return self.dbus.Manager.PowerOff(False)
async def power_off(self) -> None:
"""Power off host computer."""
await self.dbus.Manager.PowerOff(False)

View File

@ -10,7 +10,7 @@ from ..const import (
DBUS_IFACE_ACCESSPOINT,
DBUS_NAME_NM,
)
from ..interface import DBusInterfaceProxy
from ..interface import DBusInterfaceProxy, dbus_property
class NetworkWirelessAP(DBusInterfaceProxy):
@ -25,26 +25,31 @@ class NetworkWirelessAP(DBusInterfaceProxy):
self.properties = {}
@property
@dbus_property
def ssid(self) -> str:
"""Return details about ssid."""
return bytes(self.properties[DBUS_ATTR_SSID]).decode()
@property
@dbus_property
def frequency(self) -> int:
"""Return details about frequency."""
return self.properties[DBUS_ATTR_FREQUENCY]
@property
@dbus_property
def mac(self) -> str:
"""Return details about mac address."""
return self.properties[DBUS_ATTR_HWADDRESS]
@property
@dbus_property
def mode(self) -> int:
"""Return details about mac address."""
return self.properties[DBUS_ATTR_MODE]
@property
@dbus_property
def strength(self) -> int:
"""Return details about mac address."""
return int(self.properties[DBUS_ATTR_STRENGTH])

View File

@ -24,7 +24,7 @@ from ..const import (
ConnectionStateFlags,
ConnectionStateType,
)
from ..interface import DBusInterfaceProxy
from ..interface import DBusInterfaceProxy, dbus_property
from ..utils import dbus_connected
from .configuration import IpConfiguration
@ -45,21 +45,25 @@ class NetworkConnection(DBusInterfaceProxy):
self._state_flags: set[ConnectionStateFlags] = {ConnectionStateFlags.NONE}
@property
@dbus_property
def id(self) -> str:
"""Return the id of the connection."""
return self.properties[DBUS_ATTR_ID]
@property
@dbus_property
def type(self) -> str:
"""Return the type of the connection."""
return self.properties[DBUS_ATTR_TYPE]
@property
@dbus_property
def uuid(self) -> str:
"""Return the uuid of the connection."""
return self.properties[DBUS_ATTR_UUID]
@property
@dbus_property
def state(self) -> ConnectionStateType:
"""Return the state of the connection."""
return self.properties[DBUS_ATTR_STATE]
@ -70,7 +74,8 @@ class NetworkConnection(DBusInterfaceProxy):
return self._state_flags
@property
def setting_object(self) -> int:
@dbus_property
def setting_object(self) -> str:
"""Return the connection object path."""
return self.properties[DBUS_ATTR_CONNECTION]

View File

@ -1,4 +1,4 @@
"""D-Bus interface for hostname."""
"""Network Manager DNS Manager object."""
from ipaddress import ip_address
import logging

View File

@ -11,7 +11,7 @@ from ..const import (
DBUS_OBJECT_BASE,
DeviceType,
)
from ..interface import DBusInterfaceProxy
from ..interface import DBusInterfaceProxy, dbus_property
from .connection import NetworkConnection
from .setting import NetworkSetting
from .wireless import NetworkWireless
@ -36,21 +36,25 @@ class NetworkInterface(DBusInterfaceProxy):
self._nm_dbus: DBus = nm_dbus
@property
@dbus_property
def name(self) -> str:
"""Return interface name."""
return self.properties[DBUS_ATTR_DEVICE_INTERFACE]
@property
@dbus_property
def type(self) -> int:
"""Return interface type."""
return self.properties[DBUS_ATTR_DEVICE_TYPE]
@property
@dbus_property
def driver(self) -> str:
"""Return interface driver."""
return self.properties[DBUS_ATTR_DRIVER]
@property
@dbus_property
def managed(self) -> bool:
"""Return interface driver."""
return self.properties[DBUS_ATTR_MANAGED]

View File

@ -1,6 +1,6 @@
"""Connection object for Network Manager."""
import logging
from typing import Any, Awaitable
from typing import Any
from ....const import ATTR_METHOD, ATTR_MODE, ATTR_PSK, ATTR_SSID
from ....utils.dbus import DBus
@ -120,9 +120,9 @@ class NetworkSetting(DBusInterfaceProxy):
return self._ipv6
@dbus_connected
def get_settings(self) -> Awaitable[Any]:
async def get_settings(self) -> dict[str, Any]:
"""Return connection settings."""
return self.dbus.Settings.Connection.GetSettings()
return (await self.dbus.Settings.Connection.GetSettings())[0]
@dbus_connected
async def update(self, settings: Any) -> None:
@ -154,14 +154,14 @@ class NetworkSetting(DBusInterfaceProxy):
return await self.dbus.Settings.Connection.Update(("a{sa{sv}}", new_settings))
@dbus_connected
def delete(self) -> Awaitable[None]:
async def delete(self) -> None:
"""Delete connection settings."""
return self.dbus.Settings.Connection.Delete()
await self.dbus.Settings.Connection.Delete()
async def connect(self) -> None:
"""Get connection information."""
self.dbus = await DBus.connect(DBUS_NAME_NM, self.object_path)
data = (await self.get_settings())[0]
data = await self.get_settings()
# Get configuration settings we care about
# See: https://developer-old.gnome.org/NetworkManager/stable/ch01.html

View File

@ -1,6 +1,8 @@
"""Network Manager implementation for DBUS."""
import logging
from typing import Any, Awaitable
from typing import Any
from supervisor.dbus.network.setting import NetworkSetting
from ...exceptions import DBusError, DBusInterfaceError
from ...utils.dbus import DBus
@ -29,11 +31,16 @@ class NetworkManagerSettings(DBusInterface):
)
@dbus_connected
def add_connection(self, settings: Any) -> Awaitable[Any]:
async def add_connection(self, settings: Any) -> NetworkSetting:
"""Add new connection."""
return self.dbus.Settings.AddConnection(("a{sa{sv}}", settings))
obj_con_setting = (
await self.dbus.Settings.AddConnection(("a{sa{sv}}", settings))
)[0]
con_setting = NetworkSetting(obj_con_setting)
await con_setting.connect()
return con_setting
@dbus_connected
def reload_connections(self) -> Awaitable[Any]:
async def reload_connections(self) -> bool:
"""Reload all local connection files."""
return self.dbus.Settings.ReloadConnections()
return (await self.dbus.Settings.ReloadConnections())[0]

View File

@ -1,5 +1,6 @@
"""Connection object for Network Manager."""
from typing import Any, Awaitable
"""Wireless object for Network Manager."""
import asyncio
import logging
from ...utils.dbus import DBus
from ..const import (
@ -12,6 +13,8 @@ from ..interface import DBusInterfaceProxy
from ..utils import dbus_connected
from .accesspoint import NetworkWirelessAP
_LOGGER: logging.Logger = logging.getLogger(__name__)
class NetworkWireless(DBusInterfaceProxy):
"""Wireless object for Network Manager.
@ -32,14 +35,23 @@ class NetworkWireless(DBusInterfaceProxy):
return self._active
@dbus_connected
def request_scan(self) -> Awaitable[None]:
async def request_scan(self) -> None:
"""Request a new AP scan."""
return self.dbus.Device.Wireless.RequestScan(("a{sv}", {}))
await self.dbus.Device.Wireless.RequestScan(("a{sv}", {}))
@dbus_connected
def get_all_accesspoints(self) -> Awaitable[Any]:
async def get_all_accesspoints(self) -> list[NetworkWirelessAP]:
"""Return a list of all access points path."""
return self.dbus.Device.Wireless.GetAllAccessPoints()
accesspoints_data = (await self.dbus.Device.Wireless.GetAllAccessPoints())[0]
accesspoints = [NetworkWirelessAP(ap_obj) for ap_obj in accesspoints_data]
for err in await asyncio.gather(
*[ap.connect() for ap in accesspoints], return_exceptions=True
):
if err:
_LOGGER.warning("Can't process an AP: %s", err)
return accesspoints
async def connect(self) -> None:
"""Get connection information."""

View File

@ -1,8 +1,9 @@
"""D-Bus interface for rauc."""
import logging
from typing import Any
from ..exceptions import DBusError, DBusInterfaceError
from ..utils.dbus import DBus
from ..utils.dbus import DBus, DBusSignalWrapper
from .const import (
DBUS_ATTR_BOOT_SLOT,
DBUS_ATTR_COMPATIBLE,
@ -69,36 +70,24 @@ class Rauc(DBusInterface):
return self._boot_slot
@dbus_connected
def install(self, raucb_file):
"""Install rauc bundle file.
Return a coroutine.
"""
return self.dbus.Installer.Install(str(raucb_file))
async def install(self, raucb_file) -> None:
"""Install rauc bundle file."""
await self.dbus.Installer.Install(str(raucb_file))
@dbus_connected
def get_slot_status(self):
"""Get slot status.
Return a coroutine.
"""
return self.dbus.Installer.GetSlotStatus()
async def get_slot_status(self) -> list[tuple[str, dict[str, Any]]]:
"""Get slot status."""
return (await self.dbus.Installer.GetSlotStatus())[0]
@dbus_connected
def signal_completed(self):
"""Return a signal wrapper for completed signal.
Return a coroutine.
"""
return self.dbus.wait_signal(DBUS_SIGNAL_RAUC_INSTALLER_COMPLETED)
def signal_completed(self) -> DBusSignalWrapper:
"""Return a signal wrapper for completed signal."""
return self.dbus.signal(DBUS_SIGNAL_RAUC_INSTALLER_COMPLETED)
@dbus_connected
def mark(self, state: RaucState, slot_identifier: str):
"""Get slot status.
Return a coroutine.
"""
return self.dbus.Installer.Mark(state, slot_identifier)
async def mark(self, state: RaucState, slot_identifier: str) -> tuple[str, str]:
"""Get slot status."""
return await self.dbus.Installer.Mark(state, slot_identifier)
@dbus_connected
async def update(self):

View File

@ -43,7 +43,10 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
class Resolved(DBusInterface):
"""Handle D-Bus interface for systemd-resolved."""
"""Handle D-Bus interface for systemd-resolved.
https://www.freedesktop.org/software/systemd/man/org.freedesktop.resolve1.html
"""
name = DBUS_NAME_RESOLVED

View File

@ -21,7 +21,10 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
class Systemd(DBusInterface):
"""Systemd function handler."""
"""Systemd function handler.
https://www.freedesktop.org/software/systemd/man/org.freedesktop.systemd1.html
"""
name = DBUS_NAME_SYSTEMD
@ -58,60 +61,41 @@ class Systemd(DBusInterface):
return self.properties[DBUS_ATTR_FINISH_TIMESTAMP]
@dbus_connected
def reboot(self):
"""Reboot host computer.
Return a coroutine.
"""
return self.dbus.Manager.Reboot()
async def reboot(self) -> None:
"""Reboot host computer."""
await self.dbus.Manager.Reboot()
@dbus_connected
def power_off(self):
"""Power off host computer.
Return a coroutine.
"""
return self.dbus.Manager.PowerOff()
async def power_off(self) -> None:
"""Power off host computer."""
await self.dbus.Manager.PowerOff()
@dbus_connected
def start_unit(self, unit, mode):
"""Start a systemd service unit.
Return a coroutine.
"""
return self.dbus.Manager.StartUnit(unit, mode)
async def start_unit(self, unit, mode) -> str:
"""Start a systemd service unit. Return job object path."""
return (await self.dbus.Manager.StartUnit(unit, mode))[0]
@dbus_connected
def stop_unit(self, unit, mode):
"""Stop a systemd service unit.
Return a coroutine.
"""
return self.dbus.Manager.StopUnit(unit, mode)
async def stop_unit(self, unit, mode) -> str:
"""Stop a systemd service unit."""
return (await self.dbus.Manager.StopUnit(unit, mode))[0]
@dbus_connected
def reload_unit(self, unit, mode):
"""Reload a systemd service unit.
Return a coroutine.
"""
return self.dbus.Manager.ReloadOrRestartUnit(unit, mode)
async def reload_unit(self, unit, mode) -> str:
"""Reload a systemd service unit."""
return (await self.dbus.Manager.ReloadOrRestartUnit(unit, mode))[0]
@dbus_connected
def restart_unit(self, unit, mode):
"""Restart a systemd service unit.
Return a coroutine.
"""
return self.dbus.Manager.RestartUnit(unit, mode)
async def restart_unit(self, unit, mode):
"""Restart a systemd service unit."""
return (await self.dbus.Manager.RestartUnit(unit, mode))[0]
@dbus_connected
def list_units(self):
"""Return a list of available systemd services.
Return a coroutine.
"""
return self.dbus.Manager.ListUnits()
async def list_units(
self,
) -> list[str, str, str, str, str, str, str, int, str, str]:
"""Return a list of available systemd services."""
return (await self.dbus.Manager.ListUnits())[0]
@dbus_connected
async def update(self):

View File

@ -22,7 +22,10 @@ _LOGGER: logging.Logger = logging.getLogger(__name__)
class TimeDate(DBusInterface):
"""Timedate function handler."""
"""Timedate function handler.
https://www.freedesktop.org/software/systemd/man/org.freedesktop.timedate1.html
"""
name = DBUS_NAME_TIMEDATE
@ -66,20 +69,14 @@ class TimeDate(DBusInterface):
)
@dbus_connected
def set_time(self, utc: datetime):
"""Set time & date on host as UTC.
Return a coroutine.
"""
return self.dbus.SetTime(int(utc.timestamp() * 1000000), False, False)
async def set_time(self, utc: datetime) -> None:
"""Set time & date on host as UTC."""
await self.dbus.SetTime(int(utc.timestamp() * 1000000), False, False)
@dbus_connected
def set_ntp(self, use_ntp: bool):
"""Turn NTP on or off.
Return a coroutine.
"""
return self.dbus.SetNTP(use_ntp)
async def set_ntp(self, use_ntp: bool) -> None:
"""Turn NTP on or off."""
await self.dbus.SetNTP(use_ntp, False)
@dbus_connected
async def update(self):

View File

@ -19,7 +19,6 @@ from ..dbus.const import (
InterfaceMethod as NMInterfaceMethod,
WirelessMethodType,
)
from ..dbus.network.accesspoint import NetworkWirelessAP
from ..dbus.network.connection import NetworkConnection
from ..dbus.network.interface import NetworkInterface
from ..dbus.network.setting.generate import get_connection_from_interface
@ -270,27 +269,17 @@ class NetworkManager(CoreSysAttributes):
await asyncio.sleep(5)
# Process AP
accesspoints: list[AccessPoint] = []
for ap_object in (await inet.wireless.get_all_accesspoints())[0]:
accesspoint = NetworkWirelessAP(ap_object)
try:
await accesspoint.connect()
except DBusError as err:
_LOGGER.warning("Can't process an AP: %s", err)
continue
else:
accesspoints.append(
AccessPoint(
WifiMode[WirelessMethodType(accesspoint.mode).name],
accesspoint.ssid,
accesspoint.mac,
accesspoint.frequency,
accesspoint.strength,
)
)
return accesspoints
return [
AccessPoint(
WifiMode[WirelessMethodType(accesspoint.mode).name],
accesspoint.ssid,
accesspoint.mac,
accesspoint.frequency,
accesspoint.strength,
)
for accesspoint in await inet.wireless.get_all_accesspoints()
if accesspoint.dbus
]
@attr.s(slots=True)

View File

@ -1,5 +1,6 @@
"""Service control for host."""
import logging
from typing import Awaitable
import attr
@ -33,35 +34,47 @@ class ServiceManager(CoreSysAttributes):
if unit and not self.exists(unit):
raise HostServiceError(f"Unit '{unit}' not found", _LOGGER.error)
def start(self, unit):
"""Start a service on host."""
def start(self, unit) -> Awaitable[str]:
"""Start a service on host.
Returns a coroutine.
"""
self._check_dbus(unit)
_LOGGER.info("Starting local service %s", unit)
return self.sys_dbus.systemd.start_unit(unit, MOD_REPLACE)
def stop(self, unit):
"""Stop a service on host."""
def stop(self, unit) -> Awaitable[str]:
"""Stop a service on host.
Returns a coroutine.
"""
self._check_dbus(unit)
_LOGGER.info("Stopping local service %s", unit)
return self.sys_dbus.systemd.stop_unit(unit, MOD_REPLACE)
def reload(self, unit):
"""Reload a service on host."""
def reload(self, unit) -> Awaitable[str]:
"""Reload a service on host.
Returns a coroutine.
"""
self._check_dbus(unit)
_LOGGER.info("Reloading local service %s", unit)
return self.sys_dbus.systemd.reload_unit(unit, MOD_REPLACE)
def restart(self, unit):
"""Restart a service on host."""
def restart(self, unit) -> Awaitable[str]:
"""Restart a service on host.
Returns a coroutine.
"""
self._check_dbus(unit)
_LOGGER.info("Restarting local service %s", unit)
return self.sys_dbus.systemd.restart_unit(unit, MOD_REPLACE)
def exists(self, unit):
def exists(self, unit) -> bool:
"""Check if a unit exists and return True."""
for service in self._services:
if unit == service.name:
@ -76,7 +89,7 @@ class ServiceManager(CoreSysAttributes):
self._services.clear()
try:
systemd_units = await self.sys_dbus.systemd.list_units()
for service_data in systemd_units[0]:
for service_data in systemd_units:
if (
not service_data[0].endswith(".service")
or service_data[2] != "loaded"

View File

@ -195,8 +195,12 @@ class OSManager(CoreSysAttributes):
ext_ota = Path(self.sys_config.path_extern_tmp, int_ota.name)
try:
await self.sys_dbus.rauc.install(ext_ota)
completed = await self.sys_dbus.rauc.signal_completed()
async with self.sys_dbus.rauc.signal_completed() as signal:
# Start listening for signals before triggering install
# This prevents a race condition with install complete signal
await self.sys_dbus.rauc.install(ext_ota)
completed = await signal.wait_for_signal()
except DBusError as err:
raise HassOSUpdateError("Rauc communication error", _LOGGER.error) from err

View File

@ -204,15 +204,10 @@ class DBus:
"""Set a property from interface."""
return await self.call_dbus(DBUS_METHOD_SET, interface, name, value)
def signal(self, signal_member) -> DBusSignalWrapper:
def signal(self, signal_member: str) -> DBusSignalWrapper:
"""Get signal context manager for this object."""
return DBusSignalWrapper(self, signal_member)
async def wait_signal(self, signal_member) -> Any:
"""Wait for signal on this object."""
async with self.signal(signal_member) as signal:
return await signal.wait_for_signal()
def __getattr__(self, name: str) -> DBusCallWrapper:
"""Map to dbus method."""
return getattr(DBusCallWrapper(self, self.bus_name), name)
@ -299,7 +294,7 @@ class DBusSignalWrapper:
self._dbus._bus.add_message_handler(self._message_handler)
return self
async def wait_for_signal(self) -> Message:
async def wait_for_signal(self) -> Any:
"""Wait for signal and returns signal payload."""
msg = await self._messages.get()
return msg.body

View File

@ -37,7 +37,11 @@ from supervisor.const import (
)
from supervisor.coresys import CoreSys
from supervisor.dbus.agent import OSAgent
from supervisor.dbus.const import DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED
from supervisor.dbus.const import (
DBUS_OBJECT_BASE,
DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED,
DBUS_SIGNAL_RAUC_INSTALLER_COMPLETED,
)
from supervisor.dbus.hostname import Hostname
from supervisor.dbus.interface import DBusInterface
from supervisor.dbus.network import NetworkManager
@ -113,16 +117,20 @@ def dbus() -> DBus:
return load_json_fixture(f"{fixture}.json")
async def mock_get_property(dbus_obj, interface, name):
dbus_commands.append(f"{dbus_obj.object_path}-{interface}.{name}")
properties = await mock_get_properties(dbus_obj, interface)
return properties[name]
async def mock_wait_for_signal(self):
if (
self._interface + "." + self._method
self._interface + "." + self._member
== DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED
):
return [2, 0]
if self._interface + "." + self._member == DBUS_SIGNAL_RAUC_INSTALLER_COMPLETED:
return [0]
async def mock_signal___aenter__(self):
return self
@ -132,7 +140,12 @@ def dbus() -> DBus:
async def mock_init_proxy(self):
filetype = "xml"
fixture = self.object_path.replace("/", "_")[1:]
fixture = (
self.object_path.replace("/", "_")[1:]
if self.object_path != DBUS_OBJECT_BASE
else self.bus_name.replace(".", "_")
)
if not exists_fixture(f"{fixture}.{filetype}"):
fixture = re.sub(r"_[0-9]+$", "", fixture)
@ -147,12 +160,18 @@ def dbus() -> DBus:
async def mock_call_dbus(
self, method: str, *args: list[Any], remove_signature: bool = True
):
if self.object_path != DBUS_OBJECT_BASE:
fixture = self.object_path.replace("/", "_")[1:]
fixture = f"{fixture}-{method.split('.')[-1]}"
else:
fixture = method.replace(".", "_")
fixture = self.object_path.replace("/", "_")[1:]
fixture = f"{fixture}-{method.split('.')[-1]}"
dbus_commands.append(fixture)
dbus_commands.append(f"{self.object_path}-{method}")
return load_json_fixture(f"{fixture}.json")
if exists_fixture(f"{fixture}.json"):
return load_json_fixture(f"{fixture}.json")
return []
with patch("supervisor.utils.dbus.DBus.call_dbus", new=mock_call_dbus), patch(
"supervisor.dbus.interface.DBusInterface.is_connected",

View File

@ -17,9 +17,8 @@ async def test_dbus_osagent_apparmor(coresys: CoreSys):
assert coresys.dbus.agent.apparmor.version == "2.13.2"
async def test_dbus_osagent_apparmor_load(coresys: CoreSys):
async def test_dbus_osagent_apparmor_load(coresys: CoreSys, dbus: list[str]):
"""Load AppArmor Profile on host."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.agent.apparmor.load_profile(
Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
@ -27,17 +26,18 @@ async def test_dbus_osagent_apparmor_load(coresys: CoreSys):
await coresys.dbus.agent.connect()
dbus.clear()
assert (
await coresys.dbus.agent.apparmor.load_profile(
Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
)
is None
)
assert dbus == ["/io/hass/os/AppArmor-io.hass.os.AppArmor.LoadProfile"]
async def test_dbus_osagent_apparmor_unload(coresys: CoreSys):
async def test_dbus_osagent_apparmor_unload(coresys: CoreSys, dbus: list[str]):
"""Unload AppArmor Profile on host."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.agent.apparmor.unload_profile(
Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
@ -45,9 +45,11 @@ async def test_dbus_osagent_apparmor_unload(coresys: CoreSys):
await coresys.dbus.agent.connect()
dbus.clear()
assert (
await coresys.dbus.agent.apparmor.unload_profile(
Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
)
is None
)
assert dbus == ["/io/hass/os/AppArmor-io.hass.os.AppArmor.UnloadProfile"]

View File

@ -6,7 +6,7 @@ from supervisor.coresys import CoreSys
from supervisor.exceptions import DBusNotConnectedError
async def test_dbus_osagent_cgroup_add_devices(coresys: CoreSys):
async def test_dbus_osagent_cgroup_add_devices(coresys: CoreSys, dbus: list[str]):
"""Test wipe data partition on host."""
with pytest.raises(DBusNotConnectedError):
@ -14,7 +14,9 @@ async def test_dbus_osagent_cgroup_add_devices(coresys: CoreSys):
await coresys.dbus.agent.connect()
dbus.clear()
assert (
await coresys.dbus.agent.cgroup.add_devices_allowed("9324kl23j4kl", "*:* rwm")
is None
)
assert dbus == ["/io/hass/os/CGroup-io.hass.os.CGroup.AddDevicesAllowed"]

View File

@ -17,23 +17,25 @@ async def test_dbus_osagent_datadisk(coresys: CoreSys):
assert coresys.dbus.agent.datadisk.current_device.as_posix() == "/dev/sda"
async def test_dbus_osagent_datadisk_change_device(coresys: CoreSys):
async def test_dbus_osagent_datadisk_change_device(coresys: CoreSys, dbus: list[str]):
"""Change datadisk on device."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.agent.datadisk.change_device(Path("/dev/sdb"))
await coresys.dbus.agent.connect()
dbus.clear()
assert await coresys.dbus.agent.datadisk.change_device(Path("/dev/sdb")) is None
assert dbus == ["/io/hass/os/DataDisk-io.hass.os.DataDisk.ChangeDevice"]
async def test_dbus_osagent_datadisk_reload_device(coresys: CoreSys):
async def test_dbus_osagent_datadisk_reload_device(coresys: CoreSys, dbus: list[str]):
"""Change datadisk on device."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.agent.datadisk.reload_device()
await coresys.dbus.agent.connect()
dbus.clear()
assert await coresys.dbus.agent.datadisk.reload_device() is None
assert dbus == ["/io/hass/os/DataDisk-io.hass.os.DataDisk.ReloadDevice"]

View File

@ -6,12 +6,13 @@ from supervisor.coresys import CoreSys
from supervisor.exceptions import DBusNotConnectedError
async def test_dbus_osagent_system_wipe(coresys: CoreSys):
async def test_dbus_osagent_system_wipe(coresys: CoreSys, dbus: list[str]):
"""Test wipe data partition on host."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.agent.system.schedule_wipe_device()
await coresys.dbus.agent.connect()
dbus.clear()
assert await coresys.dbus.agent.system.schedule_wipe_device() is None
assert dbus == ["/io/hass/os/System-io.hass.os.System.ScheduleWipeDevice"]

View File

@ -11,6 +11,61 @@ from supervisor.host.network import Interface
from tests.const import TEST_INTERFACE
SETTINGS_WITH_SIGNATURE = {
"connection": {
"id": Variant("s", "Wired connection 1"),
"interface-name": Variant("s", "eth0"),
"permissions": Variant("as", []),
"timestamp": Variant("t", 1598125548),
"type": Variant("s", "802-3-ethernet"),
"uuid": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"),
},
"ipv4": {
"address-data": Variant(
"aa{sv}",
[
{
"address": Variant("s", "192.168.2.148"),
"prefix": Variant("u", 24),
}
],
),
"addresses": Variant("aau", [[2483202240, 24, 16951488]]),
"dns": Variant("au", [16951488]),
"dns-search": Variant("as", []),
"gateway": Variant("s", "192.168.2.1"),
"method": Variant("s", "auto"),
"route-data": Variant(
"aa{sv}",
[
{
"dest": Variant("s", "192.168.122.0"),
"prefix": Variant("u", 24),
"next-hop": Variant("s", "10.10.10.1"),
}
],
),
"routes": Variant("aau", [[8038592, 24, 17435146, 0]]),
},
"ipv6": {
"address-data": Variant("aa{sv}", []),
"addresses": Variant("a(ayuay)", []),
"dns": Variant("au", []),
"dns-search": Variant("as", []),
"method": Variant("s", "auto"),
"route-data": Variant("aa{sv}", []),
"routes": Variant("aau", []),
"addr-gen-mode": Variant("i", 0),
},
"proxy": {},
"802-3-ethernet": {
"auto-negotiate": Variant("b", False),
"mac-address-blacklist": Variant("as", []),
"s390-options": Variant("a{ss}", {}),
},
"802-11-wireless": {"ssid": Variant("ay", bytes([78, 69, 84, 84]))},
}
async def mock_call_dbus_get_settings_signature(
method: str, *args: list[Any], remove_signature: bool = True
@ -20,62 +75,7 @@ async def mock_call_dbus_get_settings_signature(
method == "org.freedesktop.NetworkManager.Settings.Connection.GetSettings"
and not remove_signature
):
return [
{
"connection": {
"id": Variant("s", "Wired connection 1"),
"interface-name": Variant("s", "eth0"),
"permissions": Variant("as", []),
"timestamp": Variant("t", 1598125548),
"type": Variant("s", "802-3-ethernet"),
"uuid": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"),
},
"ipv4": {
"address-data": Variant(
"aa{sv}",
[
{
"address": Variant("s", "192.168.2.148"),
"prefix": Variant("u", 24),
}
],
),
"addresses": Variant("aau", [[2483202240, 24, 16951488]]),
"dns": Variant("au", [16951488]),
"dns-search": Variant("as", []),
"gateway": Variant("s", "192.168.2.1"),
"method": Variant("s", "auto"),
"route-data": Variant(
"aa{sv}",
[
{
"dest": Variant("s", "192.168.122.0"),
"prefix": Variant("u", 24),
"next-hop": Variant("s", "10.10.10.1"),
}
],
),
"routes": Variant("aau", [[8038592, 24, 17435146, 0]]),
},
"ipv6": {
"address-data": Variant("aa{sv}", []),
"addresses": Variant("a(ayuay)", []),
"dns": Variant("au", []),
"dns-search": Variant("as", []),
"method": Variant("s", "auto"),
"route-data": Variant("aa{sv}", []),
"routes": Variant("aau", []),
"addr-gen-mode": Variant("i", 0),
},
"proxy": {},
"802-3-ethernet": {
"auto-negotiate": Variant("b", False),
"mac-address-blacklist": Variant("as", []),
"s390-options": Variant("a{ss}", {}),
},
"802-11-wireless": {"ssid": Variant("ay", bytes([78, 69, 84, 84]))},
}
]
return [SETTINGS_WITH_SIGNATURE]
else:
assert method == "org.freedesktop.NetworkManager.Settings.Connection.Update"
assert len(args[0]) == 2

View File

@ -0,0 +1,15 @@
"""Test NetworkWireless AP object."""
from supervisor.dbus.network.accesspoint import NetworkWirelessAP
async def test_accesspoint(dbus: list[str]):
"""Test accesspoint."""
wireless_ap = NetworkWirelessAP("/org/freedesktop/NetworkManager/AccessPoint/43099")
assert wireless_ap.mac is None
assert wireless_ap.mode is None
await wireless_ap.connect()
assert wireless_ap.mac == "E4:57:40:A9:D7:DE"
assert wireless_ap.mode == 2

View File

@ -0,0 +1,16 @@
"""Test DNS Manager object."""
from ipaddress import IPv4Address
from supervisor.dbus.network import NetworkManager
from supervisor.dbus.network.configuration import DNSConfiguration
async def test_dns(network_manager: NetworkManager):
"""Test dns manager."""
assert network_manager.dns.mode == "default"
assert network_manager.dns.rc_manager == "file"
assert network_manager.dns.configuration == [
DNSConfiguration(
[IPv4Address("192.168.30.1")], ["syshack.ch"], "eth0", 100, False
)
]

View File

@ -1,12 +1,14 @@
"""Test NetwrokInterface."""
from unittest.mock import AsyncMock, patch
"""Test NetworkInterface."""
from unittest.mock import AsyncMock
import pytest
from supervisor.dbus.const import ConnectionStateType
from supervisor.dbus.network import NetworkManager
from supervisor.exceptions import HostNotSupportedError
from tests.const import TEST_INTERFACE
from tests.dbus.network.setting.test_init import SETTINGS_WITH_SIGNATURE
# pylint: disable=protected-access
@ -29,26 +31,48 @@ async def test_network_manager_version(network_manager: NetworkManager):
assert network_manager.version == "1.13.9"
async def test_check_connectivity(network_manager: NetworkManager):
async def test_check_connectivity(network_manager: NetworkManager, dbus: list[str]):
"""Test connectivity check."""
dbus.clear()
assert await network_manager.check_connectivity() == 4
assert dbus == [
"/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.Connectivity"
]
dbus.clear()
assert await network_manager.check_connectivity(force=True) == 4
assert dbus == [
"/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.CheckConnectivity"
]
with patch.object(
type(network_manager.dbus), "call_dbus"
) as call_dbus, patch.object(
type(network_manager.dbus), "get_property"
) as get_property:
await network_manager.check_connectivity()
call_dbus.assert_not_called()
get_property.assert_called_once_with(
"org.freedesktop.NetworkManager", "Connectivity"
)
get_property.reset_mock()
await network_manager.check_connectivity(force=True)
async def test_activate_connection(network_manager: NetworkManager, dbus: list[str]):
"""Test activate connection."""
dbus.clear()
connection = await network_manager.activate_connection(
"/org/freedesktop/NetworkManager/Settings/1",
"/org/freedesktop/NetworkManager/Devices/1",
)
assert connection.state == ConnectionStateType.ACTIVATED
assert connection.setting_object == "/org/freedesktop/NetworkManager/Settings/1"
assert dbus == [
"/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.ActivateConnection"
]
call_dbus.assert_called_once_with(
"org.freedesktop.NetworkManager.CheckConnectivity", remove_signature=True
)
get_property.assert_not_called()
async def test_add_and_activate_connection(
network_manager: NetworkManager, dbus: list[str]
):
"""Test add and activate connection."""
dbus.clear()
settings, connection = await network_manager.add_and_activate_connection(
SETTINGS_WITH_SIGNATURE, "/org/freedesktop/NetworkManager/Devices/1"
)
assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6"
assert settings.ipv4.method == "auto"
assert connection.state == ConnectionStateType.ACTIVATED
assert connection.setting_object == "/org/freedesktop/NetworkManager/Settings/1"
assert dbus == [
"/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.AddAndActivateConnection",
"/org/freedesktop/NetworkManager/Settings/1-org.freedesktop.NetworkManager.Settings.Connection.GetSettings",
]

View File

@ -0,0 +1,25 @@
"""Test Network Manager Connection Settings Profile Manager."""
from supervisor.dbus.network import NetworkManager
from tests.dbus.network.setting.test_init import SETTINGS_WITH_SIGNATURE
async def test_add_connection(network_manager: NetworkManager, dbus: list[str]):
"""Test adding settings connection."""
dbus.clear()
settings = await network_manager.settings.add_connection(SETTINGS_WITH_SIGNATURE)
assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6"
assert settings.ipv4.method == "auto"
assert dbus == [
"/org/freedesktop/NetworkManager/Settings-org.freedesktop.NetworkManager.Settings.AddConnection",
"/org/freedesktop/NetworkManager/Settings/1-org.freedesktop.NetworkManager.Settings.Connection.GetSettings",
]
async def test_reload_connections(network_manager: NetworkManager, dbus: list[str]):
"""Test reload connections."""
dbus.clear()
assert await network_manager.settings.reload_connections() is True
assert dbus == [
"/org/freedesktop/NetworkManager/Settings-org.freedesktop.NetworkManager.Settings.ReloadConnections"
]

View File

@ -0,0 +1,27 @@
"""Test Network Manager Wireless object."""
from supervisor.dbus.network import NetworkManager
async def test_request_scan(network_manager: NetworkManager, dbus: list[str]):
"""Test request scan."""
dbus.clear()
assert await network_manager.interfaces["wlan0"].wireless.request_scan() is None
assert dbus == [
"/org/freedesktop/NetworkManager/Devices/3-org.freedesktop.NetworkManager.Device.Wireless.RequestScan"
]
async def test_get_all_access_points(network_manager: NetworkManager, dbus: list[str]):
"""Test get all access points."""
dbus.clear()
accesspoints = await network_manager.interfaces[
"wlan0"
].wireless.get_all_accesspoints()
assert len(accesspoints) == 2
assert accesspoints[0].mac == "E4:57:40:A9:D7:DE"
assert accesspoints[0].mode == 2
assert accesspoints[1].mac == "18:4B:0D:23:A1:9C"
assert accesspoints[1].mode == 2
assert dbus == [
"/org/freedesktop/NetworkManager/Devices/3-org.freedesktop.NetworkManager.Device.Wireless.GetAllAccessPoints"
]

View File

@ -22,12 +22,15 @@ async def test_dbus_hostname_info(coresys: CoreSys):
assert coresys.dbus.hostname.operating_system == "Home Assistant OS 6.0.dev20210504"
async def test_dbus_sethostname(coresys: CoreSys):
async def test_dbus_sethostname(coresys: CoreSys, dbus: list[str]):
"""Set hostname on backend."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.hostname.set_static_hostname("StarWars")
await coresys.dbus.hostname.connect()
dbus.clear()
await coresys.dbus.hostname.set_static_hostname("StarWars")
assert dbus == [
"/org/freedesktop/hostname1-org.freedesktop.hostname1.SetStaticHostname"
]

29
tests/dbus/test_login.py Normal file
View File

@ -0,0 +1,29 @@
"""Test login dbus interface."""
import pytest
from supervisor.coresys import CoreSys
from supervisor.exceptions import DBusNotConnectedError
async def test_reboot(coresys: CoreSys, dbus: list[str]):
"""Test reboot."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.logind.reboot()
await coresys.dbus.logind.connect()
dbus.clear()
assert await coresys.dbus.logind.reboot() is None
assert dbus == ["/org/freedesktop/login1-org.freedesktop.login1.Manager.Reboot"]
async def test_power_off(coresys: CoreSys, dbus: list[str]):
"""Test power off."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.logind.power_off()
await coresys.dbus.logind.connect()
dbus.clear()
assert await coresys.dbus.logind.power_off() is None
assert dbus == ["/org/freedesktop/login1-org.freedesktop.login1.Manager.PowerOff"]

68
tests/dbus/test_rauc.py Normal file
View File

@ -0,0 +1,68 @@
"""Test rauc dbus interface."""
import pytest
from supervisor.coresys import CoreSys
from supervisor.dbus.const import RaucState
from supervisor.exceptions import DBusNotConnectedError
async def test_rauc(coresys: CoreSys):
"""Test rauc properties."""
assert coresys.dbus.rauc.boot_slot is None
assert coresys.dbus.rauc.operation is None
await coresys.dbus.rauc.connect()
await coresys.dbus.rauc.update()
assert coresys.dbus.rauc.boot_slot == "B"
assert coresys.dbus.rauc.operation == "idle"
async def test_install(coresys: CoreSys, dbus: list[str]):
"""Test install."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.rauc.install("rauc_file")
await coresys.dbus.rauc.connect()
dbus.clear()
async with coresys.dbus.rauc.signal_completed() as signal:
assert await coresys.dbus.rauc.install("rauc_file") is None
assert await signal.wait_for_signal() == [0]
assert dbus == ["/-de.pengutronix.rauc.Installer.Install"]
async def test_get_slot_status(coresys: CoreSys, dbus: list[str]):
"""Test get slot status."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.rauc.get_slot_status()
await coresys.dbus.rauc.connect()
dbus.clear()
slot_status = await coresys.dbus.rauc.get_slot_status()
assert len(slot_status) == 6
assert slot_status[0][0] == "kernel.0"
assert slot_status[0][1]["boot-status"] == "good"
assert slot_status[0][1]["device"] == "/dev/disk/by-partlabel/hassos-kernel0"
assert slot_status[0][1]["bootname"] == "A"
assert slot_status[4][0] == "kernel.1"
assert slot_status[4][1]["boot-status"] == "good"
assert slot_status[4][1]["device"] == "/dev/disk/by-partlabel/hassos-kernel1"
assert slot_status[4][1]["bootname"] == "B"
assert dbus == ["/-de.pengutronix.rauc.Installer.GetSlotStatus"]
async def test_mark(coresys: CoreSys, dbus: list[str]):
"""Test mark."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.rauc.mark(RaucState.GOOD, "booted")
await coresys.dbus.rauc.connect()
dbus.clear()
mark = await coresys.dbus.rauc.mark(RaucState.GOOD, "booted")
assert mark[0] == "kernel.1"
assert mark[1] == "marked slot kernel.1 as good"
assert dbus == ["/-de.pengutronix.rauc.Installer.Mark"]

View File

@ -2,8 +2,11 @@
from unittest.mock import patch
import pytest
from supervisor.coresys import CoreSys
from supervisor.dbus.const import DBUS_NAME_SYSTEMD
from supervisor.exceptions import DBusNotConnectedError
from tests.common import load_json_fixture
@ -25,3 +28,116 @@ async def test_dbus_systemd_info(coresys: CoreSys):
assert coresys.dbus.systemd.boot_timestamp == 1632236713344227
assert coresys.dbus.systemd.startup_time == 45.304696
async def test_reboot(coresys: CoreSys, dbus: list[str]):
"""Test reboot."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.reboot()
await coresys.dbus.systemd.connect()
dbus.clear()
assert await coresys.dbus.systemd.reboot() is None
assert dbus == ["/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.Reboot"]
async def test_power_off(coresys: CoreSys, dbus: list[str]):
"""Test power off."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.power_off()
await coresys.dbus.systemd.connect()
dbus.clear()
assert await coresys.dbus.systemd.power_off() is None
assert dbus == [
"/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.PowerOff"
]
async def test_start_unit(coresys: CoreSys, dbus: list[str]):
"""Test start unit."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.start_unit("test_unit", "replace")
await coresys.dbus.systemd.connect()
dbus.clear()
assert (
await coresys.dbus.systemd.start_unit("test_unit", "replace")
== "/org/freedesktop/systemd1/job/7623"
)
assert dbus == [
"/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.StartUnit"
]
async def test_stop_unit(coresys: CoreSys, dbus: list[str]):
"""Test stop unit."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.stop_unit("test_unit", "replace")
await coresys.dbus.systemd.connect()
dbus.clear()
assert (
await coresys.dbus.systemd.stop_unit("test_unit", "replace")
== "/org/freedesktop/systemd1/job/7623"
)
assert dbus == [
"/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.StopUnit"
]
async def test_restart_unit(coresys: CoreSys, dbus: list[str]):
"""Test restart unit."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.restart_unit("test_unit", "replace")
await coresys.dbus.systemd.connect()
dbus.clear()
assert (
await coresys.dbus.systemd.restart_unit("test_unit", "replace")
== "/org/freedesktop/systemd1/job/7623"
)
assert dbus == [
"/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.RestartUnit"
]
async def test_reload_unit(coresys: CoreSys, dbus: list[str]):
"""Test reload unit."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.reload_unit("test_unit", "replace")
await coresys.dbus.systemd.connect()
dbus.clear()
assert (
await coresys.dbus.systemd.reload_unit("test_unit", "replace")
== "/org/freedesktop/systemd1/job/7623"
)
assert dbus == [
"/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.ReloadOrRestartUnit"
]
async def test_list_units(coresys: CoreSys, dbus: list[str]):
"""Test list units."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.systemd.list_units()
await coresys.dbus.systemd.connect()
dbus.clear()
units = await coresys.dbus.systemd.list_units()
assert len(units) == 4
assert units[1][0] == "firewalld.service"
assert units[1][2] == "not-found"
assert units[3][0] == "zram-swap.service"
assert units[3][2] == "loaded"
assert dbus == [
"/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.ListUnits"
]

View File

@ -23,7 +23,7 @@ async def test_dbus_timezone(coresys: CoreSys):
)
async def test_dbus_settime(coresys: CoreSys):
async def test_dbus_settime(coresys: CoreSys, dbus: list[str]):
"""Set timestamp on backend."""
test_dt = datetime(2021, 5, 19, 8, 36, 54, 405718, tzinfo=timezone.utc)
@ -32,14 +32,18 @@ async def test_dbus_settime(coresys: CoreSys):
await coresys.dbus.timedate.connect()
await coresys.dbus.timedate.set_time(test_dt)
dbus.clear()
assert await coresys.dbus.timedate.set_time(test_dt) is None
assert dbus == ["/org/freedesktop/timedate1-org.freedesktop.timedate1.SetTime"]
async def test_dbus_setntp(coresys: CoreSys):
async def test_dbus_setntp(coresys: CoreSys, dbus: list[str]):
"""Disable NTP on backend."""
with pytest.raises(DBusNotConnectedError):
await coresys.dbus.timedate.set_ntp(False)
await coresys.dbus.timedate.connect()
await coresys.dbus.timedate.set_ntp(False)
dbus.clear()
assert await coresys.dbus.timedate.set_ntp(False) is None
assert dbus == ["/org/freedesktop/timedate1-org.freedesktop.timedate1.SetNTP"]

74
tests/fixtures/de_pengutronix_rauc.xml vendored Normal file
View File

@ -0,0 +1,74 @@
<!DOCTYPE node PUBLIC "-//freedesktop//DTD D-BUS Object Introspection 1.0//EN" "http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd">
<node>
<interface name="org.freedesktop.DBus.Properties">
<method name="Get">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="out"/>
</method>
<method name="GetAll">
<arg type="s" name="interface_name" direction="in"/>
<arg type="a{sv}" name="properties" direction="out"/>
</method>
<method name="Set">
<arg type="s" name="interface_name" direction="in"/>
<arg type="s" name="property_name" direction="in"/>
<arg type="v" name="value" direction="in"/>
</method>
<signal name="PropertiesChanged">
<arg type="s" name="interface_name"/>
<arg type="a{sv}" name="changed_properties"/>
<arg type="as" name="invalidated_properties"/>
</signal>
</interface>
<interface name="org.freedesktop.DBus.Introspectable">
<method name="Introspect">
<arg type="s" name="xml_data" direction="out"/>
</method>
</interface>
<interface name="org.freedesktop.DBus.Peer">
<method name="Ping"/>
<method name="GetMachineId">
<arg type="s" name="machine_uuid" direction="out"/>
</method>
</interface>
<interface name="de.pengutronix.rauc.Installer">
<method name="Install">
<arg type="s" name="source" direction="in"/>
</method>
<method name="InstallBundle">
<annotation name="org.qtproject.QtDBus.QtTypeName.In1" value="QVariantMap"/>
<arg type="s" name="source" direction="in"/>
<arg type="a{sv}" name="args" direction="in"/>
</method>
<method name="Info">
<arg type="s" name="bundle" direction="in"/>
<arg type="s" name="compatible" direction="out"/>
<arg type="s" name="version" direction="out"/>
</method>
<method name="Mark">
<arg type="s" name="state" direction="in"/>
<arg type="s" name="slot_identifier" direction="in"/>
<arg type="s" name="slot_name" direction="out"/>
<arg type="s" name="message" direction="out"/>
</method>
<method name="GetSlotStatus">
<annotation name="org.qtproject.QtDBus.QtTypeName.Out0" value="RaucSlotStatusArray"/>
<arg type="a(sa{sv})" name="slot_status_array" direction="out"/>
</method>
<method name="GetPrimary">
<arg type="s" name="primary" direction="out"/>
</method>
<signal name="Completed">
<arg type="i" name="result"/>
</signal>
<property type="s" name="Operation" access="read"/>
<property type="s" name="LastError" access="read"/>
<property type="(isi)" name="Progress" access="read">
<annotation name="org.qtproject.QtDBus.QtTypeName" value="RaucProgress"/>
</property>
<property type="s" name="Compatible" access="read"/>
<property type="s" name="Variant" access="read"/>
<property type="s" name="BootSlot" access="read"/>
</interface>
</node>

View File

@ -0,0 +1,8 @@
{
"Operation": "idle",
"LastError": "",
"Progress": [0, "", 0],
"Compatible": "haos-odroid-n2",
"Variant": "",
"BootSlot": "B"
}

View File

@ -0,0 +1,110 @@
[
[
[
"kernel.0",
{
"activated.count": 9,
"activated.timestamp": "2022-08-23T21:03:22Z",
"boot-status": "good",
"bundle.compatible": "haos-odroid-n2",
"sha256": "c624db648b8401fae37ee5bb1a6ec90bdf4183aef364b33314a73c7198e49d5b",
"state": "inactive",
"size": 10371072,
"installed.count": 9,
"class": "kernel",
"device": "/dev/disk/by-partlabel/hassos-kernel0",
"type": "raw",
"bootname": "A",
"bundle.version": "9.0.dev20220818",
"installed.timestamp": "2022-08-23T21:03:16Z",
"status": "ok"
}
],
[
"boot.0",
{
"bundle.compatible": "haos-odroid-n2",
"sha256": "a5019b335f33be2cf89c96bb2d0695030adb72c1d13d650a5bbe1806dd76d6cc",
"state": "inactive",
"size": 25165824,
"installed.count": 19,
"class": "boot",
"device": "/dev/disk/by-partlabel/hassos-boot",
"type": "vfat",
"status": "ok",
"bundle.version": "9.0.dev20220824",
"installed.timestamp": "2022-08-25T21:11:46Z"
}
],
[
"rootfs.0",
{
"bundle.compatible": "haos-odroid-n2",
"parent": "kernel.0",
"state": "inactive",
"size": 117456896,
"sha256": "7d908b4d578d072b1b0f75de8250fd97b6e119bff09518a96fffd6e4aec61721",
"class": "rootfs",
"device": "/dev/disk/by-partlabel/hassos-system0",
"type": "raw",
"status": "ok",
"bundle.version": "9.0.dev20220818",
"installed.timestamp": "2022-08-23T21:03:21Z",
"installed.count": 9
}
],
[
"spl.0",
{
"bundle.compatible": "haos-odroid-n2",
"sha256": "9856a94df1d6abbc672adaf95746ec76abd3a8191f9d08288add6bb39e63ef45",
"state": "inactive",
"size": 8388608,
"installed.count": 19,
"class": "spl",
"device": "/dev/disk/by-partlabel/hassos-boot",
"type": "raw",
"status": "ok",
"bundle.version": "9.0.dev20220824",
"installed.timestamp": "2022-08-25T21:11:51Z"
}
],
[
"kernel.1",
{
"activated.count": 10,
"activated.timestamp": "2022-08-25T21:11:52Z",
"boot-status": "good",
"bundle.compatible": "haos-odroid-n2",
"sha256": "f57e354b8bd518022721e71fafaf278972af966d8f6cbefb4610db13785801c8",
"state": "booted",
"size": 10371072,
"installed.count": 10,
"class": "kernel",
"device": "/dev/disk/by-partlabel/hassos-kernel1",
"type": "raw",
"bootname": "B",
"bundle.version": "9.0.dev20220824",
"installed.timestamp": "2022-08-25T21:11:46Z",
"status": "ok"
}
],
[
"rootfs.1",
{
"bundle.compatible": "haos-odroid-n2",
"parent": "kernel.1",
"state": "active",
"size": 117456896,
"sha256": "55936b64d391954ae1aed24dd1460e191e021e78655470051fa7939d12fff68a",
"class": "rootfs",
"device": "/dev/disk/by-partlabel/hassos-system1",
"type": "raw",
"status": "ok",
"bundle.version": "9.0.dev20220824",
"installed.timestamp": "2022-08-25T21:11:51Z",
"installed.count": 10
}
]
]
]

View File

@ -0,0 +1 @@
["kernel.1", "marked slot kernel.1 as good"]

View File

@ -0,0 +1 @@
["/org/freedesktop/NetworkManager/Settings/1"]

View File

@ -0,0 +1 @@
[true]

View File

@ -1,50 +1,52 @@
[
[
"etc-machine\\x2did.mount",
"/etc/machine-id",
"loaded",
"active",
"mounted",
"",
"/org/freedesktop/systemd1/unit/etc_2dmachine_5cx2did_2emount",
0,
"",
"/"
],
[
"firewalld.service",
"firewalld.service",
"not-found",
"inactive",
"dead",
"",
"/org/freedesktop/systemd1/unit/firewalld_2eservice",
0,
"",
"/"
],
[
"sys-devices-virtual-tty-ttypd.device",
"/sys/devices/virtual/tty/ttypd",
"loaded",
"active",
"plugged",
"",
"/org/freedesktop/systemd1/unit/sys_2ddevices_2dvirtual_2dtty_2dttypd_2edevice",
0,
"",
"/"
],
[
"zram-swap.service",
"HassOS ZRAM swap",
"loaded",
"active",
"exited",
"",
"/org/freedesktop/systemd1/unit/zram_2dswap_2eservice",
0,
"",
"/"
[
"etc-machine\\x2did.mount",
"/etc/machine-id",
"loaded",
"active",
"mounted",
"",
"/org/freedesktop/systemd1/unit/etc_2dmachine_5cx2did_2emount",
0,
"",
"/"
],
[
"firewalld.service",
"firewalld.service",
"not-found",
"inactive",
"dead",
"",
"/org/freedesktop/systemd1/unit/firewalld_2eservice",
0,
"",
"/"
],
[
"sys-devices-virtual-tty-ttypd.device",
"/sys/devices/virtual/tty/ttypd",
"loaded",
"active",
"plugged",
"",
"/org/freedesktop/systemd1/unit/sys_2ddevices_2dvirtual_2dtty_2dttypd_2edevice",
0,
"",
"/"
],
[
"zram-swap.service",
"HassOS ZRAM swap",
"loaded",
"active",
"exited",
"",
"/org/freedesktop/systemd1/unit/zram_2dswap_2eservice",
0,
"",
"/"
]
]
]

View File

@ -0,0 +1 @@
["/org/freedesktop/systemd1/job/7623"]

View File

@ -0,0 +1 @@
["/org/freedesktop/systemd1/job/7623"]

View File

@ -0,0 +1 @@
["/org/freedesktop/systemd1/job/7623"]

View File

@ -0,0 +1 @@
["/org/freedesktop/systemd1/job/7623"]

View File

@ -1 +0,0 @@
[]

View File

@ -1 +0,0 @@
[]

View File

@ -2,10 +2,14 @@
from ipaddress import IPv4Address, IPv6Address
from unittest.mock import Mock, PropertyMock, patch
import pytest
from supervisor.coresys import CoreSys
from supervisor.dbus.const import ConnectionStateFlags, InterfaceMethod
from supervisor.host.const import InterfaceType
from supervisor.exceptions import DBusFatalError, HostNotSupportedError
from supervisor.host.const import InterfaceType, WifiMode
from supervisor.host.network import Interface, IpConfig
from supervisor.utils.dbus import DBus
async def test_load(coresys: CoreSys):
@ -103,3 +107,46 @@ async def test_load_with_network_connection_issues(coresys: CoreSys):
assert coresys.host.network.interfaces[0].ipv6.gateway == IPv6Address(
"fe80::da58:d7ff:fe00:9c69"
)
async def test_scan_wifi(coresys: CoreSys):
"""Test scanning wifi."""
with pytest.raises(HostNotSupportedError):
await coresys.host.network.scan_wifi(coresys.host.network.get("eth0"))
with patch("supervisor.host.network.asyncio.sleep"):
aps = await coresys.host.network.scan_wifi(coresys.host.network.get("wlan0"))
assert len(aps) == 2
assert aps[0].mac == "E4:57:40:A9:D7:DE"
assert aps[0].mode == WifiMode.INFRASTRUCTURE
assert aps[1].mac == "18:4B:0D:23:A1:9C"
assert aps[1].mode == WifiMode.INFRASTRUCTURE
async def test_scan_wifi_with_failures(coresys: CoreSys, caplog):
"""Test scanning wifi with accesspoint processing failures."""
# pylint: disable=protected-access
init_proxy = coresys.dbus.network.dbus._init_proxy
async def mock_init_proxy(self):
if self.object_path != "/org/freedesktop/NetworkManager/AccessPoint/99999":
return await init_proxy()
raise DBusFatalError("Fail")
with patch("supervisor.host.network.asyncio.sleep"), patch.object(
DBus,
"call_dbus",
return_value=[
[
"/org/freedesktop/NetworkManager/AccessPoint/43099",
"/org/freedesktop/NetworkManager/AccessPoint/43100",
"/org/freedesktop/NetworkManager/AccessPoint/99999",
]
],
), patch.object(DBus, "_init_proxy", new=mock_init_proxy):
aps = await coresys.host.network.scan_wifi(coresys.host.network.get("wlan0"))
assert len(aps) == 2
assert "Can't process an AP" in caplog.text