starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-04 02:34:40 -05:00
Code Issues Packages Projects Releases Wiki Activity

Switchbot Cloud: Enable Webhook for Bot (#165647)

Browse Source
This commit is contained in:
Samuel Xiao
2026-04-10 20:13:56 +00:00
committed by Franck Nijhof
parent ab601e5717
commit b880876e0e
4 changed files with 122 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/switchbot_cloud/__init__.py
+38 -30
View File
@@ -13,6 +13,7 @@ from switchbot_api import (
SwitchBotAPI,
SwitchBotAuthenticationError,
SwitchBotConnectionError,
SwitchBotDeviceOfflineError,
)
from homeassistant.components import webhook
@@ -202,7 +203,7 @@ async def make_device_data(
if isinstance(device, Device) and device.device_type == "Bot":
coordinator = await coordinator_for_device(
hass, entry, api, device, coordinators_by_id
hass, entry, api, device, coordinators_by_id, True
)
devices_data.sensors.append((device, coordinator))
if coordinator.data is not None:
@@ -405,42 +406,49 @@ async def _initialize_webhook(
hass,
entry.data[CONF_WEBHOOK_ID],
)
# check if webhook is configured in switchbot cloud
check_webhook_result = None
with contextlib.suppress(Exception):
check_webhook_result = await api.get_webook_configuration()
actual_webhook_urls = (
check_webhook_result["urls"]
if check_webhook_result and "urls" in check_webhook_result
else []
)
need_add_webhook = (
len(actual_webhook_urls) == 0 or webhook_url not in actual_webhook_urls
)
need_clean_previous_webhook = (
len(actual_webhook_urls) > 0 and webhook_url not in actual_webhook_urls
)
try:
check_webhook_result = None
with contextlib.suppress(Exception):
check_webhook_result = await api.get_webook_configuration()
if need_clean_previous_webhook:
# it seems is impossible to register multiple webhook.
# So, if webhook already exists, we delete it
await api.delete_webhook(actual_webhook_urls[0])
_LOGGER.debug(
"Deleted previous Switchbot cloud webhook url: %s",
actual_webhook_urls[0],
actual_webhook_urls = (
check_webhook_result["urls"]
if check_webhook_result and "urls" in check_webhook_result
else []
)
need_add_webhook = (
len(actual_webhook_urls) == 0 or webhook_url not in actual_webhook_urls
)
need_clean_previous_webhook = (
len(actual_webhook_urls) > 0 and webhook_url not in actual_webhook_urls
)
if need_add_webhook:
# call api for register webhookurl
await api.setup_webhook(webhook_url)
_LOGGER.debug("Registered Switchbot cloud webhook at hass: %s", webhook_url)
if need_clean_previous_webhook:
# it seems is impossible to register multiple webhook.
# So, if webhook already exists, we delete it
await api.delete_webhook(actual_webhook_urls[0])
_LOGGER.debug(
"Deleted previous Switchbot cloud webhook url: %s",
actual_webhook_urls[0],
)
for coordinator in coordinators_by_id.values():
coordinator.webhook_subscription_listener(True)
if need_add_webhook:
# call api for register webhookurl
await api.setup_webhook(webhook_url)
_LOGGER.debug(
"Registered Switchbot cloud webhook at hass: %s", webhook_url
)
_LOGGER.debug("Registered Switchbot cloud webhook at: %s", webhook_url)
for coordinator in coordinators_by_id.values():
coordinator.webhook_subscription_listener(True)
_LOGGER.debug("Registered Switchbot cloud webhook at: %s", webhook_url)
except SwitchBotDeviceOfflineError as e:
_LOGGER.error("Failed to connect Switchbot cloud device: %s", e)
except SwitchBotConnectionError as e:
_LOGGER.error("Failed to connect Switchbot cloud device: %s", e)
def _create_handle_webhook(
tests/components/switchbot_cloud/conftest.py
+23
View File
@@ -32,6 +32,29 @@ def mock_get_status():
yield mock_get_status
@pytest.fixture
def mock_setup_webhook():
"""Mock setup_webhook."""
with patch.object(SwitchBotAPI, "setup_webhook") as mock_setup_webhook:
yield mock_setup_webhook
@pytest.fixture
def mock_delete_webhook():
"""Mock delete_webhook."""
with patch.object(SwitchBotAPI, "delete_webhook") as mock_delete_webhook:
yield mock_delete_webhook
@pytest.fixture
def mock_get_webook_configuration():
"""Mock get_webook_configuration."""
with patch.object(
SwitchBotAPI, "get_webook_configuration"
) as mock_get_webook_configuration:
yield mock_get_webook_configuration
@pytest.fixture(scope="package", autouse=True)
def mock_after_command_refresh():
"""Mock after command refresh."""
tests/components/switchbot_cloud/test_button.py
+31 -2
View File
@@ -16,7 +16,11 @@ from . import configure_integration
async def test_pressmode_bot(
hass: HomeAssistant, mock_list_devices, mock_get_status
hass: HomeAssistant,
mock_list_devices,
mock_get_status,
mock_setup_webhook,
mock_get_webook_configuration,
) -> None:
"""Test press."""
mock_list_devices.return_value = [
@@ -31,6 +35,17 @@ async def test_pressmode_bot(
mock_get_status.return_value = {"deviceMode": "pressMode"}
mock_setup_webhook.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
mock_get_webook_configuration.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
@@ -49,7 +64,11 @@ async def test_pressmode_bot(
async def test_switchmode_bot_no_button_entity(
hass: HomeAssistant, mock_list_devices, mock_get_status
hass: HomeAssistant,
mock_list_devices,
mock_get_status,
mock_setup_webhook,
mock_get_webook_configuration,
) -> None:
"""Test a switchMode bot isn't added as a button."""
mock_list_devices.return_value = [
@@ -63,6 +82,16 @@ async def test_switchmode_bot_no_button_entity(
]
mock_get_status.return_value = {"deviceMode": "switchMode"}
mock_setup_webhook.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
mock_get_webook_configuration.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
tests/components/switchbot_cloud/test_switch.py
+30 -2
View File
@@ -56,7 +56,11 @@ async def test_relay_switch(
async def test_switchmode_bot(
hass: HomeAssistant, mock_list_devices, mock_get_status
hass: HomeAssistant,
mock_list_devices,
mock_get_status,
mock_setup_webhook,
mock_get_webook_configuration,
) -> None:
"""Test turn on and turn off."""
mock_list_devices.return_value = [
@@ -71,6 +75,16 @@ async def test_switchmode_bot(
mock_get_status.return_value = {"deviceMode": "switchMode", "power": "off"}
mock_setup_webhook.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
mock_get_webook_configuration.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
@@ -91,7 +105,11 @@ async def test_switchmode_bot(
async def test_pressmode_bot_no_switch_entity(
hass: HomeAssistant, mock_list_devices, mock_get_status
hass: HomeAssistant,
mock_list_devices,
mock_get_status,
mock_setup_webhook,
mock_get_webook_configuration,
) -> None:
"""Test a pressMode bot isn't added as a switch."""
mock_list_devices.return_value = [
@@ -105,6 +123,16 @@ async def test_pressmode_bot_no_switch_entity(
]
mock_get_status.return_value = {"deviceMode": "pressMode"}
mock_setup_webhook.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
mock_get_webook_configuration.return_value = {
"statusCode": 100,
"body": {},
"message": "success",
}
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED