Decouple Withings webhook tests from YAML (#100143)
parent
ad5e9e9f5b
commit
5c206de906
|
@ -53,6 +53,8 @@ NOT_AUTHENTICATED_ERROR = re.compile(
|
|||
re.IGNORECASE,
|
||||
)
|
||||
DATA_UPDATED_SIGNAL = "withings_entity_state_updated"
|
||||
SUBSCRIBE_DELAY = datetime.timedelta(seconds=5)
|
||||
UNSUBSCRIBE_DELAY = datetime.timedelta(seconds=1)
|
||||
|
||||
|
||||
class UpdateType(StrEnum):
|
||||
|
@ -229,8 +231,8 @@ class DataManager:
|
|||
self._user_id = user_id
|
||||
self._profile = profile
|
||||
self._webhook_config = webhook_config
|
||||
self._notify_subscribe_delay = datetime.timedelta(seconds=5)
|
||||
self._notify_unsubscribe_delay = datetime.timedelta(seconds=1)
|
||||
self._notify_subscribe_delay = SUBSCRIBE_DELAY
|
||||
self._notify_unsubscribe_delay = UNSUBSCRIBE_DELAY
|
||||
|
||||
self._is_available = True
|
||||
self._cancel_interval_update_interval: CALLBACK_TYPE | None = None
|
||||
|
|
|
@ -1,27 +1,21 @@
|
|||
"""Tests for the withings component."""
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import arrow
|
||||
from withings_api import DateType
|
||||
from withings_api.common import (
|
||||
GetSleepSummaryField,
|
||||
MeasureGetMeasGroupCategory,
|
||||
MeasureGetMeasResponse,
|
||||
MeasureType,
|
||||
NotifyAppli,
|
||||
NotifyListResponse,
|
||||
SleepGetSummaryResponse,
|
||||
UserGetDeviceResponse,
|
||||
)
|
||||
|
||||
from homeassistant.components.webhook import async_generate_url
|
||||
from homeassistant.config import async_process_ha_core_config
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .common import WebhookResponse
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
from tests.common import load_json_object_fixture
|
||||
|
||||
@dataclass
|
||||
class WebhookResponse:
|
||||
"""Response data from a webhook."""
|
||||
|
||||
message: str
|
||||
message_code: int
|
||||
|
||||
|
||||
async def call_webhook(
|
||||
|
@ -44,56 +38,13 @@ async def call_webhook(
|
|||
return WebhookResponse(message=data["message"], message_code=data["code"])
|
||||
|
||||
|
||||
class MockWithings:
|
||||
"""Mock object for Withings."""
|
||||
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
|
||||
"""Fixture for setting up the component."""
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
device_fixture: str = "person0_get_device.json",
|
||||
measurement_fixture: str = "person0_get_meas.json",
|
||||
sleep_fixture: str = "person0_get_sleep.json",
|
||||
notify_list_fixture: str = "person0_notify_list.json",
|
||||
):
|
||||
"""Initialize mock."""
|
||||
self.device_fixture = device_fixture
|
||||
self.measurement_fixture = measurement_fixture
|
||||
self.sleep_fixture = sleep_fixture
|
||||
self.notify_list_fixture = notify_list_fixture
|
||||
await async_process_ha_core_config(
|
||||
hass,
|
||||
{"internal_url": "http://example.local:8123"},
|
||||
)
|
||||
|
||||
def user_get_device(self) -> UserGetDeviceResponse:
|
||||
"""Get devices."""
|
||||
fixture = load_json_object_fixture(f"withings/{self.device_fixture}")
|
||||
return UserGetDeviceResponse(**fixture)
|
||||
|
||||
def measure_get_meas(
|
||||
self,
|
||||
meastype: MeasureType | None = None,
|
||||
category: MeasureGetMeasGroupCategory | None = None,
|
||||
startdate: DateType | None = None,
|
||||
enddate: DateType | None = None,
|
||||
offset: int | None = None,
|
||||
lastupdate: DateType | None = None,
|
||||
) -> MeasureGetMeasResponse:
|
||||
"""Get measurements."""
|
||||
fixture = load_json_object_fixture(f"withings/{self.measurement_fixture}")
|
||||
return MeasureGetMeasResponse(**fixture)
|
||||
|
||||
def sleep_get_summary(
|
||||
self,
|
||||
data_fields: Iterable[GetSleepSummaryField],
|
||||
startdateymd: DateType | None = arrow.utcnow(),
|
||||
enddateymd: DateType | None = arrow.utcnow(),
|
||||
offset: int | None = None,
|
||||
lastupdate: DateType | None = arrow.utcnow(),
|
||||
) -> SleepGetSummaryResponse:
|
||||
"""Get sleep."""
|
||||
fixture = load_json_object_fixture(f"withings/{self.sleep_fixture}")
|
||||
return SleepGetSummaryResponse(**fixture)
|
||||
|
||||
def notify_list(
|
||||
self,
|
||||
appli: NotifyAppli | None = None,
|
||||
) -> NotifyListResponse:
|
||||
"""Get sleep."""
|
||||
fixture = load_json_object_fixture(f"withings/{self.notify_list_fixture}")
|
||||
return NotifyListResponse(**fixture)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
|
|
|
@ -44,6 +44,7 @@ from homeassistant.setup import async_setup_component
|
|||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.components.withings import WebhookResponse
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
|
||||
|
||||
|
@ -91,14 +92,6 @@ def new_profile_config(
|
|||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WebhookResponse:
|
||||
"""Response data from a webhook."""
|
||||
|
||||
message: str
|
||||
message_code: int
|
||||
|
||||
|
||||
class ComponentFactory:
|
||||
"""Manages the setup and unloading of the withing component and profiles."""
|
||||
|
||||
|
|
|
@ -1,28 +1,30 @@
|
|||
"""Fixtures for tests."""
|
||||
from collections.abc import Awaitable, Callable, Coroutine
|
||||
from datetime import timedelta
|
||||
import time
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from withings_api import (
|
||||
MeasureGetMeasResponse,
|
||||
NotifyListResponse,
|
||||
SleepGetSummaryResponse,
|
||||
UserGetDeviceResponse,
|
||||
)
|
||||
|
||||
from homeassistant.components.application_credentials import (
|
||||
ClientCredential,
|
||||
async_import_client_credential,
|
||||
)
|
||||
from homeassistant.components.withings.common import ConfigEntryWithingsApi
|
||||
from homeassistant.components.withings.const import DOMAIN
|
||||
from homeassistant.config import async_process_ha_core_config
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import MockWithings
|
||||
from .common import ComponentFactory
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.common import MockConfigEntry, load_json_object_fixture
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
|
||||
ComponentSetup = Callable[[], Awaitable[MockWithings]]
|
||||
|
||||
CLIENT_ID = "1234"
|
||||
CLIENT_SECRET = "5678"
|
||||
SCOPES = [
|
||||
|
@ -100,33 +102,40 @@ def mock_config_entry(expires_at: int, scopes: list[str]) -> MockConfigEntry:
|
|||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="setup_integration")
|
||||
async def mock_setup_integration(
|
||||
hass: HomeAssistant, config_entry: MockConfigEntry
|
||||
) -> Callable[[], Coroutine[Any, Any, MockWithings]]:
|
||||
"""Fixture for setting up the component."""
|
||||
config_entry.add_to_hass(hass)
|
||||
@pytest.fixture(name="withings")
|
||||
def mock_withings():
|
||||
"""Mock withings."""
|
||||
|
||||
assert await async_setup_component(hass, "application_credentials", {})
|
||||
await async_import_client_credential(
|
||||
hass,
|
||||
DOMAIN,
|
||||
ClientCredential(CLIENT_ID, CLIENT_SECRET),
|
||||
DOMAIN,
|
||||
mock = AsyncMock(spec=ConfigEntryWithingsApi)
|
||||
mock.user_get_device.return_value = UserGetDeviceResponse(
|
||||
**load_json_object_fixture("withings/get_device.json")
|
||||
)
|
||||
await async_process_ha_core_config(
|
||||
hass,
|
||||
{"internal_url": "http://example.local:8123"},
|
||||
mock.measure_get_meas.return_value = MeasureGetMeasResponse(
|
||||
**load_json_object_fixture("withings/get_meas.json")
|
||||
)
|
||||
mock.sleep_get_summary.return_value = SleepGetSummaryResponse(
|
||||
**load_json_object_fixture("withings/get_sleep.json")
|
||||
)
|
||||
mock.notify_list.return_value = NotifyListResponse(
|
||||
**load_json_object_fixture("withings/notify_list.json")
|
||||
)
|
||||
|
||||
async def func() -> MockWithings:
|
||||
mock = MockWithings()
|
||||
with patch(
|
||||
"homeassistant.components.withings.common.ConfigEntryWithingsApi",
|
||||
return_value=mock,
|
||||
):
|
||||
assert await async_setup_component(hass, DOMAIN, {})
|
||||
await hass.async_block_till_done()
|
||||
return mock
|
||||
with patch(
|
||||
"homeassistant.components.withings.common.ConfigEntryWithingsApi",
|
||||
return_value=mock,
|
||||
):
|
||||
yield mock
|
||||
|
||||
return func
|
||||
|
||||
@pytest.fixture(name="disable_webhook_delay")
|
||||
def disable_webhook_delay():
|
||||
"""Disable webhook delay."""
|
||||
|
||||
mock = AsyncMock()
|
||||
with patch(
|
||||
"homeassistant.components.withings.common.SUBSCRIBE_DELAY", timedelta(seconds=0)
|
||||
), patch(
|
||||
"homeassistant.components.withings.common.UNSUBSCRIBE_DELAY",
|
||||
timedelta(seconds=0),
|
||||
):
|
||||
yield mock
|
||||
|
|
|
@ -0,0 +1,15 @@
|
|||
{
|
||||
"devices": [
|
||||
{
|
||||
"type": "Scale",
|
||||
"battery": "high",
|
||||
"model": "Body+",
|
||||
"model_id": 5,
|
||||
"timezone": "Europe/Amsterdam",
|
||||
"first_session_date": null,
|
||||
"last_session_date": 1693867179,
|
||||
"deviceid": "f998be4b9ccc9e136fd8cd8e8e344c31ec3b271d",
|
||||
"hash_deviceid": "f998be4b9ccc9e136fd8cd8e8e344c31ec3b271d"
|
||||
}
|
||||
]
|
||||
}
|
|
@ -0,0 +1,22 @@
|
|||
{
|
||||
"profiles": [
|
||||
{
|
||||
"appli": 50,
|
||||
"callbackurl": "https://not.my.callback/url",
|
||||
"expires": 2147483647,
|
||||
"comment": null
|
||||
},
|
||||
{
|
||||
"appli": 50,
|
||||
"callbackurl": "http://example.local:8123/api/webhook/55a7335ea8dee830eed4ef8f84cda8f6d80b83af0847dc74032e86120bffed5e",
|
||||
"expires": 2147483647,
|
||||
"comment": null
|
||||
},
|
||||
{
|
||||
"appli": 51,
|
||||
"callbackurl": "http://example.local:8123/api/webhook/55a7335ea8dee830eed4ef8f84cda8f6d80b83af0847dc74032e86120bffed5e",
|
||||
"expires": 2147483647,
|
||||
"comment": null
|
||||
}
|
||||
]
|
||||
}
|
|
@ -1,18 +0,0 @@
|
|||
{
|
||||
"status": 0,
|
||||
"body": {
|
||||
"devices": [
|
||||
{
|
||||
"type": "Scale",
|
||||
"battery": "high",
|
||||
"model": "Body+",
|
||||
"model_id": 5,
|
||||
"timezone": "Europe/Amsterdam",
|
||||
"first_session_date": null,
|
||||
"last_session_date": 1693867179,
|
||||
"deviceid": "f998be4b9ccc9e136fd8cd8e8e344c31ec3b271d",
|
||||
"hash_deviceid": "f998be4b9ccc9e136fd8cd8e8e344c31ec3b271d"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
|
@ -1,3 +0,0 @@
|
|||
{
|
||||
"profiles": []
|
||||
}
|
|
@ -1,51 +1,50 @@
|
|||
"""Tests for the Withings component."""
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from withings_api.common import NotifyAppli
|
||||
|
||||
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import MockWithings, call_webhook
|
||||
from .conftest import USER_ID, WEBHOOK_ID, ComponentSetup
|
||||
from . import call_webhook, setup_integration
|
||||
from .conftest import USER_ID, WEBHOOK_ID
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
async def test_binary_sensor(
|
||||
hass: HomeAssistant,
|
||||
setup_integration: ComponentSetup,
|
||||
withings: AsyncMock,
|
||||
disable_webhook_delay,
|
||||
config_entry: MockConfigEntry,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
) -> None:
|
||||
"""Test binary sensor."""
|
||||
await setup_integration()
|
||||
mock = MockWithings()
|
||||
with patch(
|
||||
"homeassistant.components.withings.common.ConfigEntryWithingsApi",
|
||||
return_value=mock,
|
||||
):
|
||||
client = await hass_client_no_auth()
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
entity_id = "binary_sensor.henk_in_bed"
|
||||
client = await hass_client_no_auth()
|
||||
|
||||
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
|
||||
entity_id = "binary_sensor.henk_in_bed"
|
||||
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.BED_IN},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ON
|
||||
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
|
||||
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.BED_OUT},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_OFF
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.BED_IN},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_ON
|
||||
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.BED_OUT},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.get(entity_id).state == STATE_OFF
|
||||
|
|
|
@ -1,5 +1,4 @@
|
|||
"""Tests for the Withings component."""
|
||||
import datetime
|
||||
from http import HTTPStatus
|
||||
import re
|
||||
from typing import Any
|
||||
|
@ -9,20 +8,15 @@ from urllib.parse import urlparse
|
|||
from aiohttp.test_utils import TestClient
|
||||
import pytest
|
||||
import requests_mock
|
||||
from withings_api.common import NotifyAppli, NotifyListProfile, NotifyListResponse
|
||||
from withings_api.common import NotifyAppli
|
||||
|
||||
from homeassistant.components.withings.common import (
|
||||
ConfigEntryWithingsApi,
|
||||
DataManager,
|
||||
WebhookConfig,
|
||||
)
|
||||
from homeassistant.components.withings.common import ConfigEntryWithingsApi
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.config_entry_oauth2_flow import AbstractOAuth2Implementation
|
||||
|
||||
from .common import ComponentFactory, get_data_manager_by_user_id, new_profile_config
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
|
@ -101,137 +95,3 @@ async def test_webhook_post(
|
|||
resp.close()
|
||||
|
||||
assert data["code"] == expected_code
|
||||
|
||||
|
||||
async def test_webhook_head(
|
||||
hass: HomeAssistant,
|
||||
component_factory: ComponentFactory,
|
||||
aiohttp_client: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
) -> None:
|
||||
"""Test head method on webhook view."""
|
||||
person0 = new_profile_config("person0", 0)
|
||||
|
||||
await component_factory.configure_component(profile_configs=(person0,))
|
||||
await component_factory.setup_profile(person0.user_id)
|
||||
data_manager = get_data_manager_by_user_id(hass, person0.user_id)
|
||||
|
||||
client: TestClient = await aiohttp_client(hass.http.app)
|
||||
resp = await client.head(urlparse(data_manager.webhook_config.url).path)
|
||||
assert resp.status == HTTPStatus.OK
|
||||
|
||||
|
||||
async def test_webhook_put(
|
||||
hass: HomeAssistant,
|
||||
component_factory: ComponentFactory,
|
||||
aiohttp_client: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
) -> None:
|
||||
"""Test webhook callback."""
|
||||
person0 = new_profile_config("person0", 0)
|
||||
|
||||
await component_factory.configure_component(profile_configs=(person0,))
|
||||
await component_factory.setup_profile(person0.user_id)
|
||||
data_manager = get_data_manager_by_user_id(hass, person0.user_id)
|
||||
|
||||
client: TestClient = await aiohttp_client(hass.http.app)
|
||||
resp = await client.put(urlparse(data_manager.webhook_config.url).path)
|
||||
|
||||
# Wait for remaining tasks to complete.
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert resp.status == HTTPStatus.OK
|
||||
data = await resp.json()
|
||||
assert data
|
||||
assert data["code"] == 2
|
||||
|
||||
|
||||
async def test_data_manager_webhook_subscription(
|
||||
hass: HomeAssistant,
|
||||
component_factory: ComponentFactory,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
) -> None:
|
||||
"""Test data manager webhook subscriptions."""
|
||||
person0 = new_profile_config("person0", 0)
|
||||
await component_factory.configure_component(profile_configs=(person0,))
|
||||
|
||||
api: ConfigEntryWithingsApi = MagicMock(spec=ConfigEntryWithingsApi)
|
||||
data_manager = DataManager(
|
||||
hass,
|
||||
"person0",
|
||||
api,
|
||||
0,
|
||||
WebhookConfig(id="1234", url="http://localhost/api/webhook/1234", enabled=True),
|
||||
)
|
||||
|
||||
data_manager._notify_subscribe_delay = datetime.timedelta(seconds=0)
|
||||
data_manager._notify_unsubscribe_delay = datetime.timedelta(seconds=0)
|
||||
|
||||
api.notify_list.return_value = NotifyListResponse(
|
||||
profiles=(
|
||||
NotifyListProfile(
|
||||
appli=NotifyAppli.BED_IN,
|
||||
callbackurl="https://not.my.callback/url",
|
||||
expires=None,
|
||||
comment=None,
|
||||
),
|
||||
NotifyListProfile(
|
||||
appli=NotifyAppli.BED_IN,
|
||||
callbackurl=data_manager.webhook_config.url,
|
||||
expires=None,
|
||||
comment=None,
|
||||
),
|
||||
NotifyListProfile(
|
||||
appli=NotifyAppli.BED_OUT,
|
||||
callbackurl=data_manager.webhook_config.url,
|
||||
expires=None,
|
||||
comment=None,
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
aioclient_mock.clear_requests()
|
||||
aioclient_mock.request(
|
||||
"HEAD",
|
||||
data_manager.webhook_config.url,
|
||||
status=HTTPStatus.OK,
|
||||
)
|
||||
|
||||
# Test subscribing
|
||||
await data_manager.async_subscribe_webhook()
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.WEIGHT
|
||||
)
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.CIRCULATORY
|
||||
)
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.ACTIVITY
|
||||
)
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.SLEEP
|
||||
)
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.USER
|
||||
)
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.BED_IN
|
||||
)
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
api.notify_subscribe.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.BED_OUT
|
||||
)
|
||||
|
||||
# Test unsubscribing.
|
||||
await data_manager.async_unsubscribe_webhook()
|
||||
api.notify_revoke.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.BED_IN
|
||||
)
|
||||
api.notify_revoke.assert_any_call(
|
||||
data_manager.webhook_config.url, NotifyAppli.BED_OUT
|
||||
)
|
||||
|
|
|
@ -86,6 +86,7 @@ async def test_config_non_unique_profile(
|
|||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
current_request_with_host: None,
|
||||
disable_webhook_delay,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
) -> None:
|
||||
"""Test setup a non-unique profile."""
|
||||
|
@ -154,6 +155,7 @@ async def test_config_reauth_profile(
|
|||
hass_client_no_auth: ClientSessionGenerator,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
config_entry: MockConfigEntry,
|
||||
disable_webhook_delay,
|
||||
current_request_with_host,
|
||||
) -> None:
|
||||
"""Test reauth an existing profile re-creates the config entry."""
|
||||
|
|
|
@ -1,11 +1,14 @@
|
|||
"""Tests for the Withings component."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
from datetime import timedelta
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
from withings_api.common import UnauthorizedException
|
||||
from withings_api.common import NotifyAppli, UnauthorizedException
|
||||
|
||||
import homeassistant.components.webhook as webhook
|
||||
from homeassistant.components.webhook import async_generate_url
|
||||
from homeassistant.components.withings import CONFIG_SCHEMA, DOMAIN, async_setup, const
|
||||
from homeassistant.components.withings.common import ConfigEntryWithingsApi, DataManager
|
||||
from homeassistant.config import async_process_ha_core_config
|
||||
|
@ -19,10 +22,14 @@ from homeassistant.const import (
|
|||
from homeassistant.core import DOMAIN as HA_DOMAIN, HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from . import setup_integration
|
||||
from .common import ComponentFactory, get_data_manager_by_user_id, new_profile_config
|
||||
from .conftest import WEBHOOK_ID
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
def config_schema_validate(withings_config) -> dict:
|
||||
|
@ -224,3 +231,57 @@ async def test_set_convert_unique_id_to_string(hass: HomeAssistant) -> None:
|
|||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.unique_id == "1234"
|
||||
|
||||
|
||||
async def test_data_manager_webhook_subscription(
|
||||
hass: HomeAssistant,
|
||||
withings: AsyncMock,
|
||||
disable_webhook_delay,
|
||||
config_entry: MockConfigEntry,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
) -> None:
|
||||
"""Test data manager webhook subscriptions."""
|
||||
await setup_integration(hass, config_entry)
|
||||
await hass_client_no_auth()
|
||||
await hass.async_block_till_done()
|
||||
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1))
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert withings.notify_subscribe.call_count == 4
|
||||
|
||||
webhook_url = "http://example.local:8123/api/webhook/55a7335ea8dee830eed4ef8f84cda8f6d80b83af0847dc74032e86120bffed5e"
|
||||
|
||||
withings.notify_subscribe.assert_any_call(webhook_url, NotifyAppli.WEIGHT)
|
||||
withings.notify_subscribe.assert_any_call(webhook_url, NotifyAppli.CIRCULATORY)
|
||||
withings.notify_subscribe.assert_any_call(webhook_url, NotifyAppli.ACTIVITY)
|
||||
withings.notify_subscribe.assert_any_call(webhook_url, NotifyAppli.SLEEP)
|
||||
|
||||
withings.notify_revoke.assert_any_call(webhook_url, NotifyAppli.BED_IN)
|
||||
withings.notify_revoke.assert_any_call(webhook_url, NotifyAppli.BED_OUT)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"method",
|
||||
[
|
||||
"PUT",
|
||||
"HEAD",
|
||||
],
|
||||
)
|
||||
async def test_requests(
|
||||
hass: HomeAssistant,
|
||||
withings: AsyncMock,
|
||||
config_entry: MockConfigEntry,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
method: str,
|
||||
disable_webhook_delay,
|
||||
) -> None:
|
||||
"""Test we handle request methods Withings sends."""
|
||||
await setup_integration(hass, config_entry)
|
||||
client = await hass_client_no_auth()
|
||||
webhook_url = async_generate_url(hass, WEBHOOK_ID)
|
||||
|
||||
response = await client.request(
|
||||
method=method,
|
||||
path=urlparse(webhook_url).path,
|
||||
)
|
||||
assert response.status == 200
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
"""Tests for the Withings component."""
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from syrupy import SnapshotAssertion
|
||||
|
@ -14,10 +14,11 @@ from homeassistant.core import HomeAssistant, State
|
|||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.helpers.entity_registry import EntityRegistry
|
||||
|
||||
from . import MockWithings, call_webhook
|
||||
from . import call_webhook, setup_integration
|
||||
from .common import async_get_entity_id
|
||||
from .conftest import USER_ID, WEBHOOK_ID, ComponentSetup
|
||||
from .conftest import USER_ID, WEBHOOK_ID
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
WITHINGS_MEASUREMENTS_MAP: dict[Measurement, WithingsEntityDescription] = {
|
||||
|
@ -77,65 +78,55 @@ def async_assert_state_equals(
|
|||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_sensor_default_enabled_entities(
|
||||
hass: HomeAssistant,
|
||||
setup_integration: ComponentSetup,
|
||||
withings: AsyncMock,
|
||||
config_entry: MockConfigEntry,
|
||||
disable_webhook_delay,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
) -> None:
|
||||
"""Test entities enabled by default."""
|
||||
await setup_integration()
|
||||
await setup_integration(hass, config_entry)
|
||||
entity_registry: EntityRegistry = er.async_get(hass)
|
||||
|
||||
mock = MockWithings()
|
||||
with patch(
|
||||
"homeassistant.components.withings.common.ConfigEntryWithingsApi",
|
||||
return_value=mock,
|
||||
):
|
||||
client = await hass_client_no_auth()
|
||||
# Assert entities should exist.
|
||||
for attribute in SENSORS:
|
||||
entity_id = await async_get_entity_id(
|
||||
hass, attribute, USER_ID, SENSOR_DOMAIN
|
||||
)
|
||||
assert entity_id
|
||||
assert entity_registry.async_is_registered(entity_id)
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.SLEEP},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.WEIGHT},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
client = await hass_client_no_auth()
|
||||
# Assert entities should exist.
|
||||
for attribute in SENSORS:
|
||||
entity_id = await async_get_entity_id(hass, attribute, USER_ID, SENSOR_DOMAIN)
|
||||
assert entity_id
|
||||
assert entity_registry.async_is_registered(entity_id)
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.SLEEP},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
resp = await call_webhook(
|
||||
hass,
|
||||
WEBHOOK_ID,
|
||||
{"userid": USER_ID, "appli": NotifyAppli.WEIGHT},
|
||||
client,
|
||||
)
|
||||
assert resp.message_code == 0
|
||||
|
||||
assert resp.message_code == 0
|
||||
for measurement, expected in EXPECTED_DATA:
|
||||
attribute = WITHINGS_MEASUREMENTS_MAP[measurement]
|
||||
entity_id = await async_get_entity_id(hass, attribute, USER_ID, SENSOR_DOMAIN)
|
||||
state_obj = hass.states.get(entity_id)
|
||||
|
||||
for measurement, expected in EXPECTED_DATA:
|
||||
attribute = WITHINGS_MEASUREMENTS_MAP[measurement]
|
||||
entity_id = await async_get_entity_id(
|
||||
hass, attribute, USER_ID, SENSOR_DOMAIN
|
||||
)
|
||||
state_obj = hass.states.get(entity_id)
|
||||
|
||||
async_assert_state_equals(entity_id, state_obj, expected, attribute)
|
||||
async_assert_state_equals(entity_id, state_obj, expected, attribute)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_all_entities(
|
||||
hass: HomeAssistant, setup_integration: ComponentSetup, snapshot: SnapshotAssertion
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
withings: AsyncMock,
|
||||
disable_webhook_delay,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test all entities."""
|
||||
await setup_integration()
|
||||
await setup_integration(hass, config_entry)
|
||||
|
||||
mock = MockWithings()
|
||||
with patch(
|
||||
"homeassistant.components.withings.common.ConfigEntryWithingsApi",
|
||||
return_value=mock,
|
||||
):
|
||||
for sensor in SENSORS:
|
||||
entity_id = await async_get_entity_id(hass, sensor, USER_ID, SENSOR_DOMAIN)
|
||||
assert hass.states.get(entity_id) == snapshot
|
||||
for sensor in SENSORS:
|
||||
entity_id = await async_get_entity_id(hass, sensor, USER_ID, SENSOR_DOMAIN)
|
||||
assert hass.states.get(entity_id) == snapshot
|
||||
|
|
Loading…
Reference in New Issue