diff --git a/homeassistant/components/system_bridge/__init__.py b/homeassistant/components/system_bridge/__init__.py index 931d4df86b7cd..724498ab9a613 100644 --- a/homeassistant/components/system_bridge/__init__.py +++ b/homeassistant/components/system_bridge/__init__.py @@ -1,9 +1,7 @@ """The System Bridge integration.""" import asyncio -from dataclasses import asdict import logging -from typing import Any from systembridgeconnector.exceptions import ( AuthenticationException, @@ -11,71 +9,34 @@ ConnectionErrorException, DataMissingException, ) -from systembridgeconnector.models.keyboard_key import KeyboardKey -from systembridgeconnector.models.keyboard_text import KeyboardText -from systembridgeconnector.models.modules.processes import Process -from systembridgeconnector.models.open_path import OpenPath -from systembridgeconnector.models.open_url import OpenUrl from systembridgeconnector.version import Version -import voluptuous as vol -from homeassistant.config_entries import ConfigEntry, ConfigEntryState +from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( CONF_API_KEY, - CONF_COMMAND, CONF_ENTITY_ID, CONF_HOST, - CONF_ID, CONF_NAME, - CONF_PATH, CONF_PORT, CONF_TOKEN, - CONF_URL, Platform, ) -from homeassistant.core import ( - HomeAssistant, - ServiceCall, - ServiceResponse, - SupportsResponse, -) -from homeassistant.exceptions import ( - ConfigEntryAuthFailed, - ConfigEntryNotReady, - HomeAssistantError, - ServiceValidationError, -) -from homeassistant.helpers import ( - config_validation as cv, - device_registry as dr, - discovery, -) +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady +from homeassistant.helpers import config_validation as cv, discovery from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue +from homeassistant.helpers.typing import ConfigType from .config_flow import SystemBridgeConfigFlow from .const import DATA_WAIT_TIMEOUT, DOMAIN, MODULES from .coordinator import SystemBridgeConfigEntry, SystemBridgeDataUpdateCoordinator - - -def _get_coordinator( - hass: HomeAssistant, entry_id: str -) -> SystemBridgeDataUpdateCoordinator: - """Return the coordinator for a config entry id.""" - entry: SystemBridgeConfigEntry | None = hass.config_entries.async_get_entry( - entry_id - ) - if entry is None or entry.state is not ConfigEntryState.LOADED: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="device_not_found", - translation_placeholders={"device": entry_id}, - ) - return entry.runtime_data - +from .services import async_setup_services _LOGGER = logging.getLogger(__name__) +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) + PLATFORMS = [ Platform.BINARY_SENSOR, Platform.MEDIA_PLAYER, @@ -84,26 +45,12 @@ def _get_coordinator( Platform.UPDATE, ] -CONF_BRIDGE = "bridge" -CONF_KEY = "key" -CONF_TEXT = "text" -SERVICE_GET_PROCESS_BY_ID = "get_process_by_id" -SERVICE_GET_PROCESSES_BY_NAME = "get_processes_by_name" -SERVICE_OPEN_PATH = "open_path" -SERVICE_POWER_COMMAND = "power_command" -SERVICE_OPEN_URL = "open_url" -SERVICE_SEND_KEYPRESS = "send_keypress" -SERVICE_SEND_TEXT = "send_text" +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + """Set up the System Bridge services.""" -POWER_COMMAND_MAP = { - "hibernate": "power_hibernate", - "lock": "power_lock", - "logout": "power_logout", - "restart": "power_restart", - "shutdown": "power_shutdown", - "sleep": "power_sleep", -} + async_setup_services(hass) + return True async def async_setup_entry( @@ -231,219 +178,6 @@ async def async_setup_entry( ) ) - if hass.services.has_service(DOMAIN, SERVICE_OPEN_URL): - return True - - def valid_device(device: str) -> str: - """Check device is valid.""" - device_registry = dr.async_get(hass) - device_entry = device_registry.async_get(device) - if device_entry is not None: - try: - return next( - entry.entry_id - for entry in hass.config_entries.async_entries(DOMAIN) - if entry.entry_id in device_entry.config_entries - ) - except StopIteration as exception: - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="device_not_found", - translation_placeholders={"device": device}, - ) from exception - raise HomeAssistantError( - translation_domain=DOMAIN, - translation_key="device_not_found", - translation_placeholders={"device": device}, - ) - - async def handle_get_process_by_id(service_call: ServiceCall) -> ServiceResponse: - """Handle the get process by id service call.""" - _LOGGER.debug("Get process by id: %s", service_call.data) - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - processes: list[Process] = coordinator.data.processes - - # Find process.id from list, raise ServiceValidationError if not found - try: - return asdict( - next( - process - for process in processes - if process.id == service_call.data[CONF_ID] - ) - ) - except StopIteration as exception: - raise ServiceValidationError( - translation_domain=DOMAIN, - translation_key="process_not_found", - translation_placeholders={"id": service_call.data[CONF_ID]}, - ) from exception - - async def handle_get_processes_by_name( - service_call: ServiceCall, - ) -> ServiceResponse: - """Handle the get process by name service call.""" - _LOGGER.debug("Get process by name: %s", service_call.data) - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - - # Find processes from list - items: list[dict[str, Any]] = [ - asdict(process) - for process in coordinator.data.processes - if process.name is not None - and service_call.data[CONF_NAME].lower() in process.name.lower() - ] - - return { - "count": len(items), - "processes": list(items), - } - - async def handle_open_path(service_call: ServiceCall) -> ServiceResponse: - """Handle the open path service call.""" - _LOGGER.debug("Open path: %s", service_call.data) - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - response = await coordinator.websocket_client.open_path( - OpenPath(path=service_call.data[CONF_PATH]) - ) - return asdict(response) - - async def handle_power_command(service_call: ServiceCall) -> ServiceResponse: - """Handle the power command service call.""" - _LOGGER.debug("Power command: %s", service_call.data) - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - response = await getattr( - coordinator.websocket_client, - POWER_COMMAND_MAP[service_call.data[CONF_COMMAND]], - )() - return asdict(response) - - async def handle_open_url(service_call: ServiceCall) -> ServiceResponse: - """Handle the open url service call.""" - _LOGGER.debug("Open URL: %s", service_call.data) - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - response = await coordinator.websocket_client.open_url( - OpenUrl(url=service_call.data[CONF_URL]) - ) - return asdict(response) - - async def handle_send_keypress(service_call: ServiceCall) -> ServiceResponse: - """Handle the send_keypress service call.""" - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - response = await coordinator.websocket_client.keyboard_keypress( - KeyboardKey(key=service_call.data[CONF_KEY]) - ) - return asdict(response) - - async def handle_send_text(service_call: ServiceCall) -> ServiceResponse: - """Handle the send_keypress service call.""" - coordinator = _get_coordinator(hass, service_call.data[CONF_BRIDGE]) - response = await coordinator.websocket_client.keyboard_text( - KeyboardText(text=service_call.data[CONF_TEXT]) - ) - return asdict(response) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_GET_PROCESS_BY_ID, - handle_get_process_by_id, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_ID): cv.positive_int, - }, - ), - supports_response=SupportsResponse.ONLY, - ) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_GET_PROCESSES_BY_NAME, - handle_get_processes_by_name, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_NAME): cv.string, - }, - ), - supports_response=SupportsResponse.ONLY, - ) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_OPEN_PATH, - handle_open_path, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_PATH): cv.string, - }, - ), - supports_response=SupportsResponse.ONLY, - ) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_POWER_COMMAND, - handle_power_command, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_COMMAND): vol.In(POWER_COMMAND_MAP), - }, - ), - supports_response=SupportsResponse.ONLY, - ) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_OPEN_URL, - handle_open_url, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_URL): cv.string, - }, - ), - supports_response=SupportsResponse.ONLY, - ) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_SEND_KEYPRESS, - handle_send_keypress, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_KEY): cv.string, - }, - ), - supports_response=SupportsResponse.ONLY, - description_placeholders={ - "syntax_keys_documentation_url": "https://robotjs.dev/docs/syntax#keys" - }, - ) - - # pylint: disable-next=home-assistant-service-registered-in-setup-entry - hass.services.async_register( - DOMAIN, - SERVICE_SEND_TEXT, - handle_send_text, - schema=vol.Schema( - { - vol.Required(CONF_BRIDGE): valid_device, - vol.Required(CONF_TEXT): cv.string, - }, - ), - supports_response=SupportsResponse.ONLY, - ) - # Reload entry when its updated. entry.async_on_unload(entry.add_update_listener(async_reload_entry)) diff --git a/homeassistant/components/system_bridge/services.py b/homeassistant/components/system_bridge/services.py new file mode 100644 index 0000000000000..2eaa3bce35bc9 --- /dev/null +++ b/homeassistant/components/system_bridge/services.py @@ -0,0 +1,269 @@ +"""Service registration for System Bridge integration.""" + +from dataclasses import asdict +import logging +from typing import Any + +from systembridgeconnector.models.keyboard_key import KeyboardKey +from systembridgeconnector.models.keyboard_text import KeyboardText +from systembridgeconnector.models.modules.processes import Process +from systembridgeconnector.models.open_path import OpenPath +from systembridgeconnector.models.open_url import OpenUrl +import voluptuous as vol + +from homeassistant.const import CONF_COMMAND, CONF_ID, CONF_NAME, CONF_PATH, CONF_URL +from homeassistant.core import ( + HomeAssistant, + ServiceCall, + ServiceResponse, + SupportsResponse, + callback, +) +from homeassistant.exceptions import ServiceValidationError +from homeassistant.helpers import ( + config_validation as cv, + device_registry as dr, + service, +) + +from .const import DOMAIN +from .coordinator import SystemBridgeConfigEntry, SystemBridgeDataUpdateCoordinator + +_LOGGER = logging.getLogger(__name__) + +CONF_BRIDGE = "bridge" +CONF_KEY = "key" +CONF_TEXT = "text" + +POWER_COMMAND_MAP = { + "hibernate": "power_hibernate", + "lock": "power_lock", + "logout": "power_logout", + "restart": "power_restart", + "shutdown": "power_shutdown", + "sleep": "power_sleep", +} + + +@callback +def async_setup_services(hass: HomeAssistant) -> None: + """Set up services for System Bridge integration.""" + + hass.services.async_register( + DOMAIN, + "get_process_by_id", + handle_get_process_by_id, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_ID): cv.positive_int, + }, + ), + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + "get_processes_by_name", + handle_get_processes_by_name, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_NAME): cv.string, + }, + ), + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + "open_path", + handle_open_path, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_PATH): cv.string, + }, + ), + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + "power_command", + handle_power_command, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_COMMAND): vol.In(POWER_COMMAND_MAP), + }, + ), + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + "open_url", + handle_open_url, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_URL): cv.string, + }, + ), + supports_response=SupportsResponse.ONLY, + ) + + hass.services.async_register( + DOMAIN, + "send_keypress", + handle_send_keypress, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_KEY): cv.string, + }, + ), + supports_response=SupportsResponse.ONLY, + description_placeholders={ + "syntax_keys_documentation_url": "https://robotjs.dev/docs/syntax#keys" + }, + ) + + hass.services.async_register( + DOMAIN, + "send_text", + handle_send_text, + schema=vol.Schema( + { + vol.Required(CONF_BRIDGE): cv.string, + vol.Required(CONF_TEXT): cv.string, + }, + ), + supports_response=SupportsResponse.ONLY, + ) + + +def _get_coordinator( + hass: HomeAssistant, device_id: str +) -> SystemBridgeDataUpdateCoordinator: + """Return the coordinator for a device id.""" + + device_registry = dr.async_get(hass) + device_entry = device_registry.async_get(device_id) + + if device_entry is None: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="device_not_found", + translation_placeholders={"device": device_id}, + ) + try: + entry_id = next( + entry.entry_id + for entry in hass.config_entries.async_entries(DOMAIN) + if entry.entry_id in device_entry.config_entries + ) + except StopIteration as e: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="device_not_found", + translation_placeholders={"device": device_id}, + ) from e + entry: SystemBridgeConfigEntry = service.async_get_config_entry( + hass, DOMAIN, entry_id + ) + return entry.runtime_data + + +async def handle_get_process_by_id(service_call: ServiceCall) -> ServiceResponse: + """Handle the get process by id service call.""" + _LOGGER.debug("Get process by id: %s", service_call.data) + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + processes: list[Process] = coordinator.data.processes + + # Find process.id from list, raise ServiceValidationError if not found + try: + return asdict( + next( + process + for process in processes + if process.id == service_call.data[CONF_ID] + ) + ) + except StopIteration as e: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="process_not_found", + translation_placeholders={"id": service_call.data[CONF_ID]}, + ) from e + + +async def handle_get_processes_by_name( + service_call: ServiceCall, +) -> ServiceResponse: + """Handle the get process by name service call.""" + _LOGGER.debug("Get process by name: %s", service_call.data) + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + + # Find processes from list + items: list[dict[str, Any]] = [ + asdict(process) + for process in coordinator.data.processes + if process.name is not None + and service_call.data[CONF_NAME].lower() in process.name.lower() + ] + + return { + "count": len(items), + "processes": list(items), + } + + +async def handle_open_path(service_call: ServiceCall) -> ServiceResponse: + """Handle the open path service call.""" + _LOGGER.debug("Open path: %s", service_call.data) + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + response = await coordinator.websocket_client.open_path( + OpenPath(path=service_call.data[CONF_PATH]) + ) + return asdict(response) + + +async def handle_power_command(service_call: ServiceCall) -> ServiceResponse: + """Handle the power command service call.""" + _LOGGER.debug("Power command: %s", service_call.data) + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + response = await getattr( + coordinator.websocket_client, + POWER_COMMAND_MAP[service_call.data[CONF_COMMAND]], + )() + return asdict(response) + + +async def handle_open_url(service_call: ServiceCall) -> ServiceResponse: + """Handle the open url service call.""" + _LOGGER.debug("Open URL: %s", service_call.data) + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + response = await coordinator.websocket_client.open_url( + OpenUrl(url=service_call.data[CONF_URL]) + ) + return asdict(response) + + +async def handle_send_keypress(service_call: ServiceCall) -> ServiceResponse: + """Handle the send_keypress service call.""" + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + response = await coordinator.websocket_client.keyboard_keypress( + KeyboardKey(key=service_call.data[CONF_KEY]) + ) + return asdict(response) + + +async def handle_send_text(service_call: ServiceCall) -> ServiceResponse: + """Handle the send_text service call.""" + coordinator = _get_coordinator(service_call.hass, service_call.data[CONF_BRIDGE]) + response = await coordinator.websocket_client.keyboard_text( + KeyboardText(text=service_call.data[CONF_TEXT]) + ) + return asdict(response) diff --git a/tests/components/system_bridge/conftest.py b/tests/components/system_bridge/conftest.py index 37933803a738d..2d15134dc79d7 100644 --- a/tests/components/system_bridge/conftest.py +++ b/tests/components/system_bridge/conftest.py @@ -126,6 +126,36 @@ def mock_websocket_client( message="Data listener registered", data={EventKey.MODULES: register_data_listener_model.modules}, ) + websocket_client.open_url.return_value = Response( + id=FIXTURE_REQUEST_ID, + type=EventType.OPENED, + message="Opened url", + data={"url": "https://example.com"}, + ) + websocket_client.open_path.return_value = Response( + id=FIXTURE_REQUEST_ID, + type=EventType.OPENED, + message="Opened file", + data={"path": "/home/user/documents"}, + ) + websocket_client.power_shutdown.return_value = Response( + id=FIXTURE_REQUEST_ID, + type=EventType.POWER_SHUTDOWN, + message="Shutdown", + data={}, + ) + websocket_client.keyboard_keypress.return_value = Response( + id=FIXTURE_REQUEST_ID, + type=EventType.KEYBOARD_KEY_PRESSED, + message="Keyboard key pressed", + data={"key": "backspace"}, + ) + websocket_client.keyboard_text.return_value = Response( + id=FIXTURE_REQUEST_ID, + type=EventType.KEYBOARD_TEXT_SENT, + message="Keyboard text sent", + data={"text": "Hello world"}, + ) # Trigger callback when listener is registered websocket_client.listen.side_effect = mock_data_listener diff --git a/tests/components/system_bridge/snapshots/test_services.ambr b/tests/components/system_bridge/snapshots/test_services.ambr new file mode 100644 index 0000000000000..60059569d7fb1 --- /dev/null +++ b/tests/components/system_bridge/snapshots/test_services.ambr @@ -0,0 +1,91 @@ +# serializer version: 1 +# name: test_get_process_services[get_process_by_id] + dict({ + 'cpu_usage': 12.3, + 'created': 12.3, + 'id': 1234, + 'memory_usage': 12.3, + 'name': 'name', + 'path': '/path', + 'status': 'running', + 'username': 'username', + 'working_directory': '/working/directory', + }) +# --- +# name: test_get_process_services[get_processes_by_name] + dict({ + 'count': 1, + 'processes': list([ + dict({ + 'cpu_usage': 12.3, + 'created': 12.3, + 'id': 1234, + 'memory_usage': 12.3, + 'name': 'name', + 'path': '/path', + 'status': 'running', + 'username': 'username', + 'working_directory': '/working/directory', + }), + ]), + }) +# --- +# name: test_services[open_path] + dict({ + 'data': dict({ + 'path': '/home/user/documents', + }), + 'id': 'test', + 'message': 'Opened file', + 'module': None, + 'subtype': None, + 'type': , + }) +# --- +# name: test_services[open_url] + dict({ + 'data': dict({ + 'url': 'https://example.com', + }), + 'id': 'test', + 'message': 'Opened url', + 'module': None, + 'subtype': None, + 'type': , + }) +# --- +# name: test_services[power_command_shutdown] + dict({ + 'data': dict({ + }), + 'id': 'test', + 'message': 'Shutdown', + 'module': None, + 'subtype': None, + 'type': , + }) +# --- +# name: test_services[send_keypress] + dict({ + 'data': dict({ + 'key': 'backspace', + }), + 'id': 'test', + 'message': 'Keyboard key pressed', + 'module': None, + 'subtype': None, + 'type': , + }) +# --- +# name: test_services[send_text] + dict({ + 'data': dict({ + 'text': 'Hello world', + }), + 'id': 'test', + 'message': 'Keyboard text sent', + 'module': None, + 'subtype': None, + 'type': , + }) +# --- diff --git a/tests/components/system_bridge/test_services.py b/tests/components/system_bridge/test_services.py new file mode 100644 index 0000000000000..ac506201e8c02 --- /dev/null +++ b/tests/components/system_bridge/test_services.py @@ -0,0 +1,155 @@ +"""Tests for System Bridge actions.""" + +from typing import Any + +import pytest +from syrupy.assertion import SnapshotAssertion +from systembridgeconnector.models.keyboard_key import KeyboardKey +from systembridgeconnector.models.keyboard_text import KeyboardText +from systembridgeconnector.models.open_path import OpenPath +from systembridgeconnector.models.open_url import OpenUrl + +from homeassistant.components.system_bridge.const import DOMAIN +from homeassistant.components.system_bridge.services import ( + CONF_BRIDGE, + CONF_KEY, + CONF_TEXT, +) +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import CONF_COMMAND, CONF_ID, CONF_NAME, CONF_PATH, CONF_URL +from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr + +from . import FIXTURE_UUID + +from tests.common import AsyncMock, MockConfigEntry + + +@pytest.mark.parametrize( + ("service", "service_data", "call_method", "call_args"), + [ + ( + "open_path", + {CONF_PATH: "/home/user/documents"}, + "open_path", + [OpenPath(path="/home/user/documents")], + ), + ( + "open_url", + {CONF_URL: "https://example.com"}, + "open_url", + [OpenUrl(url="https://example.com")], + ), + ( + "power_command", + {CONF_COMMAND: "shutdown"}, + "power_shutdown", + [], + ), + ( + "send_keypress", + {CONF_KEY: "backspace"}, + "keyboard_keypress", + [KeyboardKey(key="backspace")], + ), + ( + "send_text", + {CONF_TEXT: "Hello world"}, + "keyboard_text", + [KeyboardText(text="Hello world")], + ), + ], + ids=[ + "open_path", + "open_url", + "power_command_shutdown", + "send_keypress", + "send_text", + ], +) +@pytest.mark.usefixtures("mock_version") +async def test_services( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_websocket_client: AsyncMock, + snapshot: SnapshotAssertion, + device_registry: dr.DeviceRegistry, + service: str, + service_data: dict[str, Any], + call_method: str, + call_args: list[Any], +) -> None: + """Test System Bridge service action calls.""" + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.LOADED + + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, FIXTURE_UUID)} + ) + assert device_entry + + resp = await hass.services.async_call( + DOMAIN, + service, + { + CONF_BRIDGE: device_entry.id, + **service_data, + }, + blocking=True, + return_response=True, + ) + + getattr(mock_websocket_client, call_method).assert_awaited_once_with(*call_args) + assert resp == snapshot + + +@pytest.mark.parametrize( + ("service", "service_data"), + [ + ( + "get_process_by_id", + {CONF_ID: 1234}, + ), + ( + "get_processes_by_name", + {CONF_NAME: "name"}, + ), + ], + ids=["get_process_by_id", "get_processes_by_name"], +) +@pytest.mark.usefixtures("mock_version", "mock_websocket_client") +async def test_get_process_services( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + snapshot: SnapshotAssertion, + device_registry: dr.DeviceRegistry, + service: str, + service_data: dict[str, Any], +) -> None: + """Test System Bridge get process service action calls.""" + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.LOADED + + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, FIXTURE_UUID)} + ) + assert device_entry + + resp = await hass.services.async_call( + DOMAIN, + service, + { + CONF_BRIDGE: device_entry.id, + **service_data, + }, + blocking=True, + return_response=True, + ) + + assert resp == snapshot