jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactoring flipr integration to prepare Hub device addition (#125262)

Browse Source 
* Addition of hub device

* coordinator udata updated after a hub action

* Unit tests update

* Unit tests improvements

* addition of tests on select and switch platforms

* wording

* Removal of select platform for PR containing only one platform

* Remove hub to maintain only the refactoring that prepare the hub device addition

* Review corrections

* wording

* Review corrections

* Review corrections

* Review corrections
This commit is contained in:
cnico 2024-09-11 23:34:29 +02:00 committed by GitHub
parent 0582c39d33
commit ee7bee2766
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 479 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
homeassistant/components/flipr/__init__.py
View File

@ -1,22 +1,59 @@
"""The Flipr integration."""
from collections import Counter
from dataclasses import dataclass
import logging
from flipr_api import FliprAPIRestClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers import issue_registry as ir
from .const import DOMAIN
from .coordinator import FliprDataUpdateCoordinator
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Flipr from a config entry."""
hass.data.setdefault(DOMAIN, {})
coordinator = FliprDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = coordinator
@dataclass
class FliprData:
"""The Flipr data class."""
flipr_coordinators: list[FliprDataUpdateCoordinator]
type FliprConfigEntry = ConfigEntry[FliprData]
async def async_setup_entry(hass: HomeAssistant, entry: FliprConfigEntry) -> bool:
"""Set up flipr from a config entry."""
# Detect invalid old config entry and raise error if found
detect_invalid_old_configuration(hass, entry)
config = entry.data
username = config[CONF_EMAIL]
password = config[CONF_PASSWORD]
_LOGGER.debug("Initializing Flipr client %s", username)
client = FliprAPIRestClient(username, password)
ids = await hass.async_add_executor_job(client.search_all_ids)
_LOGGER.debug("List of devices ids : %s", ids)
flipr_coordinators = []
for flipr_id in ids["flipr"]:
flipr_coordinator = FliprDataUpdateCoordinator(hass, client, flipr_id)
await flipr_coordinator.async_config_entry_first_refresh()
flipr_coordinators.append(flipr_coordinator)
entry.runtime_data = FliprData(flipr_coordinators)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
@ -25,9 +62,49 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
return unload_ok
def detect_invalid_old_configuration(hass: HomeAssistant, entry: ConfigEntry):
"""Detect invalid old configuration and raise error if found."""
def find_duplicate_entries(entries):
values = [e.data["email"] for e in entries]
_LOGGER.debug("Detecting duplicates in values : %s", values)
return any(count > 1 for count in Counter(values).values())
entries = hass.config_entries.async_entries(DOMAIN)
if find_duplicate_entries(entries):
ir.async_create_issue(
hass,
DOMAIN,
"duplicate_config",
breaks_in_ha_version="2025.4.0",
is_fixable=False,
severity=ir.IssueSeverity.ERROR,
translation_key="duplicate_config",
)
raise ConfigEntryError(
"Duplicate entries found for flipr with the same user email. Please remove one of it manually. Multiple fliprs will be automatically detected after restart."
)
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Migrate config entry."""
_LOGGER.debug("Migration of flipr config from version %s", entry.version)
if entry.version == 1:
# In version 1, we have flipr device as config entry unique id
# and one device per config entry.
# We need to migrate to a new config entry that may contain multiple devices.
# So we change the entry data to match config_flow evolution.
login = entry.data[CONF_EMAIL]
hass.config_entries.async_update_entry(entry, version=2, unique_id=login)
_LOGGER.debug("Migration of flipr config to version 2 successful")
return True

9
homeassistant/components/flipr/binary_sensor.py
View File

@ -7,11 +7,10 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from . import FliprConfigEntry
from .entity import FliprEntity
BINARY_SENSORS_TYPES: tuple[BinarySensorEntityDescription, ...] = (
@ -30,15 +29,17 @@ BINARY_SENSORS_TYPES: tuple[BinarySensorEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: FliprConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Defer sensor setup of flipr binary sensors."""
coordinator = hass.data[DOMAIN][config_entry.entry_id]
coordinators = config_entry.runtime_data.flipr_coordinators
async_add_entities(
FliprBinarySensor(coordinator, description)
for description in BINARY_SENSORS_TYPES
for coordinator in coordinators
)

109
homeassistant/components/flipr/config_flow.py
View File

@ -3,6 +3,7 @@
from __future__ import annotations
import logging
from typing import Any
from flipr_api import FliprAPIRestClient
from requests.exceptions import HTTPError, Timeout
@ -11,35 +12,37 @@ import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from .const import CONF_FLIPR_ID, DOMAIN
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_EMAIL): str,
vol.Required(CONF_PASSWORD): str,
}
)
class FliprConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Flipr."""
VERSION = 1
_username: str
_password: str
_flipr_id: str = ""
_possible_flipr_ids: list[str]
VERSION = 2
async def async_step_user(
self, user_input: dict[str, str] | None = None
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
if user_input is None:
return self._show_setup_form()
self._username = user_input[CONF_EMAIL]
self._password = user_input[CONF_PASSWORD]
errors: dict[str, str] = {}
if user_input is not None:
client = FliprAPIRestClient(
user_input[CONF_EMAIL], user_input[CONF_PASSWORD]
)
errors = {}
if not self._flipr_id:
try:
flipr_ids = await self._authenticate_and_search_flipr()
ids = await self.hass.async_add_executor_job(client.search_all_ids)
except HTTPError:
errors["base"] = "invalid_auth"
except (Timeout, ConnectionError):
@ -48,79 +51,25 @@ class FliprConfigFlow(ConfigFlow, domain=DOMAIN):
errors["base"] = "unknown"
_LOGGER.exception("Unexpected exception")
if not errors and not flipr_ids:
# No flipr_id found. Tell the user with an error message.
errors["base"] = "no_flipr_id_found"
if errors:
return self._show_setup_form(errors)
if len(flipr_ids) == 1:
self._flipr_id = flipr_ids[0]
else:
# If multiple flipr found (rare case), we ask the user to choose one in a select box.
# The user will have to run config_flow as many times as many fliprs he has.
self._possible_flipr_ids = flipr_ids
return await self.async_step_flipr_id()
_LOGGER.debug("Found flipr or hub ids : %s", ids)
# Check if already configured
await self.async_set_unique_id(self._flipr_id)
if len(ids["flipr"]) > 0 or len(ids["hub"]) > 0:
# If there is a flipr or hub, we can create a config entry.
await self.async_set_unique_id(user_input[CONF_EMAIL])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self._flipr_id,
data={
CONF_EMAIL: self._username,
CONF_PASSWORD: self._password,
CONF_FLIPR_ID: self._flipr_id,
},
title=f"Flipr {user_input[CONF_EMAIL]}",
data=user_input,
)
def _show_setup_form(self, errors=None):
"""Show the setup form to the user."""
# if no flipr or hub found. Tell the user with an error message.
errors["base"] = "no_flipr_id_found"
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
data_schema=DATA_SCHEMA,
errors=errors,
)
async def _authenticate_and_search_flipr(self) -> list[str]:
"""Validate the username and password provided and searches for a flipr id."""
# Instantiates the flipr API that does not require async since it is has no network access.
client = FliprAPIRestClient(self._username, self._password)
return await self.hass.async_add_executor_job(client.search_flipr_ids)
async def async_step_flipr_id(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
if not user_input:
# Creation of a select with the proposal of flipr ids values found by API.
flipr_ids_for_form = {}
for flipr_id in self._possible_flipr_ids:
flipr_ids_for_form[flipr_id] = f"{flipr_id}"
return self.async_show_form(
step_id="flipr_id",
data_schema=vol.Schema(
{
vol.Required(CONF_FLIPR_ID): vol.All(
vol.Coerce(str), vol.In(flipr_ids_for_form)
)
}
),
)
# Get chosen flipr_id.
self._flipr_id = user_input[CONF_FLIPR_ID]
return await self.async_step_user(
{
CONF_EMAIL: self._username,
CONF_PASSWORD: self._password,
CONF_FLIPR_ID: self._flipr_id,
}
)

4
homeassistant/components/flipr/const.py
View File

@ -2,9 +2,9 @@
DOMAIN = "flipr"
CONF_FLIPR_ID = "flipr_id"
ATTRIBUTION = "Flipr Data"
MANUFACTURER = "CTAC-TECH"
NAME = "Flipr"
CONF_ENTRY_FLIPR_COORDINATORS = "flipr_coordinators"

26
homeassistant/components/flipr/coordinator.py
View File

@ -6,39 +6,37 @@ import logging
from flipr_api import FliprAPIRestClient
from flipr_api.exceptions import FliprError
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_FLIPR_ID
_LOGGER = logging.getLogger(__name__)
class FliprDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to hold Flipr data retrieval."""
def __init__(self, hass, entry):
"""Initialize."""
username = entry.data[CONF_EMAIL]
password = entry.data[CONF_PASSWORD]
self.flipr_id = entry.data[CONF_FLIPR_ID]
config_entry: ConfigEntry
# Establishes the connection.
self.client = FliprAPIRestClient(username, password)
self.entry = entry
def __init__(
self, hass: HomeAssistant, client: FliprAPIRestClient, flipr_or_hub_id: str
) -> None:
"""Initialize."""
self.device_id = flipr_or_hub_id
self.client = client
super().__init__(
hass,
_LOGGER,
name=f"Flipr data measure for {self.flipr_id}",
update_interval=timedelta(minutes=60),
name=f"Flipr or Hub data measure for {self.device_id}",
update_interval=timedelta(minutes=15),
)
async def _async_update_data(self):
"""Fetch data from API endpoint."""
try:
data = await self.hass.async_add_executor_job(
self.client.get_pool_measure_latest, self.flipr_id
self.client.get_pool_measure_latest, self.device_id
)
except FliprError as error:
raise UpdateFailed(error) from error

24
homeassistant/components/flipr/entity.py
View File

@ -2,12 +2,10 @@
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import ATTRIBUTION, CONF_FLIPR_ID, DOMAIN, MANUFACTURER
from .const import ATTRIBUTION, DOMAIN, MANUFACTURER
from .coordinator import FliprDataUpdateCoordinator
class FliprEntity(CoordinatorEntity):
@ -17,17 +15,21 @@ class FliprEntity(CoordinatorEntity):
_attr_has_entity_name = True
def __init__(
self, coordinator: DataUpdateCoordinator, description: EntityDescription
self,
coordinator: FliprDataUpdateCoordinator,
description: EntityDescription,
is_flipr_hub: bool = False,
) -> None:
"""Initialize Flipr sensor."""
super().__init__(coordinator)
self.device_id = coordinator.device_id
self.entity_description = description
if coordinator.config_entry:
flipr_id = coordinator.config_entry.data[CONF_FLIPR_ID]
self._attr_unique_id = f"{flipr_id}-{description.key}"
self._attr_unique_id = f"{self.device_id}-{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, flipr_id)},
identifiers={(DOMAIN, self.device_id)},
manufacturer=MANUFACTURER,
name=f"Flipr {flipr_id}",
name=f"Flipr hub {self.device_id}"
if is_flipr_hub
else f"Flipr {self.device_id}",
)

14
homeassistant/components/flipr/sensor.py
View File

@ -8,12 +8,11 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, UnitOfElectricPotential, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from . import FliprConfigEntry
from .entity import FliprEntity
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
@ -57,14 +56,17 @@ SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: FliprConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Defer sensor setup to the shared sensor module."""
coordinator = hass.data[DOMAIN][config_entry.entry_id]
coordinators = config_entry.runtime_data.flipr_coordinators
sensors = [FliprSensor(coordinator, description) for description in SENSOR_TYPES]
async_add_entities(sensors)
async_add_entities(
FliprSensor(coordinator, description)
for description in SENSOR_TYPES
for coordinator in coordinators
)
class FliprSensor(FliprEntity, SensorEntity):

18
homeassistant/components/flipr/strings.json
View File

@ -8,23 +8,13 @@
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"flipr_id": {
"title": "Choose your Flipr",
"description": "Choose your Flipr ID in the list",
"data": {
"flipr_id": "Flipr ID"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"no_flipr_id_found": "No flipr id associated to your account for now. You should verify it is working with the Flipr's mobile app first."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
"no_flipr_id_found": "No flipr or hub associated to your account for now. You should verify it is working with the Flipr's mobile app first."
}
},
"entity": {
@ -50,5 +40,11 @@
"name": "Red OX"
}
}
},
"issues": {
"duplicate_config": {
"title": "Multiple flipr configurations with the same account",
"description": "The Flipr integration has been updated to work account based rather than device based. This means that if you have 2 devices, you only need one configuration. For every account you have, please delete all but one configuration and restart Home Assistant for it to set up the devices linked to your account."
}
}
}

14
tests/components/flipr/__init__.py
View File

@ -1 +1,15 @@
"""Tests for the Flipr integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Fixture for setting up the component."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()

97
tests/components/flipr/conftest.py Normal file
View File

@ -0,0 +1,97 @@
"""Common fixtures for the flipr tests."""
from collections.abc import Generator
from datetime import datetime
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.flipr.const import DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry
# Data for the mocked object returned via flipr_api client.
MOCK_DATE_TIME = datetime(2021, 2, 15, 9, 10, 32, tzinfo=dt_util.UTC)
MOCK_FLIPR_MEASURE = {
"temperature": 10.5,
"ph": 7.03,
"chlorine": 0.23654886,
"red_ox": 657.58,
"date_time": MOCK_DATE_TIME,
"ph_status": "TooLow",
"chlorine_status": "Medium",
"battery": 95.0,
}
MOCK_HUB_STATE_ON = {
"state": True,
"mode": "planning",
"planning": "dummyplanningid",
}
MOCK_HUB_STATE_OFF = {
"state": False,
"mode": "manual",
"planning": "dummyplanningid",
}
MOCK_HUB_MODE_MANUAL = {
"state": False,
"mode": "manual",
"planning": "dummyplanningid",
}
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.flipr.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock the config entry."""
return MockConfigEntry(
version=2,
domain=DOMAIN,
unique_id="toto@toto.com",
data={
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
},
)
@pytest.fixture
def mock_flipr_client() -> Generator[AsyncMock]:
"""Mock a Flipr client."""
with (
patch(
"homeassistant.components.flipr.FliprAPIRestClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.flipr.config_flow.FliprAPIRestClient",
new=mock_client,
),
):
client = mock_client.return_value
# Default values for the tests using this mock :
client.search_all_ids.return_value = {"flipr": ["myfliprid"], "hub": []}
client.get_pool_measure_latest.return_value = MOCK_FLIPR_MEASURE
client.get_hub_state.return_value = MOCK_HUB_STATE_ON
client.set_hub_state.return_value = MOCK_HUB_STATE_ON
client.set_hub_mode.return_value = MOCK_HUB_MODE_MANUAL
yield client

