mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 20:57:21 +00:00
Code quality changes for LaCrosse View (#76265)
This commit is contained in:
parent
21ebd1f612
commit
92a9011953
@ -1,18 +1,16 @@
|
||||
"""The LaCrosse View integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from lacrosse_view import LaCrosse, Location, LoginError, Sensor
|
||||
from lacrosse_view import LaCrosse, LoginError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
||||
|
||||
from .const import DOMAIN, LOGGER, SCAN_INTERVAL
|
||||
from .const import DOMAIN
|
||||
from .coordinator import LaCrosseUpdateCoordinator
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
@ -20,52 +18,22 @@ PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up LaCrosse View from a config entry."""
|
||||
|
||||
async def get_data() -> list[Sensor]:
|
||||
"""Get the data from the LaCrosse View."""
|
||||
now = datetime.utcnow()
|
||||
|
||||
if hass.data[DOMAIN][entry.entry_id]["last_update"] < now - timedelta(
|
||||
minutes=59
|
||||
): # Get new token
|
||||
hass.data[DOMAIN][entry.entry_id]["last_update"] = now
|
||||
await api.login(entry.data["username"], entry.data["password"])
|
||||
|
||||
# Get the timestamp for yesterday at 6 PM (this is what is used in the app, i noticed it when proxying the request)
|
||||
yesterday = now - timedelta(days=1)
|
||||
yesterday = yesterday.replace(hour=18, minute=0, second=0, microsecond=0)
|
||||
yesterday_timestamp = datetime.timestamp(yesterday)
|
||||
|
||||
return await api.get_sensors(
|
||||
location=Location(id=entry.data["id"], name=entry.data["name"]),
|
||||
tz=hass.config.time_zone,
|
||||
start=str(int(yesterday_timestamp)),
|
||||
end=str(int(datetime.timestamp(now))),
|
||||
)
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
|
||||
"api": LaCrosse(async_get_clientsession(hass)),
|
||||
"last_update": datetime.utcnow(),
|
||||
}
|
||||
api: LaCrosse = hass.data[DOMAIN][entry.entry_id]["api"]
|
||||
api = LaCrosse(async_get_clientsession(hass))
|
||||
|
||||
try:
|
||||
await api.login(entry.data["username"], entry.data["password"])
|
||||
except LoginError as error:
|
||||
raise ConfigEntryNotReady from error
|
||||
raise ConfigEntryAuthFailed from error
|
||||
|
||||
coordinator = DataUpdateCoordinator(
|
||||
hass,
|
||||
LOGGER,
|
||||
name="LaCrosse View",
|
||||
update_method=get_data,
|
||||
update_interval=timedelta(seconds=SCAN_INTERVAL),
|
||||
)
|
||||
coordinator = LaCrosseUpdateCoordinator(hass, api, entry)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data[DOMAIN][entry.entry_id]["coordinator"] = coordinator
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
|
||||
"coordinator": coordinator,
|
||||
}
|
||||
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
@ -1,7 +1,7 @@
|
||||
"""Config flow for LaCrosse View integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
from lacrosse_view import LaCrosse, Location, LoginError
|
||||
@ -13,9 +13,7 @@ from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
from .const import DOMAIN, LOGGER
|
||||
|
||||
STEP_USER_DATA_SCHEMA = vol.Schema(
|
||||
{
|
||||
@ -47,8 +45,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for LaCrosse View."""
|
||||
|
||||
VERSION = 1
|
||||
data: dict[str, str] = {}
|
||||
locations: list[Location] = []
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the config flow."""
|
||||
self.data: dict[str, str] = {}
|
||||
self.locations: list[Location] = []
|
||||
self._reauth_entry: config_entries.ConfigEntry | None = None
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
@ -68,11 +70,19 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
except NoLocations:
|
||||
errors["base"] = "no_locations"
|
||||
except Exception: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
self.data = user_input
|
||||
self.locations = info
|
||||
|
||||
# Check if we are reauthenticating
|
||||
if self._reauth_entry is not None:
|
||||
self.hass.config_entries.async_update_entry(
|
||||
self._reauth_entry, data=self._reauth_entry.data | self.data
|
||||
)
|
||||
await self.hass.config_entries.async_reload(self._reauth_entry.entry_id)
|
||||
return self.async_abort(reason="reauth_successful")
|
||||
return await self.async_step_location()
|
||||
|
||||
return self.async_show_form(
|
||||
@ -98,11 +108,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
location_id = user_input["location"]
|
||||
|
||||
for location in self.locations:
|
||||
if location.id == location_id:
|
||||
location_name = location.name
|
||||
location_name = next(
|
||||
location.name for location in self.locations if location.id == location_id
|
||||
)
|
||||
|
||||
await self.async_set_unique_id(location_id)
|
||||
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(
|
||||
@ -115,9 +126,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class CannotConnect(HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
|
||||
"""Reauth in case of a password change or other error."""
|
||||
self._reauth_entry = self.hass.config_entries.async_get_entry(
|
||||
self.context["entry_id"]
|
||||
)
|
||||
return await self.async_step_user()
|
||||
|
||||
|
||||
class InvalidAuth(HomeAssistantError):
|
||||
@ -126,3 +140,7 @@ class InvalidAuth(HomeAssistantError):
|
||||
|
||||
class NoLocations(HomeAssistantError):
|
||||
"""Error to indicate there are no locations."""
|
||||
|
||||
|
||||
class NonExistentEntry(HomeAssistantError):
|
||||
"""Error to indicate that the entry does not exist when it should."""
|
||||
|
79
homeassistant/components/lacrosse_view/coordinator.py
Normal file
79
homeassistant/components/lacrosse_view/coordinator.py
Normal file
@ -0,0 +1,79 @@
|
||||
"""DataUpdateCoordinator for LaCrosse View."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from lacrosse_view import HTTPError, LaCrosse, Location, LoginError, Sensor
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
|
||||
|
||||
from .const import LOGGER, SCAN_INTERVAL
|
||||
|
||||
|
||||
class LaCrosseUpdateCoordinator(DataUpdateCoordinator[list[Sensor]]):
|
||||
"""DataUpdateCoordinator for LaCrosse View."""
|
||||
|
||||
username: str
|
||||
password: str
|
||||
name: str
|
||||
id: str
|
||||
hass: HomeAssistant
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
api: LaCrosse,
|
||||
entry: ConfigEntry,
|
||||
) -> None:
|
||||
"""Initialize DataUpdateCoordinator for LaCrosse View."""
|
||||
self.api = api
|
||||
self.last_update = datetime.utcnow()
|
||||
self.username = entry.data["username"]
|
||||
self.password = entry.data["password"]
|
||||
self.hass = hass
|
||||
self.name = entry.data["name"]
|
||||
self.id = entry.data["id"]
|
||||
super().__init__(
|
||||
hass,
|
||||
LOGGER,
|
||||
name="LaCrosse View",
|
||||
update_interval=timedelta(seconds=SCAN_INTERVAL),
|
||||
)
|
||||
|
||||
async def _async_update_data(self) -> list[Sensor]:
|
||||
"""Get the data for LaCrosse View."""
|
||||
now = datetime.utcnow()
|
||||
|
||||
if self.last_update < now - timedelta(minutes=59): # Get new token
|
||||
self.last_update = now
|
||||
try:
|
||||
await self.api.login(self.username, self.password)
|
||||
except LoginError as error:
|
||||
raise ConfigEntryAuthFailed from error
|
||||
|
||||
# Get the timestamp for yesterday at 6 PM (this is what is used in the app, i noticed it when proxying the request)
|
||||
yesterday = now - timedelta(days=1)
|
||||
yesterday = yesterday.replace(hour=18, minute=0, second=0, microsecond=0)
|
||||
yesterday_timestamp = datetime.timestamp(yesterday)
|
||||
|
||||
try:
|
||||
sensors = await self.api.get_sensors(
|
||||
location=Location(id=self.id, name=self.name),
|
||||
tz=self.hass.config.time_zone,
|
||||
start=str(int(yesterday_timestamp)),
|
||||
end=str(int(datetime.timestamp(now))),
|
||||
)
|
||||
except HTTPError as error:
|
||||
raise ConfigEntryNotReady from error
|
||||
|
||||
# Verify that we have permission to read the sensors
|
||||
for sensor in sensors:
|
||||
if not sensor.permissions.get("read", False):
|
||||
raise ConfigEntryAuthFailed(
|
||||
f"This account does not have permission to read {sensor.name}"
|
||||
)
|
||||
|
||||
return sensors
|
@ -3,16 +3,23 @@ from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from re import sub
|
||||
|
||||
from lacrosse_view import Sensor
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
PERCENTAGE,
|
||||
PRECIPITATION_INCHES,
|
||||
SPEED_KILOMETERS_PER_HOUR,
|
||||
TEMP_CELSIUS,
|
||||
TEMP_FAHRENHEIT,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
@ -27,7 +34,7 @@ from .const import DOMAIN, LOGGER
|
||||
class LaCrosseSensorEntityDescriptionMixin:
|
||||
"""Mixin for required keys."""
|
||||
|
||||
value_fn: Callable[..., float]
|
||||
value_fn: Callable[[Sensor, str], float]
|
||||
|
||||
|
||||
@dataclass
|
||||
@ -37,20 +44,51 @@ class LaCrosseSensorEntityDescription(
|
||||
"""Description for LaCrosse View sensor."""
|
||||
|
||||
|
||||
def get_value(sensor: Sensor, field: str) -> float:
|
||||
"""Get the value of a sensor field."""
|
||||
return float(sensor.data[field]["values"][-1]["s"])
|
||||
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
ICON_LIST = {
|
||||
"Temperature": "mdi:thermometer",
|
||||
"Humidity": "mdi:water-percent",
|
||||
"HeatIndex": "mdi:thermometer",
|
||||
"WindSpeed": "mdi:weather-windy",
|
||||
"Rain": "mdi:water",
|
||||
}
|
||||
UNIT_LIST = {
|
||||
"degrees_celsius": "°C",
|
||||
"degrees_fahrenheit": "°F",
|
||||
"relative_humidity": "%",
|
||||
"kilometers_per_hour": "km/h",
|
||||
"inches": "in",
|
||||
SENSOR_DESCRIPTIONS = {
|
||||
"Temperature": LaCrosseSensorEntityDescription(
|
||||
key="Temperature",
|
||||
device_class=SensorDeviceClass.TEMPERATURE,
|
||||
name="Temperature",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
value_fn=get_value,
|
||||
native_unit_of_measurement=TEMP_CELSIUS,
|
||||
),
|
||||
"Humidity": LaCrosseSensorEntityDescription(
|
||||
key="Humidity",
|
||||
device_class=SensorDeviceClass.HUMIDITY,
|
||||
name="Humidity",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
value_fn=get_value,
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
),
|
||||
"HeatIndex": LaCrosseSensorEntityDescription(
|
||||
key="HeatIndex",
|
||||
device_class=SensorDeviceClass.TEMPERATURE,
|
||||
name="Heat Index",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
value_fn=get_value,
|
||||
native_unit_of_measurement=TEMP_FAHRENHEIT,
|
||||
),
|
||||
"WindSpeed": LaCrosseSensorEntityDescription(
|
||||
key="WindSpeed",
|
||||
name="Wind Speed",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
value_fn=get_value,
|
||||
native_unit_of_measurement=SPEED_KILOMETERS_PER_HOUR,
|
||||
),
|
||||
"Rain": LaCrosseSensorEntityDescription(
|
||||
key="Rain",
|
||||
name="Rain",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
value_fn=get_value,
|
||||
native_unit_of_measurement=PRECIPITATION_INCHES,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@ -66,35 +104,26 @@ async def async_setup_entry(
|
||||
sensors: list[Sensor] = coordinator.data
|
||||
|
||||
sensor_list = []
|
||||
for i, sensor in enumerate(sensors):
|
||||
if not sensor.permissions.get("read"):
|
||||
LOGGER.warning(
|
||||
"No permission to read sensor %s, are you sure you're signed into the right account?",
|
||||
sensor.name,
|
||||
)
|
||||
continue
|
||||
for sensor in sensors:
|
||||
for field in sensor.sensor_field_names:
|
||||
description = SENSOR_DESCRIPTIONS.get(field)
|
||||
if description is None:
|
||||
message = (
|
||||
f"Unsupported sensor field: {field}\nPlease create an issue on "
|
||||
"GitHub. https://github.com/home-assistant/core/issues/new?assignees=&la"
|
||||
"bels=&template=bug_report.yml&integration_name=LaCrosse%20View&integrat"
|
||||
"ion_link=https://www.home-assistant.io/integrations/lacrosse_view/&addi"
|
||||
f"tional_information=Field:%20{field}%0ASensor%20Model:%20{sensor.model}&"
|
||||
f"title=LaCrosse%20View%20Unsupported%20sensor%20field:%20{field}"
|
||||
)
|
||||
|
||||
LOGGER.warning(message)
|
||||
continue
|
||||
sensor_list.append(
|
||||
LaCrosseViewSensor(
|
||||
coordinator=coordinator,
|
||||
description=LaCrosseSensorEntityDescription(
|
||||
key=str(i),
|
||||
device_class="temperature" if field == "Temperature" else None,
|
||||
# The regex is to convert CamelCase to Human Case
|
||||
# e.g. "RelativeHumidity" -> "Relative Humidity"
|
||||
name=f"{sensor.name} {sub(r'(?<!^)(?=[A-Z])', ' ', field)}"
|
||||
if field
|
||||
!= sensor.name # If the name is the same as the field, don't include it.
|
||||
else sub(r"(?<!^)(?=[A-Z])", " ", field),
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
value_fn=lambda sensor=sensor, field=field: float(
|
||||
sensor.data[field]["values"][-1]["s"]
|
||||
),
|
||||
native_unit_of_measurement=UNIT_LIST.get(
|
||||
sensor.data[field]["unit"], None
|
||||
),
|
||||
),
|
||||
field=field,
|
||||
description=description,
|
||||
sensor=sensor,
|
||||
)
|
||||
)
|
||||
|
||||
@ -112,16 +141,14 @@ class LaCrosseViewSensor(
|
||||
self,
|
||||
description: LaCrosseSensorEntityDescription,
|
||||
coordinator: DataUpdateCoordinator[list[Sensor]],
|
||||
field: str,
|
||||
sensor: Sensor,
|
||||
) -> None:
|
||||
"""Initialize."""
|
||||
super().__init__(coordinator)
|
||||
sensor = self.coordinator.data[int(description.key)]
|
||||
|
||||
self.entity_description = description
|
||||
self._attr_unique_id = f"{sensor.location.id}-{description.key}-{field}"
|
||||
self._attr_unique_id = f"{sensor.sensor_id}-{description.key}"
|
||||
self._attr_name = f"{sensor.location.name} {description.name}"
|
||||
self._attr_icon = ICON_LIST.get(field, "mdi:thermometer")
|
||||
self._attr_device_info = {
|
||||
"identifiers": {(DOMAIN, sensor.sensor_id)},
|
||||
"name": sensor.name.split(" ")[0],
|
||||
@ -129,8 +156,11 @@ class LaCrosseViewSensor(
|
||||
"model": sensor.model,
|
||||
"via_device": (DOMAIN, sensor.location.id),
|
||||
}
|
||||
self._sensor = sensor
|
||||
|
||||
@property
|
||||
def native_value(self) -> float:
|
||||
def native_value(self) -> float | str:
|
||||
"""Return the sensor value."""
|
||||
return self.entity_description.value_fn()
|
||||
return self.entity_description.value_fn(
|
||||
self._sensor, self.entity_description.key
|
||||
)
|
||||
|
@ -14,7 +14,8 @@
|
||||
"no_locations": "No locations found"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -1,7 +1,8 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured"
|
||||
"already_configured": "Device is already configured",
|
||||
"reauth_successful": "Re-authentication was successful"
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "Invalid authentication",
|
||||
|
@ -30,3 +30,14 @@ TEST_NO_PERMISSION_SENSOR = Sensor(
|
||||
permissions={"read": False},
|
||||
model="Test",
|
||||
)
|
||||
TEST_UNSUPPORTED_SENSOR = Sensor(
|
||||
name="Test",
|
||||
device_id="1",
|
||||
type="Test",
|
||||
sensor_id="2",
|
||||
sensor_field_names=["SomeUnsupportedField"],
|
||||
location=Location(id="1", name="Test"),
|
||||
data={"SomeUnsupportedField": {"values": [{"s": "2"}], "unit": "degrees_celsius"}},
|
||||
permissions={"read": True},
|
||||
model="Test",
|
||||
)
|
||||
|
@ -6,7 +6,9 @@ from lacrosse_view import Location, LoginError
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.lacrosse_view.const import DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, FlowResultType
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_form(hass: HomeAssistant) -> None:
|
||||
@ -20,6 +22,8 @@ async def test_form(hass: HomeAssistant) -> None:
|
||||
with patch("lacrosse_view.LaCrosse.login", return_value=True,), patch(
|
||||
"lacrosse_view.LaCrosse.get_locations",
|
||||
return_value=[Location(id=1, name="Test")],
|
||||
), patch(
|
||||
"homeassistant.components.lacrosse_view.async_setup_entry", return_value=True
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
@ -46,7 +50,7 @@ async def test_form(hass: HomeAssistant) -> None:
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result3["type"] == FlowResultType.CREATE_ENTRY
|
||||
assert result3["title"] == "Test"
|
||||
assert result3["data"] == {
|
||||
"username": "test-username",
|
||||
@ -161,3 +165,116 @@ async def test_form_unexpected_error(hass: HomeAssistant) -> None:
|
||||
|
||||
assert result2["type"] == FlowResultType.FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_already_configured_device(hass: HomeAssistant) -> None:
|
||||
"""Test we handle invalid auth."""
|
||||
mock_config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"id": "1",
|
||||
"name": "Test",
|
||||
},
|
||||
unique_id="1",
|
||||
)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
# Now that we did the config once, let's try to do it again, this should raise the abort for already configured device
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["errors"] is None
|
||||
|
||||
with patch("lacrosse_view.LaCrosse.login", return_value=True,), patch(
|
||||
"lacrosse_view.LaCrosse.get_locations",
|
||||
return_value=[Location(id=1, name="Test")],
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == FlowResultType.FORM
|
||||
assert result2["step_id"] == "location"
|
||||
assert result2["errors"] is None
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.lacrosse_view.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result2["flow_id"],
|
||||
{
|
||||
"location": "1",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result3["type"] == FlowResultType.ABORT
|
||||
assert result3["reason"] == "already_configured"
|
||||
assert len(mock_setup_entry.mock_calls) == 0
|
||||
|
||||
|
||||
async def test_reauth(hass: HomeAssistant) -> None:
|
||||
"""Test reauthentication."""
|
||||
data = {
|
||||
"username": "test-username",
|
||||
"password": "test-password",
|
||||
"id": "1",
|
||||
"name": "Test",
|
||||
}
|
||||
mock_config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data=data,
|
||||
unique_id="1",
|
||||
title="Test",
|
||||
)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={
|
||||
"source": config_entries.SOURCE_REAUTH,
|
||||
"entry_id": mock_config_entry.entry_id,
|
||||
"title_placeholders": {"name": mock_config_entry.title},
|
||||
"unique_id": mock_config_entry.unique_id,
|
||||
},
|
||||
data=data,
|
||||
)
|
||||
assert result["type"] == FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
new_username = "new-username"
|
||||
new_password = "new-password"
|
||||
|
||||
with patch("lacrosse_view.LaCrosse.login", return_value=True), patch(
|
||||
"lacrosse_view.LaCrosse.get_locations",
|
||||
return_value=[Location(id=1, name="Test")],
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"username": new_username,
|
||||
"password": new_password,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == FlowResultType.ABORT
|
||||
assert result2["reason"] == "reauth_successful"
|
||||
|
||||
assert len(hass.config_entries.async_entries()) == 1
|
||||
assert hass.config_entries.async_entries()[0].data == {
|
||||
"username": new_username,
|
||||
"password": new_password,
|
||||
"id": "1",
|
||||
"name": "Test",
|
||||
}
|
||||
|
@ -47,11 +47,14 @@ async def test_login_error(hass: HomeAssistant) -> None:
|
||||
assert not await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data[DOMAIN]
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state == ConfigEntryState.SETUP_RETRY
|
||||
assert entries[0].state == ConfigEntryState.SETUP_ERROR
|
||||
flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
|
||||
assert flows
|
||||
assert len(flows) == 1
|
||||
assert flows[0]["context"]["source"] == "reauth"
|
||||
|
||||
|
||||
async def test_http_error(hass: HomeAssistant) -> None:
|
||||
@ -65,7 +68,6 @@ async def test_http_error(hass: HomeAssistant) -> None:
|
||||
assert not await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.data[DOMAIN]
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
@ -100,3 +102,40 @@ async def test_new_token(hass: HomeAssistant) -> None:
|
||||
await hass.async_block_till_done()
|
||||
|
||||
login.assert_called_once()
|
||||
|
||||
|
||||
async def test_failed_token(hass: HomeAssistant) -> None:
|
||||
"""Test if a reauth flow occurs when token refresh fails."""
|
||||
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRY_DATA)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch("lacrosse_view.LaCrosse.login", return_value=True) as login, patch(
|
||||
"lacrosse_view.LaCrosse.get_sensors",
|
||||
return_value=[TEST_SENSOR],
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
login.assert_called_once()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state == ConfigEntryState.LOADED
|
||||
|
||||
one_hour_after = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
with patch(
|
||||
"lacrosse_view.LaCrosse.login", side_effect=LoginError("Test")
|
||||
), freeze_time(one_hour_after):
|
||||
async_fire_time_changed(hass, one_hour_after)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state == ConfigEntryState.LOADED
|
||||
|
||||
flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN)
|
||||
assert flows
|
||||
assert len(flows) == 1
|
||||
assert flows[0]["context"]["source"] == "reauth"
|
||||
|
@ -5,7 +5,12 @@ from homeassistant.components.lacrosse_view import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import MOCK_ENTRY_DATA, TEST_NO_PERMISSION_SENSOR, TEST_SENSOR
|
||||
from . import (
|
||||
MOCK_ENTRY_DATA,
|
||||
TEST_NO_PERMISSION_SENSOR,
|
||||
TEST_SENSOR,
|
||||
TEST_UNSUPPORTED_SENSOR,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
@ -26,7 +31,7 @@ async def test_entities_added(hass: HomeAssistant) -> None:
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state == ConfigEntryState.LOADED
|
||||
assert hass.states.get("sensor.test_test_temperature")
|
||||
assert hass.states.get("sensor.test_temperature")
|
||||
|
||||
|
||||
async def test_sensor_permission(hass: HomeAssistant, caplog) -> None:
|
||||
@ -36,6 +41,25 @@ async def test_sensor_permission(hass: HomeAssistant, caplog) -> None:
|
||||
|
||||
with patch("lacrosse_view.LaCrosse.login", return_value=True), patch(
|
||||
"lacrosse_view.LaCrosse.get_sensors", return_value=[TEST_NO_PERMISSION_SENSOR]
|
||||
):
|
||||
assert not await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entries = hass.config_entries.async_entries(DOMAIN)
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state == ConfigEntryState.SETUP_ERROR
|
||||
assert not hass.states.get("sensor.test_temperature")
|
||||
assert "This account does not have permission to read Test" in caplog.text
|
||||
|
||||
|
||||
async def test_field_not_supported(hass: HomeAssistant, caplog) -> None:
|
||||
"""Test if it raises a warning when the field is not supported."""
|
||||
config_entry = MockConfigEntry(domain=DOMAIN, data=MOCK_ENTRY_DATA)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch("lacrosse_view.LaCrosse.login", return_value=True), patch(
|
||||
"lacrosse_view.LaCrosse.get_sensors", return_value=[TEST_UNSUPPORTED_SENSOR]
|
||||
):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
@ -45,5 +69,5 @@ async def test_sensor_permission(hass: HomeAssistant, caplog) -> None:
|
||||
assert entries
|
||||
assert len(entries) == 1
|
||||
assert entries[0].state == ConfigEntryState.LOADED
|
||||
assert hass.states.get("sensor.test_test_temperature") is None
|
||||
assert "No permission to read sensor" in caplog.text
|
||||
assert hass.states.get("sensor.test_some_unsupported_field") is None
|
||||
assert "Unsupported sensor field" in caplog.text
|
||||
|
Loading…
x
Reference in New Issue
Block a user