evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use get_url to get Home Assistant instance for Loqed webhook (#95761)

Browse source
This commit is contained in:
Mike Woudenberg 2023-07-25 08:27:18 +02:00 • committed by GitHub
parent c312dcbc4b
commit 945fffebcc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 113 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

1
homeassistant/components/loqed/const.py
View file

@ -2,3 +2,4 @@
DOMAIN = "loqed"
CONF_CLOUDHOOK_URL = "cloudhook_url"

35
homeassistant/components/loqed/coordinator.py
View file

@ -6,13 +6,13 @@ from aiohttp.web import Request
import async_timeout
from loqedAPI import loqed
from homeassistant.components import webhook
from homeassistant.components import cloud, webhook
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, CONF_WEBHOOK_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .const import CONF_CLOUDHOOK_URL, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -114,7 +114,14 @@ class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
webhook.async_register(
self.hass, DOMAIN, "Loqed", webhook_id, self._handle_webhook
)
webhook_url = webhook.async_generate_url(self.hass, webhook_id)
if cloud.async_active_subscription(self.hass):
webhook_url = await async_cloudhook_generate_url(self.hass, self._entry)
else:
webhook_url = webhook.async_generate_url(
self.hass, self._entry.data[CONF_WEBHOOK_ID]
)
_LOGGER.debug("Webhook URL: %s", webhook_url)
webhooks = await self.lock.getWebhooks()
@ -128,18 +135,22 @@ class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
webhooks = await self.lock.getWebhooks()
webhook_index = next(x["id"] for x in webhooks if x["url"] == webhook_url)
_LOGGER.info("Webhook got index %s", webhook_index)
_LOGGER.debug("Webhook got index %s", webhook_index)
async def remove_webhooks(self) -> None:
"""Remove webhook from LOQED bridge."""
webhook_id = self._entry.data[CONF_WEBHOOK_ID]
webhook_url = webhook.async_generate_url(self.hass, webhook_id)
if CONF_CLOUDHOOK_URL in self._entry.data:
webhook_url = self._entry.data[CONF_CLOUDHOOK_URL]
else:
webhook_url = webhook.async_generate_url(self.hass, webhook_id)
webhook.async_unregister(
self.hass,
webhook_id,
)
_LOGGER.info("Webhook URL: %s", webhook_url)
_LOGGER.debug("Webhook URL: %s", webhook_url)
webhooks = await self.lock.getWebhooks()
@ -149,3 +160,15 @@ class LoqedDataCoordinator(DataUpdateCoordinator[StatusMessage]):
if webhook_index:
await self.lock.deleteWebhook(webhook_index)
async def async_cloudhook_generate_url(hass: HomeAssistant, entry: ConfigEntry) -> str:
"""Generate the full URL for a webhook_id."""
if CONF_CLOUDHOOK_URL not in entry.data:
webhook_url = await cloud.async_create_cloudhook(
hass, entry.data[CONF_WEBHOOK_ID]
)
data = {**entry.data, CONF_CLOUDHOOK_URL: webhook_url}
hass.config_entries.async_update_entry(entry, data=data)
return webhook_url
return str(entry.data[CONF_CLOUDHOOK_URL])

1
homeassistant/components/loqed/manifest.json
View file

@ -1,6 +1,7 @@
{
"domain": "loqed",
"name": "LOQED Touch Smart Lock",
"after_dependencies": ["cloud"],
"codeowners": ["@mikewoudenberg"],
"config_flow": true,
"dependencies": ["webhook"],

2
homeassistant/components/loqed/strings.json
View file

@ -6,7 +6,7 @@
"description": "Login at {config_url} and: \n* Create an API-key by clicking 'Create' \n* Copy the created access token.",
"data": {
"name": "Name of your lock in the LOQED app.",
"api_key": "[%key:common::config_flow::data::api_key%]"
"api_token": "[%key:common::config_flow::data::api_token%]"
}
}
},

29
tests/components/loqed/conftest.py
View file

