jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-16 13:46:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add support for green LEDs to API (#4556)

Browse Source 
* Add support for green LEDs to API

* Save board config in supervisor and post on start

* Ignore no-value-for-parameter in validate
This commit is contained in:
Mike Degatano 2023-09-14 09:27:12 -04:00 committed by GitHub
parent d96598b5dd
commit e1232bc9e7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 424 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
supervisor/api/__init__.py
View File

@ -186,6 +186,8 @@ class RestAPI(CoreSysAttributes):
# Boards endpoints
self.webapp.add_routes(
[
web.get("/os/boards/green", api_os.boards_green_info),
web.post("/os/boards/green", api_os.boards_green_options),
web.get("/os/boards/yellow", api_os.boards_yellow_info),
web.post("/os/boards/yellow", api_os.boards_yellow_options),
web.get("/os/boards/{board}", api_os.boards_other_info),

3
supervisor/api/const.py
View File

@ -23,7 +23,6 @@ ATTR_CONNECTION_BUS = "connection_bus"
ATTR_DATA_DISK = "data_disk"
ATTR_DEVICE = "device"
ATTR_DEV_PATH = "dev_path"
ATTR_DISK_LED = "disk_led"
ATTR_DISKS = "disks"
ATTR_DRIVES = "drives"
ATTR_DT_SYNCHRONIZED = "dt_synchronized"
@ -31,7 +30,6 @@ ATTR_DT_UTC = "dt_utc"
ATTR_EJECTABLE = "ejectable"
ATTR_FALLBACK = "fallback"
ATTR_FILESYSTEMS = "filesystems"
ATTR_HEARTBEAT_LED = "heartbeat_led"
ATTR_IDENTIFIERS = "identifiers"
ATTR_JOBS = "jobs"
ATTR_LLMNR = "llmnr"
@ -41,7 +39,6 @@ ATTR_MODEL = "model"
ATTR_MOUNTS = "mounts"
ATTR_MOUNT_POINTS = "mount_points"
ATTR_PANEL_PATH = "panel_path"
ATTR_POWER_LED = "power_led"
ATTR_REMOVABLE = "removable"
ATTR_REVISION = "revision"
ATTR_SEAT = "seat"

44
supervisor/api/os.py
View File

@ -8,14 +8,19 @@ from aiohttp import web
import voluptuous as vol
from ..const import (
ATTR_ACTIVITY_LED,
ATTR_BOARD,
ATTR_BOOT,
ATTR_DEVICES,
ATTR_DISK_LED,
ATTR_HEARTBEAT_LED,
ATTR_ID,
ATTR_NAME,
ATTR_POWER_LED,
ATTR_SERIAL,
ATTR_SIZE,
ATTR_UPDATE_AVAILABLE,
ATTR_USER_LED,
ATTR_VERSION,
ATTR_VERSION_LATEST,
)
@ -27,21 +32,18 @@ from .const import (
ATTR_DATA_DISK,
ATTR_DEV_PATH,
ATTR_DEVICE,
ATTR_DISK_LED,
ATTR_DISKS,
ATTR_HEARTBEAT_LED,
ATTR_MODEL,
ATTR_POWER_LED,
ATTR_VENDOR,
)
from .utils import api_process, api_validate
_LOGGER: logging.Logger = logging.getLogger(__name__)
# pylint: disable=no-value-for-parameter
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
SCHEMA_DISK = vol.Schema({vol.Required(ATTR_DEVICE): str})
# pylint: disable=no-value-for-parameter
SCHEMA_YELLOW_OPTIONS = vol.Schema(
{
vol.Optional(ATTR_DISK_LED): vol.Boolean(),
@ -49,6 +51,14 @@ SCHEMA_YELLOW_OPTIONS = vol.Schema(
vol.Optional(ATTR_POWER_LED): vol.Boolean(),
}
)
SCHEMA_GREEN_OPTIONS = vol.Schema(
{
vol.Optional(ATTR_ACTIVITY_LED): vol.Boolean(),
vol.Optional(ATTR_POWER_LED): vol.Boolean(),
vol.Optional(ATTR_USER_LED): vol.Boolean(),
}
)
# pylint: enable=no-value-for-parameter
class APIOS(CoreSysAttributes):
@ -105,6 +115,31 @@ class APIOS(CoreSysAttributes):
],
}
@api_process
async def boards_green_info(self, request: web.Request) -> dict[str, Any]:
"""Get green board settings."""
return {
ATTR_ACTIVITY_LED: self.sys_dbus.agent.board.green.activity_led,
ATTR_POWER_LED: self.sys_dbus.agent.board.green.power_led,
ATTR_USER_LED: self.sys_dbus.agent.board.green.user_led,
}
@api_process
async def boards_green_options(self, request: web.Request) -> None:
"""Update green board settings."""
body = await api_validate(SCHEMA_GREEN_OPTIONS, request)
if ATTR_ACTIVITY_LED in body:
self.sys_dbus.agent.board.green.activity_led = body[ATTR_ACTIVITY_LED]
if ATTR_POWER_LED in body:
self.sys_dbus.agent.board.green.power_led = body[ATTR_POWER_LED]
if ATTR_USER_LED in body:
self.sys_dbus.agent.board.green.user_led = body[ATTR_USER_LED]
self.sys_dbus.agent.board.green.save_data()
@api_process
async def boards_yellow_info(self, request: web.Request) -> dict[str, Any]:
"""Get yellow board settings."""
@ -128,6 +163,7 @@ class APIOS(CoreSysAttributes):
if ATTR_POWER_LED in body:
self.sys_dbus.agent.board.yellow.power_led = body[ATTR_POWER_LED]
self.sys_dbus.agent.board.yellow.save_data()
self.sys_resolution.create_issue(
IssueType.REBOOT_REQUIRED,
ContextType.SYSTEM,

6
supervisor/const.py
View File

@ -19,6 +19,7 @@ SUPERVISOR_DATA = Path("/data")
FILE_HASSIO_ADDONS = Path(SUPERVISOR_DATA, "addons.json")
FILE_HASSIO_AUTH = Path(SUPERVISOR_DATA, "auth.json")
FILE_HASSIO_BACKUPS = Path(SUPERVISOR_DATA, "backups.json")
FILE_HASSIO_BOARD = Path(SUPERVISOR_DATA, "board.json")
FILE_HASSIO_CONFIG = Path(SUPERVISOR_DATA, "config.json")
FILE_HASSIO_DISCOVERY = Path(SUPERVISOR_DATA, "discovery.json")
FILE_HASSIO_DOCKER = Path(SUPERVISOR_DATA, "docker.json")
@ -88,6 +89,7 @@ REQUEST_FROM = "HASSIO_FROM"
ATTR_ACCESS_TOKEN = "access_token"
ATTR_ACCESSPOINTS = "accesspoints"
ATTR_ACTIVE = "active"
ATTR_ACTIVITY_LED = "activity_led"
ATTR_ADDON = "addon"
ATTR_ADDONS = "addons"
ATTR_ADDONS_CUSTOM_LIST = "addons_custom_list"
@ -152,6 +154,7 @@ ATTR_DIAGNOSTICS = "diagnostics"
ATTR_DISCOVERY = "discovery"
ATTR_DISK = "disk"
ATTR_DISK_FREE = "disk_free"
ATTR_DISK_LED = "disk_led"
ATTR_DISK_LIFE_TIME = "disk_life_time"
ATTR_DISK_TOTAL = "disk_total"
ATTR_DISK_USED = "disk_used"
@ -177,6 +180,7 @@ ATTR_HASSIO_API = "hassio_api"
ATTR_HASSIO_ROLE = "hassio_role"
ATTR_HASSOS = "hassos"
ATTR_HEALTHY = "healthy"
ATTR_HEARTBEAT_LED = "heartbeat_led"
ATTR_HOMEASSISTANT = "homeassistant"
ATTR_HOMEASSISTANT_API = "homeassistant_api"
ATTR_HOST = "host"
@ -252,6 +256,7 @@ ATTR_PLUGINS = "plugins"
ATTR_PORT = "port"
ATTR_PORTS = "ports"
ATTR_PORTS_DESCRIPTION = "ports_description"
ATTR_POWER_LED = "power_led"
ATTR_PREFIX = "prefix"
ATTR_PRIMARY = "primary"
ATTR_PRIORITY = "priority"
@ -315,6 +320,7 @@ ATTR_UPDATE_KEY = "update_key"
ATTR_URL = "url"
ATTR_USB = "usb"
ATTR_USER = "user"
ATTR_USER_LED = "user_led"
ATTR_USERNAME = "username"
ATTR_UUID = "uuid"
ATTR_VALID = "valid"

13
supervisor/dbus/agent/boards/__init__.py
View File

@ -11,7 +11,8 @@ from ...const import (
DBUS_OBJECT_HAOS_BOARDS,
)
from ...interface import DBusInterfaceProxy, dbus_property
from .const import BOARD_NAME_SUPERVISED, BOARD_NAME_YELLOW
from .const import BOARD_NAME_GREEN, BOARD_NAME_SUPERVISED, BOARD_NAME_YELLOW
from .green import Green
from .interface import BoardProxy
from .supervised import Supervised
from .yellow import Yellow
@ -39,6 +40,14 @@ class BoardManager(DBusInterfaceProxy):
"""Get board name."""
return self.properties[DBUS_ATTR_BOARD]
@property
def green(self) -> Green:
"""Get Green board."""
if self.board != BOARD_NAME_GREEN:
raise BoardInvalidError("Green board is not in use", _LOGGER.error)
return self._board_proxy
@property
def supervised(self) -> Supervised:
"""Get Supervised board."""
@ -61,6 +70,8 @@ class BoardManager(DBusInterfaceProxy):
if self.board == BOARD_NAME_YELLOW:
self._board_proxy = Yellow()
elif self.board == BOARD_NAME_GREEN:
self._board_proxy = Green()
elif self.board == BOARD_NAME_SUPERVISED:
self._board_proxy = Supervised()

1
supervisor/dbus/agent/boards/const.py
View File

@ -1,4 +1,5 @@
"""Constants for boards."""
BOARD_NAME_GREEN = "Green"
BOARD_NAME_SUPERVISED = "Supervised"
BOARD_NAME_YELLOW = "Yellow"

65
supervisor/dbus/agent/boards/green.py Normal file
View File

@ -0,0 +1,65 @@
"""Green board management."""
import asyncio
from dbus_fast.aio.message_bus import MessageBus
from ....const import ATTR_ACTIVITY_LED, ATTR_POWER_LED, ATTR_USER_LED
from ...const import DBUS_ATTR_ACTIVITY_LED, DBUS_ATTR_POWER_LED, DBUS_ATTR_USER_LED
from ...interface import dbus_property
from .const import BOARD_NAME_GREEN
from .interface import BoardProxy
from .validate import SCHEMA_GREEN_BOARD
class Green(BoardProxy):
"""Green board manager object."""
def __init__(self) -> None:
"""Initialize properties."""
super().__init__(BOARD_NAME_GREEN, SCHEMA_GREEN_BOARD)
@property
@dbus_property
def activity_led(self) -> bool:
"""Get activity LED enabled."""
return self.properties[DBUS_ATTR_ACTIVITY_LED]
@activity_led.setter
def activity_led(self, enabled: bool) -> None:
"""Enable/disable activity LED."""
self._data[ATTR_ACTIVITY_LED] = enabled
asyncio.create_task(self.dbus.Boards.Green.set_activity_led(enabled))
@property
@dbus_property
def power_led(self) -> bool:
"""Get power LED enabled."""
return self.properties[DBUS_ATTR_POWER_LED]
@power_led.setter
def power_led(self, enabled: bool) -> None:
"""Enable/disable power LED."""
self._data[ATTR_POWER_LED] = enabled
asyncio.create_task(self.dbus.Boards.Green.set_power_led(enabled))
@property
@dbus_property
def user_led(self) -> bool:
"""Get user LED enabled."""
return self.properties[DBUS_ATTR_USER_LED]
@user_led.setter
def user_led(self, enabled: bool) -> None:
"""Enable/disable disk LED."""
self._data[ATTR_USER_LED] = enabled
asyncio.create_task(self.dbus.Boards.Green.set_user_led(enabled))
async def connect(self, bus: MessageBus) -> None:
"""Connect to D-Bus."""
await super().connect(bus)
# Set LEDs based on settings on connect
self.activity_led = self._data[ATTR_ACTIVITY_LED]
self.power_led = self._data[ATTR_POWER_LED]
self.user_led = self._data[ATTR_USER_LED]

12
supervisor/dbus/agent/boards/interface.py
View File

@ -1,17 +1,23 @@
"""Board dbus proxy interface."""
from voluptuous import Schema
from ....const import FILE_HASSIO_BOARD
from ....utils.common import FileConfiguration
from ...const import DBUS_IFACE_HAOS_BOARDS, DBUS_NAME_HAOS, DBUS_OBJECT_HAOS_BOARDS
from ...interface import DBusInterfaceProxy
from .validate import SCHEMA_BASE_BOARD
class BoardProxy(DBusInterfaceProxy):
class BoardProxy(FileConfiguration, DBusInterfaceProxy):
"""DBus interface proxy for os board."""
bus_name: str = DBUS_NAME_HAOS
def __init__(self, name: str) -> None:
def __init__(self, name: str, file_schema: Schema | None = None) -> None:
"""Initialize properties."""
super().__init__()
super().__init__(FILE_HASSIO_BOARD, file_schema or SCHEMA_BASE_BOARD)
super(FileConfiguration, self).__init__()
self._name: str = name
self.object_path: str = f"{DBUS_OBJECT_HAOS_BOARDS}/{name}"

32
supervisor/dbus/agent/boards/validate.py Normal file
View File

@ -0,0 +1,32 @@
"""Validation for board config."""
import voluptuous as vol
from ....const import (
ATTR_ACTIVITY_LED,
ATTR_DISK_LED,
ATTR_HEARTBEAT_LED,
ATTR_POWER_LED,
ATTR_USER_LED,
)
# pylint: disable=no-value-for-parameter
SCHEMA_BASE_BOARD = vol.Schema({}, extra=vol.REMOVE_EXTRA)
SCHEMA_GREEN_BOARD = vol.Schema(
{
vol.Optional(ATTR_ACTIVITY_LED, default=True): vol.Boolean(),
vol.Optional(ATTR_POWER_LED, default=True): vol.Boolean(),
vol.Optional(ATTR_USER_LED, default=True): vol.Boolean(),
},
extra=vol.REMOVE_EXTRA,
)
SCHEMA_YELLOW_BOARD = vol.Schema(
{
vol.Optional(ATTR_DISK_LED, default=True): vol.Boolean(),
vol.Optional(ATTR_HEARTBEAT_LED, default=True): vol.Boolean(),
vol.Optional(ATTR_POWER_LED, default=True): vol.Boolean(),
},
extra=vol.REMOVE_EXTRA,
)

18
supervisor/dbus/agent/boards/yellow.py
View File

@ -2,10 +2,14 @@
import asyncio
from dbus_fast.aio.message_bus import MessageBus
from ....const import ATTR_DISK_LED, ATTR_HEARTBEAT_LED, ATTR_POWER_LED
from ...const import DBUS_ATTR_DISK_LED, DBUS_ATTR_HEARTBEAT_LED, DBUS_ATTR_POWER_LED
from ...interface import dbus_property
from .const import BOARD_NAME_YELLOW
from .interface import BoardProxy
from .validate import SCHEMA_YELLOW_BOARD
class Yellow(BoardProxy):
@ -13,7 +17,7 @@ class Yellow(BoardProxy):
def __init__(self) -> None:
"""Initialize properties."""
super().__init__(BOARD_NAME_YELLOW)
super().__init__(BOARD_NAME_YELLOW, SCHEMA_YELLOW_BOARD)
@property
@dbus_property
@ -24,6 +28,7 @@ class Yellow(BoardProxy):
@heartbeat_led.setter
def heartbeat_led(self, enabled: bool) -> None:
"""Enable/disable heartbeat LED."""
self._data[ATTR_HEARTBEAT_LED] = enabled
asyncio.create_task(self.dbus.Boards.Yellow.set_heartbeat_led(enabled))
@property
@ -35,6 +40,7 @@ class Yellow(BoardProxy):
@power_led.setter
def power_led(self, enabled: bool) -> None:
"""Enable/disable power LED."""
self._data[ATTR_POWER_LED] = enabled
asyncio.create_task(self.dbus.Boards.Yellow.set_power_led(enabled))
@property
@ -46,4 +52,14 @@ class Yellow(BoardProxy):
@disk_led.setter
def disk_led(self, enabled: bool) -> None:
"""Enable/disable disk LED."""
self._data[ATTR_DISK_LED] = enabled
asyncio.create_task(self.dbus.Boards.Yellow.set_disk_led(enabled))
async def connect(self, bus: MessageBus) -> None:
"""Connect to D-Bus."""
await super().connect(bus)
# Set LEDs based on settings on connect
self.disk_led = self._data[ATTR_DISK_LED]
self.heartbeat_led = self._data[ATTR_HEARTBEAT_LED]
self.power_led = self._data[ATTR_POWER_LED]

2
supervisor/dbus/const.py
View File

@ -64,6 +64,7 @@ DBUS_OBJECT_UDISKS2 = "/org/freedesktop/UDisks2/Manager"
DBUS_ATTR_ACTIVE_ACCESSPOINT = "ActiveAccessPoint"
DBUS_ATTR_ACTIVE_CONNECTION = "ActiveConnection"
DBUS_ATTR_ACTIVE_CONNECTIONS = "ActiveConnections"
DBUS_ATTR_ACTIVITY_LED = "ActivityLED"
DBUS_ATTR_ADDRESS_DATA = "AddressData"
DBUS_ATTR_BITRATE = "Bitrate"
DBUS_ATTR_BOARD = "Board"
@ -169,6 +170,7 @@ DBUS_ATTR_TIMEUSEC = "TimeUSec"
DBUS_ATTR_TIMEZONE = "Timezone"
DBUS_ATTR_TRANSACTION_STATISTICS = "TransactionStatistics"
DBUS_ATTR_TYPE = "Type"
DBUS_ATTR_USER_LED = "UserLED"
DBUS_ATTR_USERSPACE_TIMESTAMP_MONOTONIC = "UserspaceTimestampMonotonic"
DBUS_ATTR_UUID_UPPERCASE = "UUID"
DBUS_ATTR_UUID = "Uuid"

71
tests/api/test_os.py
View File

@ -6,6 +6,7 @@ from aiohttp.test_utils import TestClient
import pytest
from supervisor.coresys import CoreSys
from supervisor.dbus.agent.boards.interface import BoardProxy
from supervisor.host.control import SystemControl
from supervisor.os.manager import OSManager
from supervisor.resolution.const import ContextType, IssueType, SuggestionType
@ -13,6 +14,7 @@ from supervisor.resolution.data import Issue, Suggestion
from tests.common import mock_dbus_services
from tests.dbus_service_mocks.agent_boards import Boards as BoardsService
from tests.dbus_service_mocks.agent_boards_green import Green as GreenService
from tests.dbus_service_mocks.agent_boards_yellow import Yellow as YellowService
from tests.dbus_service_mocks.agent_datadisk import DataDisk as DataDiskService
from tests.dbus_service_mocks.base import DBusServiceMock
@ -121,6 +123,7 @@ async def test_api_board_yellow_info(api_client: TestClient, coresys: CoreSys):
assert result["data"]["heartbeat_led"] is True
assert result["data"]["power_led"] is True
assert (await api_client.get("/os/boards/green")).status == 400
assert (await api_client.get("/os/boards/supervised")).status == 400
assert (await api_client.get("/os/boards/not-real")).status == 400
@ -137,11 +140,13 @@ async def test_api_board_yellow_options(
assert coresys.dbus.agent.board.yellow.heartbeat_led is True
assert coresys.dbus.agent.board.yellow.power_led is True
assert len(coresys.resolution.issues) == 0
resp = await api_client.post(
"/os/boards/yellow",
json={"disk_led": False, "heartbeat_led": False, "power_led": False},
)
assert resp.status == 200
with patch.object(BoardProxy, "save_data") as save_data:
resp = await api_client.post(
"/os/boards/yellow",
json={"disk_led": False, "heartbeat_led": False, "power_led": False},
)
assert resp.status == 200
save_data.assert_called_once()
await yellow_service.ping()
assert coresys.dbus.agent.board.yellow.disk_led is False
@ -158,13 +163,65 @@ async def test_api_board_yellow_options(
)
async def test_api_board_green_info(
api_client: TestClient, coresys: CoreSys, boards_service: BoardsService
):
"""Test green board info."""
await mock_dbus_services({"agent_boards_green": None}, coresys.dbus.bus)
boards_service.board = "Green"
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
resp = await api_client.get("/os/boards/green")
assert resp.status == 200
result = await resp.json()
assert result["data"]["activity_led"] is True
assert result["data"]["power_led"] is True
assert result["data"]["user_led"] is True
assert (await api_client.get("/os/boards/yellow")).status == 400
assert (await api_client.get("/os/boards/supervised")).status == 400
assert (await api_client.get("/os/boards/not-real")).status == 400
async def test_api_board_green_options(
api_client: TestClient,
coresys: CoreSys,
boards_service: BoardsService,
):
"""Test yellow board options."""
green_service: GreenService = (
await mock_dbus_services({"agent_boards_green": None}, coresys.dbus.bus)
)["agent_boards_green"]
boards_service.board = "Green"
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
assert coresys.dbus.agent.board.green.activity_led is True
assert coresys.dbus.agent.board.green.power_led is True
assert coresys.dbus.agent.board.green.user_led is True
assert len(coresys.resolution.issues) == 0
with patch.object(BoardProxy, "save_data") as save_data:
resp = await api_client.post(
"/os/boards/green",
json={"activity_led": False, "power_led": False, "user_led": False},
)
assert resp.status == 200
save_data.assert_called_once()
await green_service.ping()
assert coresys.dbus.agent.board.green.activity_led is False
assert coresys.dbus.agent.board.green.power_led is False
assert coresys.dbus.agent.board.green.user_led is False
assert len(coresys.resolution.issues) == 0
async def test_api_board_supervised_info(
api_client: TestClient, coresys: CoreSys, boards_service: BoardsService
):
"""Test supervised board info."""
await mock_dbus_services({"agent_boards_supervised": None}, coresys.dbus.bus)
boards_service.board = "Supervised"
await coresys.dbus.agent.board.update()
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
with patch("supervisor.os.manager.CPE.get_product", return_value=["not-hassos"]):
await coresys.os.load()
@ -180,7 +237,7 @@ async def test_api_board_other_info(
):
"""Test info for other board without dbus object."""
boards_service.board = "not-real"
await coresys.dbus.agent.board.update()
await coresys.dbus.agent.board.connect(coresys.dbus.bus)
with patch.object(OSManager, "board", new=PropertyMock(return_value="not-real")):
assert (await api_client.get("/os/boards/not-real")).status == 200

25
tests/dbus/agent/boards/test_board.py
View File

@ -30,6 +30,27 @@ async def test_dbus_board(dbus_session_bus: MessageBus):
with pytest.raises(BoardInvalidError):
assert not board.supervised
with pytest.raises(BoardInvalidError):
assert not board.green
async def test_dbus_board_green(
boards_service: BoardsService, dbus_session_bus: MessageBus
):
"""Test DBus Board load with Green board."""
await mock_dbus_services({"agent_boards_green": None}, dbus_session_bus)
boards_service.board = "Green"
board = BoardManager()
await board.connect(dbus_session_bus)
assert board.board == "Green"
assert board.green.activity_led is True
with pytest.raises(BoardInvalidError):
assert not board.supervised
with pytest.raises(BoardInvalidError):
assert not board.yellow
async def test_dbus_board_supervised(
@ -47,6 +68,8 @@ async def test_dbus_board_supervised(
with pytest.raises(BoardInvalidError):
assert not board.yellow
with pytest.raises(BoardInvalidError):
assert not board.green
async def test_dbus_board_other(
@ -64,3 +87,5 @@ async def test_dbus_board_other(
assert not board.yellow
with pytest.raises(BoardInvalidError):
assert not board.supervised
with pytest.raises(BoardInvalidError):
assert not board.green

81
tests/dbus/agent/boards/test_green.py Normal file
View File

@ -0,0 +1,81 @@
"""Test Green board."""
# pylint: disable=import-error
import asyncio
from unittest.mock import patch
from dbus_fast.aio.message_bus import MessageBus
import pytest
from supervisor.dbus.agent.boards.green import Green
from tests.common import mock_dbus_services
from tests.dbus_service_mocks.agent_boards_green import Green as GreenService
@pytest.fixture(name="green_service", autouse=True)
async def fixture_green_service(dbus_session_bus: MessageBus) -> GreenService:
"""Mock Green Board dbus service."""
yield (await mock_dbus_services({"agent_boards_green": None}, dbus_session_bus))[
"agent_boards_green"
]
async def test_dbus_green(green_service: GreenService, dbus_session_bus: MessageBus):
"""Test Green board load."""
with patch("supervisor.utils.common.Path.is_file", return_value=True), patch(
"supervisor.utils.common.read_json_file",
return_value={"activity_led": False, "user_led": False},
):
green = Green()
await green.connect(dbus_session_bus)
assert green.name == "Green"
assert green.activity_led is True
assert green.power_led is True
assert green.user_led is True
await asyncio.sleep(0)
await green_service.ping()
assert green.activity_led is False
assert green.user_led is False
async def test_dbus_green_set_activity_led(
green_service: GreenService, dbus_session_bus: MessageBus
):
"""Test setting activity led for Green board."""
green = Green()
await green.connect(dbus_session_bus)
green.activity_led = False
await asyncio.sleep(0) # Set property via dbus is separate async task
await green_service.ping()
assert green.activity_led is False
async def test_dbus_green_set_power_led(
green_service: GreenService, dbus_session_bus: MessageBus
):
"""Test setting power led for Green board."""
green = Green()
await green.connect(dbus_session_bus)
green.power_led = False
await asyncio.sleep(0) # Set property via dbus is separate async task
await green_service.ping()
assert green.power_led is False
async def test_dbus_green_set_user_led(
green_service: GreenService, dbus_session_bus: MessageBus
):
"""Test setting user led for Green board."""
green = Green()
await green.connect(dbus_session_bus)
green.user_led = False
await asyncio.sleep(0) # Set property via dbus is separate async task
await green_service.ping()
assert green.user_led is False

15
tests/dbus/agent/boards/test_yellow.py
View File

@ -1,6 +1,7 @@
"""Test Yellow board."""
# pylint: disable=import-error
import asyncio
from unittest.mock import patch
from dbus_fast.aio.message_bus import MessageBus
import pytest
@ -19,9 +20,13 @@ async def fixture_yellow_service(dbus_session_bus: MessageBus) -> YellowService:
]
async def test_dbus_yellow(dbus_session_bus: MessageBus):
async def test_dbus_yellow(yellow_service: YellowService, dbus_session_bus: MessageBus):
"""Test Yellow board load."""
yellow = Yellow()
with patch("supervisor.utils.common.Path.is_file", return_value=True), patch(
"supervisor.utils.common.read_json_file",
return_value={"disk_led": False, "heartbeat_led": False},
):
yellow = Yellow()
await yellow.connect(dbus_session_bus)
assert yellow.name == "Yellow"
@ -29,6 +34,12 @@ async def test_dbus_yellow(dbus_session_bus: MessageBus):
assert yellow.heartbeat_led is True
assert yellow.power_led is True
await asyncio.sleep(0)
await yellow_service.ping()
assert yellow.disk_led is False
assert yellow.heartbeat_led is False
async def test_dbus_yellow_set_disk_led(
yellow_service: YellowService, dbus_session_bus: MessageBus

55
tests/dbus_service_mocks/agent_boards_green.py Normal file
View File

@ -0,0 +1,55 @@
"""Mock of OS Agent Boards Green dbus service."""
from dbus_fast.service import dbus_property
from .base import DBusServiceMock
BUS_NAME = "io.hass.os"
def setup(object_path: str | None = None) -> DBusServiceMock:
"""Create dbus mock object."""
return Green()
# pylint: disable=invalid-name
class Green(DBusServiceMock):
"""Green mock.
gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/Boards/Green
"""
object_path = "/io/hass/os/Boards/Green"
interface = "io.hass.os.Boards.Green"
@dbus_property()
def ActivityLED(self) -> "b":
"""Get Activity LED."""
return True
@ActivityLED.setter
def ActivityLED(self, value: "b"):
"""Set Activity LED."""
self.emit_properties_changed({"ActivityLED": value})
@dbus_property()
def PowerLED(self) -> "b":
"""Get Power LED."""
return True
@PowerLED.setter
def PowerLED(self, value: "b"):
"""Set Power LED."""
self.emit_properties_changed({"PowerLED": value})
@dbus_property()
def UserLED(self) -> "b":
"""Get User LED."""
return True
@UserLED.setter
def UserLED(self, value: "b"):
"""Set User LED."""
self.emit_properties_changed({"UserLED": value})