45
tests/components/flipr/test_binary_sensor.py
View File

@ -1,49 +1,24 @@
"""Test the Flipr binary sensor."""
from datetime import datetime
from unittest.mock import patch
from unittest.mock import AsyncMock
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import setup_integration
from tests.common import MockConfigEntry
# Data for the mocked object returned via flipr_api client.
MOCK_DATE_TIME = datetime(2021, 2, 15, 9, 10, 32, tzinfo=dt_util.UTC)
MOCK_FLIPR_MEASURE = {
"temperature": 10.5,
"ph": 7.03,
"chlorine": 0.23654886,
"red_ox": 657.58,
"date_time": MOCK_DATE_TIME,
"ph_status": "TooLow",
"chlorine_status": "Medium",
}
async def test_sensors(hass: HomeAssistant, entity_registry: er.EntityRegistry) -> None:
async def test_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
mock_flipr_client: AsyncMock,
) -> None:
"""Test the creation and values of the Flipr binary sensors."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="test_entry_unique_id",
data={
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
CONF_FLIPR_ID: "myfliprid",
},
)
entry.add_to_hass(hass)
with patch(
"flipr_api.FliprAPIRestClient.get_pool_measure_latest",
return_value=MOCK_FLIPR_MEASURE,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await setup_integration(hass, mock_config_entry)
# Check entity unique_id value that is generated in FliprEntity base class.
entity = entity_registry.async_get("binary_sensor.flipr_myfliprid_ph_status")

184
tests/components/flipr/test_config_flow.py
View File

@ -1,169 +1,131 @@
"""Test the Flipr config flow."""
from unittest.mock import patch
from unittest.mock import AsyncMock
import pytest
from requests.exceptions import HTTPError, Timeout
from homeassistant import config_entries
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.components.flipr.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@pytest.fixture(name="mock_setup")
def mock_setups():
"""Prevent setup."""
with patch(
"homeassistant.components.flipr.async_setup_entry",
return_value=True,
):
yield
async def test_show_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
async def test_full_flow(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_flipr_client: AsyncMock
) -> None:
"""Test the full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == config_entries.SOURCE_USER
assert result["step_id"] == "user"
assert not result["errors"]
async def test_invalid_credential(hass: HomeAssistant, mock_setup) -> None:
"""Test invalid credential."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids", side_effect=HTTPError()
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "bad_login",
CONF_PASSWORD: "bad_pass",
CONF_FLIPR_ID: "",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_auth"}
async def test_nominal_case(hass: HomeAssistant, mock_setup) -> None:
"""Test valid login form."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
return_value=["flipid"],
) as mock_flipr_client:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
context={"source": SOURCE_USER},
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "flipid",
},
)
await hass.async_block_till_done()
assert len(mock_flipr_client.mock_calls) == 1
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "flipid"
assert result["title"] == "Flipr dummylogin"
assert result["result"].unique_id == "dummylogin"
assert result["data"] == {
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "flipid",
}
async def test_multiple_flip_id(hass: HomeAssistant, mock_setup) -> None:
"""Test multiple flipr id adding a config step."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
return_value=["FLIP1", "FLIP2"],
) as mock_flipr_client:
@pytest.mark.parametrize(
("exception", "expected"),
[
(Exception("Bad request Boy :) --"), {"base": "unknown"}),
(HTTPError, {"base": "invalid_auth"}),
(Timeout, {"base": "cannot_connect"}),
(ConnectionError, {"base": "cannot_connect"}),
],
)
async def test_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_flipr_client: AsyncMock,
exception: Exception,
expected: dict[str, str],
) -> None:
"""Test we handle any error."""
mock_flipr_client.search_all_ids.side_effect = exception
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
context={"source": SOURCE_USER},
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_EMAIL: "nada",
CONF_PASSWORD: "nadap",
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "flipr_id"
assert result["step_id"] == "user"
assert result["errors"] == expected
# Test of recover in normal state after correction of the 1st error
mock_flipr_client.search_all_ids.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_FLIPR_ID: "FLIP2"},
{
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
},
)
assert len(mock_flipr_client.mock_calls) == 1
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "FLIP2"
assert result["title"] == "Flipr dummylogin"
assert result["data"] == {
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "FLIP2",
}
async def test_no_flip_id(hass: HomeAssistant, mock_setup) -> None:
"""Test no flipr id found."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
return_value=[],
) as mock_flipr_client:
async def test_no_flipr_found(
hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_flipr_client: AsyncMock
) -> None:
"""Test the case where there is no flipr found."""
mock_flipr_client.search_all_ids.return_value = {"flipr": [], "hub": []}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
context={"source": SOURCE_USER},
data={
CONF_EMAIL: "nada",
CONF_PASSWORD: "nadap",
},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "no_flipr_id_found"}
# Test of recover in normal state after correction of the 1st error
mock_flipr_client.search_all_ids.return_value = {"flipr": ["myfliprid"], "hub": []}
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
},
)
assert result["step_id"] == "user"
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "no_flipr_id_found"}
assert len(mock_flipr_client.mock_calls) == 1
async def test_http_errors(hass: HomeAssistant, mock_setup) -> None:
"""Test HTTP Errors."""
with patch("flipr_api.FliprAPIRestClient.search_flipr_ids", side_effect=Timeout()):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "nada",
CONF_PASSWORD: "nada",
CONF_FLIPR_ID: "",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
side_effect=Exception("Bad request Boy :) --"),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "nada",
CONF_PASSWORD: "nada",
CONF_FLIPR_ID: "",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "unknown"}
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Flipr dummylogin"
assert result["data"] == {
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
}

