mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 11:17:21 +00:00
parent
eefc6cd50a
commit
ae70729932
@ -1,25 +1,12 @@
|
|||||||
"""Init the tedee component."""
|
"""Init the tedee component."""
|
||||||
from collections.abc import Awaitable, Callable
|
|
||||||
from http import HTTPStatus
|
|
||||||
import logging
|
import logging
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
from aiohttp.hdrs import METH_POST
|
|
||||||
from aiohttp.web import Request, Response
|
|
||||||
from pytedee_async.exception import TedeeWebhookException
|
|
||||||
|
|
||||||
from homeassistant.components.http import HomeAssistantView
|
|
||||||
from homeassistant.components.webhook import (
|
|
||||||
async_generate_url as webhook_generate_url,
|
|
||||||
async_register as webhook_register,
|
|
||||||
async_unregister as webhook_unregister,
|
|
||||||
)
|
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import CONF_WEBHOOK_ID, EVENT_HOMEASSISTANT_STOP, Platform
|
from homeassistant.const import Platform
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers import device_registry as dr
|
from homeassistant.helpers import device_registry as dr
|
||||||
|
|
||||||
from .const import DOMAIN, NAME
|
from .const import DOMAIN
|
||||||
from .coordinator import TedeeApiCoordinator
|
from .coordinator import TedeeApiCoordinator
|
||||||
|
|
||||||
PLATFORMS = [
|
PLATFORMS = [
|
||||||
@ -50,38 +37,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||||||
|
|
||||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
|
||||||
|
|
||||||
async def unregister_webhook(_: Any) -> None:
|
|
||||||
await coordinator.async_unregister_webhook()
|
|
||||||
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
|
|
||||||
|
|
||||||
async def register_webhook() -> None:
|
|
||||||
webhook_url = webhook_generate_url(hass, entry.data[CONF_WEBHOOK_ID])
|
|
||||||
webhook_name = "Tedee"
|
|
||||||
if entry.title != NAME:
|
|
||||||
webhook_name = f"{NAME} {entry.title}"
|
|
||||||
|
|
||||||
webhook_register(
|
|
||||||
hass,
|
|
||||||
DOMAIN,
|
|
||||||
webhook_name,
|
|
||||||
entry.data[CONF_WEBHOOK_ID],
|
|
||||||
get_webhook_handler(coordinator),
|
|
||||||
allowed_methods=[METH_POST],
|
|
||||||
)
|
|
||||||
_LOGGER.debug("Registered Tedee webhook at hass: %s", webhook_url)
|
|
||||||
|
|
||||||
try:
|
|
||||||
await coordinator.async_register_webhook(webhook_url)
|
|
||||||
except TedeeWebhookException as ex:
|
|
||||||
_LOGGER.warning("Failed to register Tedee webhook from bridge: %s", ex)
|
|
||||||
else:
|
|
||||||
entry.async_on_unload(
|
|
||||||
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, unregister_webhook)
|
|
||||||
)
|
|
||||||
|
|
||||||
entry.async_create_background_task(
|
|
||||||
hass, register_webhook(), "tedee_register_webhook"
|
|
||||||
)
|
|
||||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||||
|
|
||||||
return True
|
return True
|
||||||
@ -90,34 +45,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||||
"""Unload a config entry."""
|
"""Unload a config entry."""
|
||||||
|
|
||||||
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
|
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||||
|
|
||||||
|
if unload_ok:
|
||||||
hass.data[DOMAIN].pop(entry.entry_id)
|
hass.data[DOMAIN].pop(entry.entry_id)
|
||||||
|
|
||||||
return unload_ok
|
return unload_ok
|
||||||
|
|
||||||
|
|
||||||
def get_webhook_handler(
|
|
||||||
coordinator: TedeeApiCoordinator,
|
|
||||||
) -> Callable[[HomeAssistant, str, Request], Awaitable[Response | None]]:
|
|
||||||
"""Return webhook handler."""
|
|
||||||
|
|
||||||
async def async_webhook_handler(
|
|
||||||
hass: HomeAssistant, webhook_id: str, request: Request
|
|
||||||
) -> Response | None:
|
|
||||||
# Handle http post calls to the path.
|
|
||||||
if not request.body_exists:
|
|
||||||
return HomeAssistantView.json(
|
|
||||||
result="No Body", status_code=HTTPStatus.BAD_REQUEST
|
|
||||||
)
|
|
||||||
|
|
||||||
body = await request.json()
|
|
||||||
try:
|
|
||||||
coordinator.webhook_received(body)
|
|
||||||
except TedeeWebhookException as ex:
|
|
||||||
return HomeAssistantView.json(
|
|
||||||
result=str(ex), status_code=HTTPStatus.BAD_REQUEST
|
|
||||||
)
|
|
||||||
|
|
||||||
return HomeAssistantView.json(result="OK", status_code=HTTPStatus.OK)
|
|
||||||
|
|
||||||
return async_webhook_handler
|
|
||||||
|
@ -11,9 +11,8 @@ from pytedee_async import (
|
|||||||
)
|
)
|
||||||
import voluptuous as vol
|
import voluptuous as vol
|
||||||
|
|
||||||
from homeassistant.components.webhook import async_generate_id as webhook_generate_id
|
|
||||||
from homeassistant.config_entries import ConfigEntry, ConfigFlow
|
from homeassistant.config_entries import ConfigEntry, ConfigFlow
|
||||||
from homeassistant.const import CONF_HOST, CONF_WEBHOOK_ID
|
from homeassistant.const import CONF_HOST
|
||||||
from homeassistant.data_entry_flow import FlowResult
|
from homeassistant.data_entry_flow import FlowResult
|
||||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||||
|
|
||||||
@ -62,10 +61,7 @@ class TedeeConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||||||
return self.async_abort(reason="reauth_successful")
|
return self.async_abort(reason="reauth_successful")
|
||||||
await self.async_set_unique_id(local_bridge.serial)
|
await self.async_set_unique_id(local_bridge.serial)
|
||||||
self._abort_if_unique_id_configured()
|
self._abort_if_unique_id_configured()
|
||||||
return self.async_create_entry(
|
return self.async_create_entry(title=NAME, data=user_input)
|
||||||
title=NAME,
|
|
||||||
data={**user_input, CONF_WEBHOOK_ID: webhook_generate_id()},
|
|
||||||
)
|
|
||||||
|
|
||||||
return self.async_show_form(
|
return self.async_show_form(
|
||||||
step_id="user",
|
step_id="user",
|
||||||
|
@ -3,7 +3,6 @@ from collections.abc import Awaitable, Callable
|
|||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
from pytedee_async import (
|
from pytedee_async import (
|
||||||
TedeeClient,
|
TedeeClient,
|
||||||
@ -11,7 +10,6 @@ from pytedee_async import (
|
|||||||
TedeeDataUpdateException,
|
TedeeDataUpdateException,
|
||||||
TedeeLocalAuthException,
|
TedeeLocalAuthException,
|
||||||
TedeeLock,
|
TedeeLock,
|
||||||
TedeeWebhookException,
|
|
||||||
)
|
)
|
||||||
from pytedee_async.bridge import TedeeBridge
|
from pytedee_async.bridge import TedeeBridge
|
||||||
|
|
||||||
@ -25,7 +23,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
|
|||||||
|
|
||||||
from .const import CONF_LOCAL_ACCESS_TOKEN, DOMAIN
|
from .const import CONF_LOCAL_ACCESS_TOKEN, DOMAIN
|
||||||
|
|
||||||
SCAN_INTERVAL = timedelta(seconds=30)
|
SCAN_INTERVAL = timedelta(seconds=20)
|
||||||
GET_LOCKS_INTERVAL_SECONDS = 3600
|
GET_LOCKS_INTERVAL_SECONDS = 3600
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
@ -55,7 +53,6 @@ class TedeeApiCoordinator(DataUpdateCoordinator[dict[int, TedeeLock]]):
|
|||||||
self._next_get_locks = time.time()
|
self._next_get_locks = time.time()
|
||||||
self._locks_last_update: set[int] = set()
|
self._locks_last_update: set[int] = set()
|
||||||
self.new_lock_callbacks: list[Callable[[int], None]] = []
|
self.new_lock_callbacks: list[Callable[[int], None]] = []
|
||||||
self.tedee_webhook_id: int | None = None
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def bridge(self) -> TedeeBridge:
|
def bridge(self) -> TedeeBridge:
|
||||||
@ -106,27 +103,6 @@ class TedeeApiCoordinator(DataUpdateCoordinator[dict[int, TedeeLock]]):
|
|||||||
except (TedeeClientException, TimeoutError) as ex:
|
except (TedeeClientException, TimeoutError) as ex:
|
||||||
raise UpdateFailed("Querying API failed. Error: %s" % str(ex)) from ex
|
raise UpdateFailed("Querying API failed. Error: %s" % str(ex)) from ex
|
||||||
|
|
||||||
def webhook_received(self, message: dict[str, Any]) -> None:
|
|
||||||
"""Handle webhook message."""
|
|
||||||
self.tedee_client.parse_webhook_message(message)
|
|
||||||
self.async_set_updated_data(self.tedee_client.locks_dict)
|
|
||||||
|
|
||||||
async def async_register_webhook(self, webhook_url: str) -> None:
|
|
||||||
"""Register the webhook at the Tedee bridge."""
|
|
||||||
self.tedee_webhook_id = await self.tedee_client.register_webhook(webhook_url)
|
|
||||||
|
|
||||||
async def async_unregister_webhook(self) -> None:
|
|
||||||
"""Unregister the webhook at the Tedee bridge."""
|
|
||||||
if self.tedee_webhook_id is not None:
|
|
||||||
try:
|
|
||||||
await self.tedee_client.delete_webhook(self.tedee_webhook_id)
|
|
||||||
except TedeeWebhookException as ex:
|
|
||||||
_LOGGER.warning(
|
|
||||||
"Failed to unregister Tedee webhook from bridge: %s", ex
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
_LOGGER.debug("Unregistered Tedee webhook")
|
|
||||||
|
|
||||||
def _async_add_remove_locks(self) -> None:
|
def _async_add_remove_locks(self) -> None:
|
||||||
"""Add new locks, remove non-existing locks."""
|
"""Add new locks, remove non-existing locks."""
|
||||||
if not self._locks_last_update:
|
if not self._locks_last_update:
|
||||||
|
@ -3,7 +3,7 @@
|
|||||||
"name": "Tedee",
|
"name": "Tedee",
|
||||||
"codeowners": ["@patrickhilker", "@zweckj"],
|
"codeowners": ["@patrickhilker", "@zweckj"],
|
||||||
"config_flow": true,
|
"config_flow": true,
|
||||||
"dependencies": ["http", "webhook"],
|
"dependencies": ["http"],
|
||||||
"documentation": "https://www.home-assistant.io/integrations/tedee",
|
"documentation": "https://www.home-assistant.io/integrations/tedee",
|
||||||
"iot_class": "local_push",
|
"iot_class": "local_push",
|
||||||
"requirements": ["pytedee-async==0.2.13"]
|
"requirements": ["pytedee-async==0.2.13"]
|
||||||
|
@ -10,13 +10,11 @@ from pytedee_async.lock import TedeeLock
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from homeassistant.components.tedee.const import CONF_LOCAL_ACCESS_TOKEN, DOMAIN
|
from homeassistant.components.tedee.const import CONF_LOCAL_ACCESS_TOKEN, DOMAIN
|
||||||
from homeassistant.const import CONF_HOST, CONF_WEBHOOK_ID
|
from homeassistant.const import CONF_HOST
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
from tests.common import MockConfigEntry, load_fixture
|
from tests.common import MockConfigEntry, load_fixture
|
||||||
|
|
||||||
WEBHOOK_ID = "bq33efxmdi3vxy55q2wbnudbra7iv8mjrq9x0gea33g4zqtd87093pwveg8xcb33"
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def mock_config_entry() -> MockConfigEntry:
|
def mock_config_entry() -> MockConfigEntry:
|
||||||
@ -27,7 +25,6 @@ def mock_config_entry() -> MockConfigEntry:
|
|||||||
data={
|
data={
|
||||||
CONF_LOCAL_ACCESS_TOKEN: "api_token",
|
CONF_LOCAL_ACCESS_TOKEN: "api_token",
|
||||||
CONF_HOST: "192.168.1.42",
|
CONF_HOST: "192.168.1.42",
|
||||||
CONF_WEBHOOK_ID: WEBHOOK_ID,
|
|
||||||
},
|
},
|
||||||
unique_id="0000-0000",
|
unique_id="0000-0000",
|
||||||
)
|
)
|
||||||
@ -62,8 +59,6 @@ def mock_tedee(request) -> Generator[MagicMock, None, None]:
|
|||||||
tedee.get_local_bridge.return_value = TedeeBridge(0, "0000-0000", "Bridge-AB1C")
|
tedee.get_local_bridge.return_value = TedeeBridge(0, "0000-0000", "Bridge-AB1C")
|
||||||
|
|
||||||
tedee.parse_webhook_message.return_value = None
|
tedee.parse_webhook_message.return_value = None
|
||||||
tedee.register_webhook.return_value = 1
|
|
||||||
tedee.delete_webhooks.return_value = None
|
|
||||||
|
|
||||||
locks_json = json.loads(load_fixture("locks.json", DOMAIN))
|
locks_json = json.loads(load_fixture("locks.json", DOMAIN))
|
||||||
|
|
||||||
|
@ -1,5 +1,5 @@
|
|||||||
"""Test the Tedee config flow."""
|
"""Test the Tedee config flow."""
|
||||||
from unittest.mock import MagicMock, patch
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
from pytedee_async import (
|
from pytedee_async import (
|
||||||
TedeeClientException,
|
TedeeClientException,
|
||||||
@ -10,12 +10,10 @@ import pytest
|
|||||||
|
|
||||||
from homeassistant.components.tedee.const import CONF_LOCAL_ACCESS_TOKEN, DOMAIN
|
from homeassistant.components.tedee.const import CONF_LOCAL_ACCESS_TOKEN, DOMAIN
|
||||||
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
|
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
|
||||||
from homeassistant.const import CONF_HOST, CONF_WEBHOOK_ID
|
from homeassistant.const import CONF_HOST
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.data_entry_flow import FlowResultType
|
from homeassistant.data_entry_flow import FlowResultType
|
||||||
|
|
||||||
from .conftest import WEBHOOK_ID
|
|
||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
|
|
||||||
FLOW_UNIQUE_ID = "112233445566778899"
|
FLOW_UNIQUE_ID = "112233445566778899"
|
||||||
@ -24,30 +22,25 @@ LOCAL_ACCESS_TOKEN = "api_token"
|
|||||||
|
|
||||||
async def test_flow(hass: HomeAssistant, mock_tedee: MagicMock) -> None:
|
async def test_flow(hass: HomeAssistant, mock_tedee: MagicMock) -> None:
|
||||||
"""Test config flow with one bridge."""
|
"""Test config flow with one bridge."""
|
||||||
with patch(
|
result = await hass.config_entries.flow.async_init(
|
||||||
"homeassistant.components.tedee.config_flow.webhook_generate_id",
|
DOMAIN, context={"source": SOURCE_USER}
|
||||||
return_value=WEBHOOK_ID,
|
)
|
||||||
):
|
await hass.async_block_till_done()
|
||||||
result = await hass.config_entries.flow.async_init(
|
assert result["type"] == FlowResultType.FORM
|
||||||
DOMAIN, context={"source": SOURCE_USER}
|
|
||||||
)
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
assert result["type"] == FlowResultType.FORM
|
|
||||||
|
|
||||||
result2 = await hass.config_entries.flow.async_configure(
|
result2 = await hass.config_entries.flow.async_configure(
|
||||||
result["flow_id"],
|
result["flow_id"],
|
||||||
{
|
{
|
||||||
CONF_HOST: "192.168.1.62",
|
|
||||||
CONF_LOCAL_ACCESS_TOKEN: "token",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
assert result2["type"] == FlowResultType.CREATE_ENTRY
|
|
||||||
assert result2["data"] == {
|
|
||||||
CONF_HOST: "192.168.1.62",
|
CONF_HOST: "192.168.1.62",
|
||||||
CONF_LOCAL_ACCESS_TOKEN: "token",
|
CONF_LOCAL_ACCESS_TOKEN: "token",
|
||||||
CONF_WEBHOOK_ID: WEBHOOK_ID,
|
},
|
||||||
}
|
)
|
||||||
|
|
||||||
|
assert result2["type"] == FlowResultType.CREATE_ENTRY
|
||||||
|
assert result2["data"] == {
|
||||||
|
CONF_HOST: "192.168.1.62",
|
||||||
|
CONF_LOCAL_ACCESS_TOKEN: "token",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
async def test_flow_already_configured(
|
async def test_flow_already_configured(
|
||||||
|
@ -1,27 +1,15 @@
|
|||||||
"""Test initialization of tedee."""
|
"""Test initialization of tedee."""
|
||||||
from http import HTTPStatus
|
|
||||||
from typing import Any
|
|
||||||
from unittest.mock import MagicMock
|
from unittest.mock import MagicMock
|
||||||
from urllib.parse import urlparse
|
|
||||||
|
|
||||||
from pytedee_async.exception import (
|
from pytedee_async.exception import TedeeAuthException, TedeeClientException
|
||||||
TedeeAuthException,
|
|
||||||
TedeeClientException,
|
|
||||||
TedeeWebhookException,
|
|
||||||
)
|
|
||||||
import pytest
|
import pytest
|
||||||
from syrupy import SnapshotAssertion
|
from syrupy import SnapshotAssertion
|
||||||
|
|
||||||
from homeassistant.components.webhook import async_generate_url
|
|
||||||
from homeassistant.config_entries import ConfigEntryState
|
from homeassistant.config_entries import ConfigEntryState
|
||||||
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
|
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers import device_registry as dr
|
from homeassistant.helpers import device_registry as dr
|
||||||
|
|
||||||
from .conftest import WEBHOOK_ID
|
|
||||||
|
|
||||||
from tests.common import MockConfigEntry
|
from tests.common import MockConfigEntry
|
||||||
from tests.typing import ClientSessionGenerator
|
|
||||||
|
|
||||||
|
|
||||||
async def test_load_unload_config_entry(
|
async def test_load_unload_config_entry(
|
||||||
@ -62,62 +50,6 @@ async def test_config_entry_not_ready(
|
|||||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||||
|
|
||||||
|
|
||||||
async def test_cleanup_on_shutdown(
|
|
||||||
hass: HomeAssistant,
|
|
||||||
mock_config_entry: MockConfigEntry,
|
|
||||||
mock_tedee: MagicMock,
|
|
||||||
) -> None:
|
|
||||||
"""Test the webhook is cleaned up on shutdown."""
|
|
||||||
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
|
|
||||||
|
|
||||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
mock_tedee.delete_webhook.assert_called_once()
|
|
||||||
|
|
||||||
|
|
||||||
async def test_webhook_cleanup_errors(
|
|
||||||
hass: HomeAssistant,
|
|
||||||
mock_config_entry: MockConfigEntry,
|
|
||||||
mock_tedee: MagicMock,
|
|
||||||
caplog: pytest.LogCaptureFixture,
|
|
||||||
) -> None:
|
|
||||||
"""Test the webhook is cleaned up on shutdown."""
|
|
||||||
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
|
|
||||||
|
|
||||||
mock_tedee.delete_webhook.side_effect = TedeeWebhookException("")
|
|
||||||
|
|
||||||
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
mock_tedee.delete_webhook.assert_called_once()
|
|
||||||
assert "Failed to unregister Tedee webhook from bridge" in caplog.text
|
|
||||||
|
|
||||||
|
|
||||||
async def test_webhook_registration_errors(
|
|
||||||
hass: HomeAssistant,
|
|
||||||
mock_config_entry: MockConfigEntry,
|
|
||||||
mock_tedee: MagicMock,
|
|
||||||
caplog: pytest.LogCaptureFixture,
|
|
||||||
) -> None:
|
|
||||||
"""Test the webhook is cleaned up on shutdown."""
|
|
||||||
mock_tedee.register_webhook.side_effect = TedeeWebhookException("")
|
|
||||||
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
|
|
||||||
|
|
||||||
mock_tedee.register_webhook.assert_called_once()
|
|
||||||
assert "Failed to register Tedee webhook from bridge" in caplog.text
|
|
||||||
|
|
||||||
|
|
||||||
async def test_bridge_device(
|
async def test_bridge_device(
|
||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
mock_config_entry: MockConfigEntry,
|
mock_config_entry: MockConfigEntry,
|
||||||
@ -135,37 +67,3 @@ async def test_bridge_device(
|
|||||||
)
|
)
|
||||||
assert device
|
assert device
|
||||||
assert device == snapshot
|
assert device == snapshot
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
|
||||||
("body", "expected_code", "side_effect"),
|
|
||||||
[
|
|
||||||
({"hello": "world"}, HTTPStatus.OK, None), # Success
|
|
||||||
(None, HTTPStatus.BAD_REQUEST, None), # Missing data
|
|
||||||
({}, HTTPStatus.BAD_REQUEST, TedeeWebhookException), # Error
|
|
||||||
],
|
|
||||||
)
|
|
||||||
async def test_webhook_post(
|
|
||||||
hass: HomeAssistant,
|
|
||||||
mock_config_entry: MockConfigEntry,
|
|
||||||
mock_tedee: MagicMock,
|
|
||||||
hass_client_no_auth: ClientSessionGenerator,
|
|
||||||
body: dict[str, Any],
|
|
||||||
expected_code: HTTPStatus,
|
|
||||||
side_effect: Exception,
|
|
||||||
) -> None:
|
|
||||||
"""Test webhook callback."""
|
|
||||||
|
|
||||||
mock_config_entry.add_to_hass(hass)
|
|
||||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
client = await hass_client_no_auth()
|
|
||||||
webhook_url = async_generate_url(hass, WEBHOOK_ID)
|
|
||||||
mock_tedee.parse_webhook_message.side_effect = side_effect
|
|
||||||
resp = await client.post(urlparse(webhook_url).path, json=body)
|
|
||||||
|
|
||||||
# Wait for remaining tasks to complete.
|
|
||||||
await hass.async_block_till_done()
|
|
||||||
|
|
||||||
assert resp.status == expected_code
|
|
||||||
|
@ -1,10 +1,9 @@
|
|||||||
"""Tests for tedee lock."""
|
"""Tests for tedee lock."""
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
from unittest.mock import MagicMock
|
from unittest.mock import MagicMock
|
||||||
from urllib.parse import urlparse
|
|
||||||
|
|
||||||
from freezegun.api import FrozenDateTimeFactory
|
from freezegun.api import FrozenDateTimeFactory
|
||||||
from pytedee_async import TedeeLock, TedeeLockState
|
from pytedee_async import TedeeLock
|
||||||
from pytedee_async.exception import (
|
from pytedee_async.exception import (
|
||||||
TedeeClientException,
|
TedeeClientException,
|
||||||
TedeeDataUpdateException,
|
TedeeDataUpdateException,
|
||||||
@ -18,21 +17,15 @@ from homeassistant.components.lock import (
|
|||||||
SERVICE_LOCK,
|
SERVICE_LOCK,
|
||||||
SERVICE_OPEN,
|
SERVICE_OPEN,
|
||||||
SERVICE_UNLOCK,
|
SERVICE_UNLOCK,
|
||||||
STATE_LOCKED,
|
|
||||||
STATE_LOCKING,
|
STATE_LOCKING,
|
||||||
STATE_UNLOCKED,
|
|
||||||
STATE_UNLOCKING,
|
STATE_UNLOCKING,
|
||||||
)
|
)
|
||||||
from homeassistant.components.webhook import async_generate_url
|
|
||||||
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
|
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.exceptions import HomeAssistantError
|
from homeassistant.exceptions import HomeAssistantError
|
||||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||||
|
|
||||||
from .conftest import WEBHOOK_ID
|
|
||||||
|
|
||||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||||
from tests.typing import ClientSessionGenerator
|
|
||||||
|
|
||||||
pytestmark = pytest.mark.usefixtures("init_integration")
|
pytestmark = pytest.mark.usefixtures("init_integration")
|
||||||
|
|
||||||
@ -273,28 +266,3 @@ async def test_new_lock(
|
|||||||
assert state
|
assert state
|
||||||
state = hass.states.get("lock.lock_6g7h")
|
state = hass.states.get("lock.lock_6g7h")
|
||||||
assert state
|
assert state
|
||||||
|
|
||||||
|
|
||||||
async def test_webhook_update(
|
|
||||||
hass: HomeAssistant,
|
|
||||||
mock_tedee: MagicMock,
|
|
||||||
hass_client_no_auth: ClientSessionGenerator,
|
|
||||||
) -> None:
|
|
||||||
"""Test updated data set through webhook."""
|
|
||||||
|
|
||||||
state = hass.states.get("lock.lock_1a2b")
|
|
||||||
assert state
|
|
||||||
assert state.state == STATE_UNLOCKED
|
|
||||||
|
|
||||||
webhook_data = {"dummystate": 6}
|
|
||||||
mock_tedee.locks_dict[
|
|
||||||
12345
|
|
||||||
].state = TedeeLockState.LOCKED # is updated in the lib, so mock and assert in L296
|
|
||||||
client = await hass_client_no_auth()
|
|
||||||
webhook_url = async_generate_url(hass, WEBHOOK_ID)
|
|
||||||
await client.post(urlparse(webhook_url).path, json=webhook_data)
|
|
||||||
mock_tedee.parse_webhook_message.assert_called_once_with(webhook_data)
|
|
||||||
|
|
||||||
state = hass.states.get("lock.lock_1a2b")
|
|
||||||
assert state
|
|
||||||
assert state.state == STATE_LOCKED
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user