generated from ludeeus/integration_blueprint
-
Notifications
You must be signed in to change notification settings - Fork 107
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add a websocket API to the integration for event retain (#295)
* Add a new websocket API to the integration. * Pass in the data parameter.
- Loading branch information
1 parent
ef7f549
commit dcaa534
Showing
7 changed files
with
261 additions
and
26 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,67 @@ | ||
"""Frigate HTTP views.""" | ||
from __future__ import annotations | ||
|
||
import logging | ||
|
||
import voluptuous as vol | ||
|
||
from custom_components.frigate.api import FrigateApiClient, FrigateApiClientError | ||
from custom_components.frigate.views import get_client_for_frigate_instance_id | ||
from homeassistant.components import websocket_api | ||
from homeassistant.core import HomeAssistant | ||
|
||
_LOGGER: logging.Logger = logging.getLogger(__name__) | ||
|
||
|
||
def async_setup(hass: HomeAssistant) -> None: | ||
"""Set up the recorder websocket API.""" | ||
websocket_api.async_register_command(hass, ws_retain_event) | ||
|
||
|
||
def _get_client_or_send_error( | ||
hass: HomeAssistant, | ||
instance_id: str, | ||
msg_id: int, | ||
connection: websocket_api.ActiveConnection, | ||
) -> FrigateApiClient | None: | ||
"""Get the API client or send an error that it cannot be found.""" | ||
client = get_client_for_frigate_instance_id(hass, instance_id) | ||
if client is None: | ||
connection.send_error( | ||
msg_id, | ||
websocket_api.const.ERR_NOT_FOUND, | ||
f"Unable to find Frigate instance with ID: {instance_id}", | ||
) | ||
return None | ||
return client | ||
|
||
|
||
@websocket_api.websocket_command( | ||
{ | ||
vol.Required("type"): "frigate/event/retain", | ||
vol.Required("instance_id"): str, | ||
vol.Required("event_id"): str, | ||
vol.Required("retain"): bool, | ||
} | ||
) # type: ignore[misc] | ||
@websocket_api.async_response # type: ignore[misc] | ||
async def ws_retain_event( | ||
hass: HomeAssistant, | ||
connection: websocket_api.ActiveConnection, | ||
msg: dict, | ||
) -> None: | ||
"""Un/Retain an event.""" | ||
client = _get_client_or_send_error(hass, msg["instance_id"], msg["id"], connection) | ||
if not client: | ||
return | ||
try: | ||
connection.send_result( | ||
msg["id"], await client.async_retain(msg["event_id"], msg["retain"]) | ||
) | ||
except FrigateApiClientError: | ||
connection.send_error( | ||
msg["id"], | ||
"frigate_error", | ||
f"API error whilst un/retaining event {msg['event_id']} " | ||
f"for Frigate instance {msg['instance_id']}", | ||
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,121 @@ | ||
"""Test the Frigate HA websocket API.""" | ||
from __future__ import annotations | ||
|
||
import logging | ||
from typing import Any | ||
from unittest.mock import AsyncMock | ||
|
||
from custom_components.frigate.api import FrigateApiClientError | ||
from homeassistant.core import HomeAssistant | ||
|
||
from tests import ( | ||
TEST_FRIGATE_INSTANCE_ID, | ||
create_mock_frigate_client, | ||
setup_mock_frigate_config_entry, | ||
) | ||
|
||
_LOGGER: logging.Logger = logging.getLogger(__name__) | ||
|
||
TEST_EVENT_ID = "1656282822.206673-bovnfg" | ||
|
||
|
||
async def test_retain_success(hass: HomeAssistant, hass_ws_client: Any) -> None: | ||
"""Test un/retaining an event.""" | ||
|
||
mock_client = create_mock_frigate_client() | ||
await setup_mock_frigate_config_entry(hass, client=mock_client) | ||
|
||
ws_client = await hass_ws_client() | ||
retain_json: dict[str, Any] = { | ||
"id": 1, | ||
"type": "frigate/event/retain", | ||
"instance_id": TEST_FRIGATE_INSTANCE_ID, | ||
"event_id": TEST_EVENT_ID, | ||
"retain": True, | ||
} | ||
|
||
retain_success = {"retain": "success"} | ||
mock_client.async_retain = AsyncMock(return_value=retain_success) | ||
await ws_client.send_json(retain_json) | ||
|
||
response = await ws_client.receive_json() | ||
mock_client.async_retain.assert_called_with(TEST_EVENT_ID, True) | ||
assert response["success"] | ||
assert response["result"] == retain_success | ||
|
||
unretain_success = {"unretain": "success"} | ||
mock_client.async_retain = AsyncMock(return_value=unretain_success) | ||
await ws_client.send_json( | ||
{ | ||
**retain_json, | ||
"id": 2, | ||
"retain": False, | ||
} | ||
) | ||
|
||
response = await ws_client.receive_json() | ||
mock_client.async_retain.assert_called_with(TEST_EVENT_ID, False) | ||
assert response["success"] | ||
assert response["result"] == unretain_success | ||
|
||
|
||
async def test_retain_missing_args(hass: HomeAssistant, hass_ws_client: Any) -> None: | ||
"""Test retaining an event with missing arguments.""" | ||
|
||
await setup_mock_frigate_config_entry(hass) | ||
|
||
ws_client = await hass_ws_client() | ||
retain_json = { | ||
"id": 1, | ||
"type": "frigate/event/retain", | ||
} | ||
|
||
await ws_client.send_json(retain_json) | ||
response = await ws_client.receive_json() | ||
assert not response["success"] | ||
assert response["error"]["code"] == "invalid_format" | ||
|
||
|
||
async def test_retain_instance_not_found( | ||
hass: HomeAssistant, hass_ws_client: Any | ||
) -> None: | ||
"""Test retaining an event with an instance that is not found.""" | ||
|
||
await setup_mock_frigate_config_entry(hass) | ||
|
||
ws_client = await hass_ws_client() | ||
retain_json = { | ||
"id": 1, | ||
"type": "frigate/event/retain", | ||
"instance_id": "THIS-IS-NOT-A-REAL-INSTANCE-ID", | ||
"event_id": TEST_EVENT_ID, | ||
"retain": True, | ||
} | ||
|
||
await ws_client.send_json(retain_json) | ||
response = await ws_client.receive_json() | ||
assert not response["success"] | ||
assert response["error"]["code"] == "not_found" | ||
|
||
|
||
async def test_retain_api_error(hass: HomeAssistant, hass_ws_client: Any) -> None: | ||
"""Test retaining an event when the API has an error.""" | ||
|
||
mock_client = create_mock_frigate_client() | ||
await setup_mock_frigate_config_entry(hass, client=mock_client) | ||
|
||
ws_client = await hass_ws_client() | ||
retain_json = { | ||
"id": 1, | ||
"type": "frigate/event/retain", | ||
"instance_id": TEST_FRIGATE_INSTANCE_ID, | ||
"event_id": TEST_EVENT_ID, | ||
"retain": True, | ||
} | ||
|
||
mock_client.async_retain = AsyncMock(side_effect=FrigateApiClientError) | ||
|
||
await ws_client.send_json(retain_json) | ||
response = await ws_client.receive_json() | ||
assert not response["success"] | ||
assert response["error"]["code"] == "frigate_error" |