87
tests/components/flipr/test_init.py
View File

@ -1,29 +1,90 @@
"""Tests for init methods."""
from unittest.mock import patch
from unittest.mock import AsyncMock
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.components.flipr.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_unload_entry(hass: HomeAssistant) -> None:
async def test_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_flipr_client: AsyncMock,
) -> None:
"""Test unload entry."""
entry = MockConfigEntry(
mock_flipr_client.search_all_ids.return_value = {
"flipr": ["myfliprid"],
"hub": ["hubid"],
}
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_duplicate_config_entries(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_flipr_client: AsyncMock,
) -> None:
"""Test duplicate config entries."""
mock_config_entry_dup = MockConfigEntry(
version=2,
domain=DOMAIN,
unique_id="toto@toto.com",
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "FLIP1",
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
"flipr_id": "myflipr_id_dup",
},
unique_id="123456",
)
entry.add_to_hass(hass)
with patch("homeassistant.components.flipr.coordinator.FliprAPIRestClient"):
await hass.config_entries.async_setup(entry.entry_id)
mock_config_entry.add_to_hass(hass)
# Initialize the first entry with default mock
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_unload(entry.entry_id)
assert entry.state is ConfigEntryState.NOT_LOADED
# Initialize the second entry with another flipr id
mock_config_entry_dup.add_to_hass(hass)
assert not await hass.config_entries.async_setup(mock_config_entry_dup.entry_id)
await hass.async_block_till_done()
assert mock_config_entry_dup.state is ConfigEntryState.SETUP_ERROR
async def test_migrate_entry(
hass: HomeAssistant,
mock_flipr_client: AsyncMock,
) -> None:
"""Test migrate config entry from v1 to v2."""
mock_config_entry_v1 = MockConfigEntry(
version=1,
domain=DOMAIN,
title="myfliprid",
unique_id="test_entry_unique_id",
data={
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
"flipr_id": "myfliprid",
},
)
await setup_integration(hass, mock_config_entry_v1)
assert mock_config_entry_v1.state is ConfigEntryState.LOADED
assert mock_config_entry_v1.version == 2
assert mock_config_entry_v1.unique_id == "toto@toto.com"
assert mock_config_entry_v1.data == {
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
"flipr_id": "myfliprid",
}

