Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve flume test coverage (#120851) 

* Add Flume init tests

* Increase test coverage

* Improve readability

* Fix pydoc for tests

* Use pytest.mark.usefixtures
pull/120941/head
Allen Porter 2024-07-01 07:41:47 -07:00 committed by GitHub
parent c9911fa8ce
commit 2506acc095
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 392 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -441,7 +441,6 @@ omit =
homeassistant/components/flick_electric/__init__.py
homeassistant/components/flick_electric/sensor.py
homeassistant/components/flock/notify.py
homeassistant/components/flume/__init__.py
homeassistant/components/flume/binary_sensor.py
homeassistant/components/flume/coordinator.py
homeassistant/components/flume/entity.py

167
tests/components/flume/conftest.py Normal file
View File

@ -0,0 +1,167 @@
"""Flume test fixtures."""
from collections.abc import Generator
import datetime
from http import HTTPStatus
import json
from unittest.mock import mock_open, patch
import jwt
import pytest
import requests
from requests_mock.mocker import Mocker
from homeassistant.components.flume.const import DOMAIN
from homeassistant.const import (
CONF_CLIENT_ID,
CONF_CLIENT_SECRET,
CONF_PASSWORD,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
USER_ID = "test-user-id"
REFRESH_TOKEN = "refresh-token"
TOKEN_URL = "https://api.flumetech.com/oauth/token"
DEVICE_LIST_URL = (
"https://api.flumetech.com/users/test-user-id/devices?user=true&location=true"
)
BRIDGE_DEVICE = {
"id": "1234",
"type": 1, # Bridge
"location": {
"name": "Bridge Location",
},
"name": "Flume Bridge",
"connected": True,
}
SENSOR_DEVICE = {
"id": "1234",
"type": 2, # Sensor
"location": {
"name": "Sensor Location",
},
"name": "Flume Sensor",
"connected": True,
}
DEVICE_LIST = [BRIDGE_DEVICE, SENSOR_DEVICE]
NOTIFICATIONS_URL = "https://api.flumetech.com/users/test-user-id/notifications?limit=50&offset=0&sort_direction=ASC"
NOTIFICATION = {
"id": 111111,
"device_id": "6248148189204194987",
"user_id": USER_ID,
"type": 1,
"message": "Low Flow Leak triggered at Home. Water has been running for 2 hours averaging 0.43 gallons every minute.",
"created_datetime": "2020-01-15T16:33:39.000Z",
"title": "Potential Leak Detected!",
"read": True,
"extra": {
"query": {
"request_id": "SYSTEM_TRIGGERED_USAGE_ALERT",
"since_datetime": "2020-01-15 06:33:59",
"until_datetime": "2020-01-15 08:33:59",
"tz": "America/Los_Angeles",
"bucket": "MIN",
"raw": False,
"group_multiplier": 2,
"device_id": ["6248148189204194987"],
}
},
"event_rule": "Low Flow Leak",
}
NOTIFICATIONS_LIST = [NOTIFICATION]
@pytest.fixture(name="config_entry")
def config_entry_fixture(hass: HomeAssistant) -> MockConfigEntry:
"""Fixture to create a config entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
title="test-username",
unique_id="test-username",
data={
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_CLIENT_ID: "client_id",
CONF_CLIENT_SECRET: "client_secret",
},
)
config_entry.add_to_hass(hass)
return config_entry
def encode_access_token() -> str:
"""Encode the payload of the access token."""
expiration_time = datetime.datetime.now() + datetime.timedelta(hours=12)
payload = {
"user_id": USER_ID,
"exp": int(expiration_time.timestamp()),
}
return jwt.encode(payload, key="secret")
@pytest.fixture(name="access_token")
def access_token_fixture(requests_mock: Mocker) -> Generator[None, None, None]:
"""Fixture to setup the access token."""
token_response = {
"refresh_token": REFRESH_TOKEN,
"access_token": encode_access_token(),
}
requests_mock.register_uri(
"POST",
TOKEN_URL,
status_code=HTTPStatus.OK,
json={"data": [token_response]},
)
with patch("builtins.open", mock_open(read_data=json.dumps(token_response))):
yield
@pytest.fixture(name="device_list")
def device_list_fixture(requests_mock: Mocker) -> None:
"""Fixture to setup the device list API response access token."""
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
status_code=HTTPStatus.OK,
json={
"data": DEVICE_LIST,
},
)
@pytest.fixture(name="device_list_timeout")
def device_list_timeout_fixture(requests_mock: Mocker) -> None:
"""Fixture to test a timeout when connecting to the device list url."""
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
exc=requests.exceptions.ConnectTimeout,
)
@pytest.fixture(name="device_list_unauthorized")
def device_list_unauthorized_fixture(requests_mock: Mocker) -> None:
"""Fixture to test an authorized error from the device list url."""
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
status_code=HTTPStatus.UNAUTHORIZED,
json={},
)
@pytest.fixture(name="notifications_list")
def notifications_list_fixture(requests_mock: Mocker) -> None:
"""Fixture to setup the device list API response access token."""
requests_mock.register_uri(
"GET",
NOTIFICATIONS_URL,
status_code=HTTPStatus.OK,
json={
"data": NOTIFICATIONS_LIST,
},
)

177
tests/components/flume/test_config_flow.py
View File

@ -1,8 +1,11 @@
"""Test the flume config flow."""
from unittest.mock import MagicMock, patch
from http import HTTPStatus
from unittest.mock import patch
import pytest
import requests.exceptions
from requests_mock.mocker import Mocker
from homeassistant import config_entries
from homeassistant.components.flume.const import DOMAIN
@ -15,15 +18,12 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import DEVICE_LIST, DEVICE_LIST_URL
from tests.common import MockConfigEntry
def _get_mocked_flume_device_list():
flume_device_list_mock = MagicMock()
type(flume_device_list_mock).device_list = ["mock"]
return flume_device_list_mock
@pytest.mark.usefixtures("access_token", "device_list")
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form and can setup from user input."""
@ -33,17 +33,7 @@ async def test_form(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
mock_flume_device_list = _get_mocked_flume_device_list()
with (
patch(
"homeassistant.components.flume.config_flow.FlumeAuth",
return_value=True,
),
patch(
"homeassistant.components.flume.config_flow.FlumeDeviceList",
return_value=mock_flume_device_list,
),
patch(
"homeassistant.components.flume.async_setup_entry",
return_value=True,
@ -71,66 +61,57 @@ async def test_form(hass: HomeAssistant) -> None:
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("access_token")
async def test_form_invalid_auth(hass: HomeAssistant, requests_mock: Mocker) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with (
patch(
"homeassistant.components.flume.config_flow.FlumeAuth",
return_value=True,
),
patch(
"homeassistant.components.flume.config_flow.FlumeDeviceList",
side_effect=Exception,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_CLIENT_ID: "client_id",
CONF_CLIENT_SECRET: "client_secret",
},
)
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
status_code=HTTPStatus.UNAUTHORIZED,
json={"message": "Failure"},
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_CLIENT_ID: "client_id",
CONF_CLIENT_SECRET: "client_secret",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"password": "invalid_auth"}
@pytest.mark.usefixtures("access_token", "device_list_timeout")
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with (
patch(
"homeassistant.components.flume.config_flow.FlumeAuth",
return_value=True,
),
patch(
"homeassistant.components.flume.config_flow.FlumeDeviceList",
side_effect=requests.exceptions.ConnectionError(),
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_CLIENT_ID: "client_id",
CONF_CLIENT_SECRET: "client_secret",
},
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_CLIENT_ID: "client_id",
CONF_CLIENT_SECRET: "client_secret",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_reauth(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("access_token")
async def test_reauth(hass: HomeAssistant, requests_mock: Mocker) -> None:
"""Test we can reauth."""
entry = MockConfigEntry(
domain=DOMAIN,
@ -151,35 +132,28 @@ async def test_reauth(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with (
patch(
"homeassistant.components.flume.config_flow.FlumeAuth",
return_value=True,
),
patch(
"homeassistant.components.flume.config_flow.FlumeDeviceList",
side_effect=Exception,
),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "test-password",
},
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "test-password",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"password": "invalid_auth"}
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
exc=requests.exceptions.ConnectTimeout,
)
with (
patch(
"homeassistant.components.flume.config_flow.FlumeAuth",
"homeassistant.components.flume.config_flow.os.path.exists",
return_value=True,
),
patch(
"homeassistant.components.flume.config_flow.FlumeDeviceList",
side_effect=requests.exceptions.ConnectionError(),
),
patch("homeassistant.components.flume.config_flow.os.unlink") as mock_unlink,
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
@ -187,21 +161,22 @@ async def test_reauth(hass: HomeAssistant) -> None:
CONF_PASSWORD: "test-password",
},
)
# The existing token file was removed
assert len(mock_unlink.mock_calls) == 1
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {"base": "cannot_connect"}
mock_flume_device_list = _get_mocked_flume_device_list()
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
status_code=HTTPStatus.OK,
json={
"data": DEVICE_LIST,
},
)
with (
patch(
"homeassistant.components.flume.config_flow.FlumeAuth",
return_value=True,
),
patch(
"homeassistant.components.flume.config_flow.FlumeDeviceList",
return_value=mock_flume_device_list,
),
patch(
"homeassistant.components.flume.async_setup_entry",
return_value=True,
@ -217,3 +192,31 @@ async def test_reauth(hass: HomeAssistant) -> None:
assert mock_setup_entry.called
assert result4["type"] is FlowResultType.ABORT
assert result4["reason"] == "reauth_successful"
@pytest.mark.usefixtures("access_token")
async def test_form_no_devices(hass: HomeAssistant, requests_mock: Mocker) -> None:
"""Test a device list response that contains no values will raise an error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
requests_mock.register_uri(
"GET",
DEVICE_LIST_URL,
status_code=HTTPStatus.OK,
json={"data": []},
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_CLIENT_ID: "client_id",
CONF_CLIENT_SECRET: "client_secret",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}

135
tests/components/flume/test_init.py Normal file
View File

@ -0,0 +1,135 @@
"""Test the flume init."""
from collections.abc import Generator
from unittest.mock import patch
import pytest
from requests_mock.mocker import Mocker
from homeassistant import config_entries
from homeassistant.components.flume.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .conftest import USER_ID
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
def platforms_fixture() -> Generator[list[str]]:
"""Return the platforms to be loaded for this test."""
# Arbitrary platform to ensure notifications are loaded
with patch("homeassistant.components.flume.PLATFORMS", [Platform.BINARY_SENSOR]):
yield
@pytest.mark.usefixtures("access_token", "device_list")
async def test_setup_config_entry(
hass: HomeAssistant,
requests_mock: Mocker,
config_entry: MockConfigEntry,
) -> None:
"""Test load and unload of a ConfigEntry."""
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is config_entries.ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(config_entry.entry_id)
assert config_entry.state is config_entries.ConfigEntryState.NOT_LOADED
@pytest.mark.usefixtures("access_token", "device_list_timeout")
async def test_device_list_timeout(
hass: HomeAssistant,
requests_mock: Mocker,
config_entry: MockConfigEntry,
) -> None:
"""Test error handling for a timeout when listing devices."""
assert not await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is config_entries.ConfigEntryState.SETUP_RETRY
@pytest.mark.usefixtures("access_token", "device_list_unauthorized")
async def test_reauth_when_unauthorized(
hass: HomeAssistant,
requests_mock: Mocker,
config_entry: MockConfigEntry,
) -> None:
"""Test error handling for an authentication error when listing devices."""
assert not await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is config_entries.ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
assert flows[0]["step_id"] == "reauth_confirm"
@pytest.mark.usefixtures("access_token", "device_list", "notifications_list")
async def test_list_notifications_service(
hass: HomeAssistant,
requests_mock: Mocker,
config_entry: MockConfigEntry,
) -> None:
"""Test the list notifications service."""
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is config_entries.ConfigEntryState.LOADED
response = await hass.services.async_call(
DOMAIN,
"list_notifications",
{},
target={
"config_entry": config_entry.entry_id,
},
blocking=True,
return_response=True,
)
notifications = response.get("notifications")
assert notifications
assert len(notifications) == 1
assert notifications[0].get("user_id") == USER_ID
@pytest.mark.usefixtures("access_token", "device_list", "notifications_list")
async def test_list_notifications_service_config_entry_errors(
hass: HomeAssistant,
requests_mock: Mocker,
config_entry: MockConfigEntry,
) -> None:
"""Test error handling for notification service with invalid config entries."""
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is config_entries.ConfigEntryState.LOADED
assert await hass.config_entries.async_unload(config_entry.entry_id)
assert config_entry.state is config_entries.ConfigEntryState.NOT_LOADED
with pytest.raises(ValueError, match="Config entry not loaded"):
await hass.services.async_call(
DOMAIN,
"list_notifications",
{},
target={
"config_entry": config_entry.entry_id,
},
blocking=True,
return_response=True,
)
with pytest.raises(ValueError, match="Invalid config entry: does-not-exist"):
await hass.services.async_call(
DOMAIN,
"list_notifications",
{},
target={
"config_entry": "does-not-exist",
},
blocking=True,
return_response=True,
)