@ -9,6 +9,7 @@ from loqedAPI import loqed
import pytest
from homeassistant.components.loqed import DOMAIN
from homeassistant.components.loqed.const import CONF_CLOUDHOOK_URL
from homeassistant.const import CONF_API_TOKEN, CONF_NAME, CONF_WEBHOOK_ID
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
@ -39,6 +40,31 @@ def config_entry_fixture() -> MockConfigEntry:
)
@pytest.fixture(name="cloud_config_entry")
def cloud_config_entry_fixture() -> MockConfigEntry:
"""Mock config entry."""
config = load_fixture("loqed/integration_config.json")
webhooks_fixture = json.loads(load_fixture("loqed/get_all_webhooks.json"))
json_config = json.loads(config)
return MockConfigEntry(
version=1,
domain=DOMAIN,
data={
"id": "Foo",
"bridge_ip": json_config["bridge_ip"],
"bridge_mdns_hostname": json_config["bridge_mdns_hostname"],
"bridge_key": json_config["bridge_key"],
"lock_key_local_id": int(json_config["lock_key_local_id"]),
"lock_key_key": json_config["lock_key_key"],
CONF_WEBHOOK_ID: "Webhook_id",
CONF_API_TOKEN: "Token",
CONF_NAME: "Home",
CONF_CLOUDHOOK_URL: webhooks_fixture[0]["url"],
},
)
@pytest.fixture(name="lock")
def lock_fixture() -> loqed.Lock:
"""Set up a mock implementation of a Lock."""
@ -64,9 +90,6 @@ async def integration_fixture(
with patch("loqedAPI.loqed.LoqedAPI.async_get_lock", return_value=lock), patch(
"loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_status
), patch(
"homeassistant.components.webhook.async_generate_url",
return_value="http://hook_id",
):
await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()

2
tests/components/loqed/fixtures/get_all_webhooks.json
View file

@ -1,7 +1,7 @@
[
{
"id": 1,
"url": "http://hook_id",
"url": "http://10.10.10.10:8123/api/webhook/Webhook_id",
"trigger_state_changed_open": 1,
"trigger_state_changed_latch": 1,
"trigger_state_changed_night_lock": 1,

58
tests/components/loqed/test_init.py
View file

@ -10,6 +10,7 @@ from homeassistant.components.loqed.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_WEBHOOK_ID
from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import get_url
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
@ -50,14 +51,63 @@ async def test_setup_webhook_in_bridge(
with patch("loqedAPI.loqed.LoqedAPI.async_get_lock", return_value=lock), patch(
"loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_status
), patch(
"homeassistant.components.webhook.async_generate_url",
return_value="http://hook_id",
):
await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
lock.registerWebhook.assert_called_with("http://hook_id")
lock.registerWebhook.assert_called_with(f"{get_url(hass)}/api/webhook/Webhook_id")
async def test_setup_cloudhook_in_bridge(
hass: HomeAssistant, config_entry: MockConfigEntry, lock: loqed.Lock
):
"""Test webhook setup in loqed bridge."""
config: dict[str, Any] = {DOMAIN: {}}
config_entry.add_to_hass(hass)
lock_status = json.loads(load_fixture("loqed/status_ok.json"))
webhooks_fixture = json.loads(load_fixture("loqed/get_all_webhooks.json"))
lock.getWebhooks = AsyncMock(side_effect=[[], webhooks_fixture])
with patch("loqedAPI.loqed.LoqedAPI.async_get_lock", return_value=lock), patch(
"loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_status
), patch(
"homeassistant.components.cloud.async_active_subscription", return_value=True
), patch(
"homeassistant.components.cloud.async_create_cloudhook",
return_value=webhooks_fixture[0]["url"],
):
await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
lock.registerWebhook.assert_called_with(f"{get_url(hass)}/api/webhook/Webhook_id")
async def test_setup_cloudhook_from_entry_in_bridge(
hass: HomeAssistant, cloud_config_entry: MockConfigEntry, lock: loqed.Lock
):
"""Test webhook setup in loqed bridge."""
webhooks_fixture = json.loads(load_fixture("loqed/get_all_webhooks.json"))
config: dict[str, Any] = {DOMAIN: {}}
cloud_config_entry.add_to_hass(hass)
lock_status = json.loads(load_fixture("loqed/status_ok.json"))
lock.getWebhooks = AsyncMock(side_effect=[[], webhooks_fixture])
with patch("loqedAPI.loqed.LoqedAPI.async_get_lock", return_value=lock), patch(
"loqedAPI.loqed.LoqedAPI.async_get_lock_details", return_value=lock_status
), patch(
"homeassistant.components.cloud.async_active_subscription", return_value=True
), patch(
"homeassistant.components.cloud.async_create_cloudhook",
return_value=webhooks_fixture[0]["url"],
):
await async_setup_component(hass, DOMAIN, config)
await hass.async_block_till_done()
lock.registerWebhook.assert_called_with(f"{get_url(hass)}/api/webhook/Webhook_id")
async def test_unload_entry(hass, integration: MockConfigEntry, lock: loqed.Lock):