80
tests/components/flipr/test_sensor.py
View File

@ -1,59 +1,28 @@
"""Test the Flipr sensor."""
from datetime import datetime
from unittest.mock import patch
from unittest.mock import AsyncMock
from flipr_api.exceptions import FliprError
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.components.sensor import ATTR_STATE_CLASS, SensorStateClass
from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
CONF_EMAIL,
CONF_PASSWORD,
PERCENTAGE,
UnitOfTemperature,
)
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, PERCENTAGE, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.util import dt as dt_util
from . import setup_integration
from tests.common import MockConfigEntry
# Data for the mocked object returned via flipr_api client.
MOCK_DATE_TIME = datetime(2021, 2, 15, 9, 10, 32, tzinfo=dt_util.UTC)
MOCK_FLIPR_MEASURE = {
"temperature": 10.5,
"ph": 7.03,
"chlorine": 0.23654886,
"red_ox": 657.58,
"date_time": MOCK_DATE_TIME,
"ph_status": "TooLow",
"chlorine_status": "Medium",
"battery": 95.0,
}
async def test_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
mock_flipr_client: AsyncMock,
) -> None:
"""Test the creation and values of the Flipr binary sensors."""
async def test_sensors(hass: HomeAssistant, entity_registry: er.EntityRegistry) -> None:
"""Test the creation and values of the Flipr sensors."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="test_entry_unique_id",
data={
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
CONF_FLIPR_ID: "myfliprid",
},
)
entry.add_to_hass(hass)
with patch(
"flipr_api.FliprAPIRestClient.get_pool_measure_latest",
return_value=MOCK_FLIPR_MEASURE,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await setup_integration(hass, mock_config_entry)
# Check entity unique_id value that is generated in FliprEntity base class.
entity = entity_registry.async_get("sensor.flipr_myfliprid_red_ox")
@ -97,27 +66,18 @@ async def test_sensors(hass: HomeAssistant, entity_registry: er.EntityRegistry)
async def test_error_flipr_api_sensors(
hass: HomeAssistant, entity_registry: er.EntityRegistry
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
mock_flipr_client: AsyncMock,
) -> None:
"""Test the Flipr sensors error."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="test_entry_unique_id",
data={
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
CONF_FLIPR_ID: "myfliprid",
},
mock_flipr_client.get_pool_measure_latest.side_effect = FliprError(
"Error during flipr data retrieval..."
)
entry.add_to_hass(hass)
with patch(
"flipr_api.FliprAPIRestClient.get_pool_measure_latest",
side_effect=FliprError("Error during flipr data retrieval..."),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await setup_integration(hass, mock_config_entry)
# Check entity is not generated because of the FliprError raised.
entity = entity_registry.async_get("sensor.flipr_myfliprid_red_ox")