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

Upgrade thethingsnetwork to v3 (#113375) 

* thethingsnetwork upgrade to v3

* add en translations and requirements_all

* fix most of the findings

* hassfest

* use ttn_client v0.0.3

* reduce content of initial release

* remove features that trigger errors

* remove unneeded

* add initial testcases

* Exception warning

* add strict type checking

* add strict type checking

* full coverage

* rename to conftest

* review changes

* avoid using private attributes - use protected instead

* simplify config_flow

* remove unused options

* review changes

* upgrade client

* add types client library - no need to cast

* use add_suggested_values_to_schema

* add ruff fix

* review changes

* remove unneeded comment

* use typevar for TTN value

* use typevar for TTN value

* review

* ruff error not detected in local

* test review

* re-order fixture

* fix test

* reviews

* fix case

* split testcases

* review feedback

* Update homeassistant/components/thethingsnetwork/__init__.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Update homeassistant/components/thethingsnetwork/__init__.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Update tests/components/thethingsnetwork/test_config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Remove deprecated var

* Update tests/components/thethingsnetwork/test_config_flow.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Remove unused import

* fix ruff warning

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
pull/118177/head
Angel Nunez Mencias 2024-05-26 16:30:33 +02:00 committed by GitHub
parent a7938091bf
commit b85cf36a68
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 725 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -1422,7 +1422,6 @@ omit =
homeassistant/components/tensorflow/image_processing.py
homeassistant/components/tfiac/climate.py
homeassistant/components/thermoworks_smoke/sensor.py
homeassistant/components/thethingsnetwork/*
homeassistant/components/thingspeak/*
homeassistant/components/thinkingcleaner/*
homeassistant/components/thomson/device_tracker.py

1
.strict-typing
View File

@ -428,6 +428,7 @@ homeassistant.components.tcp.*
homeassistant.components.technove.*
homeassistant.components.tedee.*
homeassistant.components.text.*
homeassistant.components.thethingsnetwork.*
homeassistant.components.threshold.*
homeassistant.components.tibber.*
homeassistant.components.tile.*

3
CODEOWNERS
View File

@ -1421,7 +1421,8 @@ build.json @home-assistant/supervisor
/tests/components/thermobeacon/ @bdraco
/homeassistant/components/thermopro/ @bdraco @h3ss
/tests/components/thermopro/ @bdraco @h3ss
/homeassistant/components/thethingsnetwork/ @fabaff
/homeassistant/components/thethingsnetwork/ @angelnu
/tests/components/thethingsnetwork/ @angelnu
/homeassistant/components/thread/ @home-assistant/core
/tests/components/thread/ @home-assistant/core
/homeassistant/components/tibber/ @danielhiversen

76
homeassistant/components/thethingsnetwork/__init__.py
View File

@ -1,29 +1,28 @@
"""Support for The Things network."""
import logging
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
CONF_ACCESS_KEY = "access_key"
CONF_APP_ID = "app_id"
from .const import CONF_APP_ID, DOMAIN, PLATFORMS, TTN_API_HOST
from .coordinator import TTNCoordinator
DATA_TTN = "data_thethingsnetwork"
DOMAIN = "thethingsnetwork"
TTN_ACCESS_KEY = "ttn_access_key"
TTN_APP_ID = "ttn_app_id"
TTN_DATA_STORAGE_URL = (
"https://{app_id}.data.thethingsnetwork.org/{endpoint}/{device_id}"
)
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = vol.Schema(
{
# Configuration via yaml not longer supported - keeping to warn about migration
DOMAIN: vol.Schema(
{
vol.Required(CONF_APP_ID): cv.string,
vol.Required(CONF_ACCESS_KEY): cv.string,
vol.Required("access_key"): cv.string,
}
)
},
@ -33,10 +32,57 @@ CONFIG_SCHEMA = vol.Schema(
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Initialize of The Things Network component."""
conf = config[DOMAIN]
app_id = conf.get(CONF_APP_ID)
access_key = conf.get(CONF_ACCESS_KEY)
hass.data[DATA_TTN] = {TTN_ACCESS_KEY: access_key, TTN_APP_ID: app_id}
if DOMAIN in config:
ir.async_create_issue(
hass,
DOMAIN,
"manual_migration",
breaks_in_ha_version="2024.12.0",
is_fixable=False,
severity=ir.IssueSeverity.ERROR,
translation_key="manual_migration",
translation_placeholders={
"domain": DOMAIN,
"v2_v3_migration_url": "https://www.thethingsnetwork.org/forum/c/v2-to-v3-upgrade/102",
"v2_deprecation_url": "https://www.thethingsnetwork.org/forum/t/the-things-network-v2-is-permanently-shutting-down-completed/50710",
},
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Establish connection with The Things Network."""
_LOGGER.debug(
"Set up %s at %s",
entry.data[CONF_API_KEY],
entry.data.get(CONF_HOST, TTN_API_HOST),
)
coordinator = TTNCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
_LOGGER.debug(
"Remove %s at %s",
entry.data[CONF_API_KEY],
entry.data.get(CONF_HOST, TTN_API_HOST),
)
# Unload entities created for each supported platform
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
del hass.data[DOMAIN][entry.entry_id]
return True

108
homeassistant/components/thethingsnetwork/config_flow.py Normal file
View File

@ -0,0 +1,108 @@
"""The Things Network's integration config flow."""
from collections.abc import Mapping
import logging
from typing import Any
from ttn_client import TTNAuthError, TTNClient
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import CONF_APP_ID, DOMAIN, TTN_API_HOST
_LOGGER = logging.getLogger(__name__)
class TTNFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 1
_reauth_entry: ConfigEntry | None = None
async def async_step_user(
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
"""User initiated config flow."""
errors = {}
if user_input is not None:
client = TTNClient(
user_input[CONF_HOST],
user_input[CONF_APP_ID],
user_input[CONF_API_KEY],
0,
)
try:
await client.fetch_data()
except TTNAuthError:
_LOGGER.exception("Error authenticating with The Things Network")
errors["base"] = "invalid_auth"
except Exception:
_LOGGER.exception("Unknown error occurred")
errors["base"] = "unknown"
if not errors:
# Create entry
if self._reauth_entry:
return self.async_update_reload_and_abort(
self._reauth_entry,
data=user_input,
reason="reauth_successful",
)
await self.async_set_unique_id(user_input[CONF_APP_ID])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=str(user_input[CONF_APP_ID]),
data=user_input,
)
# Show form for user to provide settings
if not user_input:
if self._reauth_entry:
user_input = self._reauth_entry.data
else:
user_input = {CONF_HOST: TTN_API_HOST}
schema = self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_APP_ID): str,
vol.Required(CONF_API_KEY): TextSelector(
TextSelectorConfig(
type=TextSelectorType.PASSWORD, autocomplete="api_key"
)
),
}
),
user_input,
)
return self.async_show_form(step_id="user", data_schema=schema, errors=errors)
async def async_step_reauth(
self, user_input: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle a flow initialized by a reauth event."""
self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that informs the user that reauth is required."""
if user_input is None:
return self.async_show_form(step_id="reauth_confirm")
return await self.async_step_user()

12
homeassistant/components/thethingsnetwork/const.py Normal file
View File

@ -0,0 +1,12 @@
"""The Things Network's integration constants."""
from homeassistant.const import Platform
DOMAIN = "thethingsnetwork"
TTN_API_HOST = "eu1.cloud.thethings.network"
PLATFORMS = [Platform.SENSOR]
CONF_APP_ID = "app_id"
POLLING_PERIOD_S = 60

66
homeassistant/components/thethingsnetwork/coordinator.py Normal file
View File

@ -0,0 +1,66 @@
"""The Things Network's integration DataUpdateCoordinator."""
from datetime import timedelta
import logging
from ttn_client import TTNAuthError, TTNClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import CONF_APP_ID, POLLING_PERIOD_S
_LOGGER = logging.getLogger(__name__)
class TTNCoordinator(DataUpdateCoordinator[TTNClient.DATA_TYPE]):
"""TTN coordinator."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize my coordinator."""
super().__init__(
hass,
_LOGGER,
# Name of the data. For logging purposes.
name=f"TheThingsNetwork_{entry.data[CONF_APP_ID]}",
# Polling interval. Will only be polled if there are subscribers.
update_interval=timedelta(
seconds=POLLING_PERIOD_S,
),
)
self._client = TTNClient(
entry.data[CONF_HOST],
entry.data[CONF_APP_ID],
entry.data[CONF_API_KEY],
push_callback=self._push_callback,
)
async def _async_update_data(self) -> TTNClient.DATA_TYPE:
"""Fetch data from API endpoint.
This is the place to pre-process the data to lookup tables
so entities can quickly look up their data.
"""
try:
# Note: asyncio.TimeoutError and aiohttp.ClientError are already
# handled by the data update coordinator.
measurements = await self._client.fetch_data()
except TTNAuthError as err:
# Raising ConfigEntryAuthFailed will cancel future updates
# and start a config flow with SOURCE_REAUTH (async_step_reauth)
_LOGGER.error("TTNAuthError")
raise ConfigEntryAuthFailed from err
else:
# Return measurements
_LOGGER.debug("fetched data: %s", measurements)
return measurements
async def _push_callback(self, data: TTNClient.DATA_TYPE) -> None:
_LOGGER.debug("pushed data: %s", data)
# Push data to entities
self.async_set_updated_data(data)

71
homeassistant/components/thethingsnetwork/entity.py Normal file
View File

@ -0,0 +1,71 @@
"""Support for The Things Network entities."""
import logging
from ttn_client import TTNBaseValue
from homeassistant.core import callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import TTNCoordinator
_LOGGER = logging.getLogger(__name__)
class TTNEntity(CoordinatorEntity[TTNCoordinator]):
"""Representation of a The Things Network Data Storage sensor."""
_attr_has_entity_name = True
_ttn_value: TTNBaseValue
def __init__(
self,
coordinator: TTNCoordinator,
app_id: str,
ttn_value: TTNBaseValue,
) -> None:
"""Initialize a The Things Network Data Storage sensor."""
# Pass coordinator to CoordinatorEntity
super().__init__(coordinator)
self._ttn_value = ttn_value
self._attr_unique_id = f"{self.device_id}_{self.field_id}"
self._attr_name = self.field_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{app_id}_{self.device_id}")},
name=self.device_id,
)
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
my_entity_update = self.coordinator.data.get(self.device_id, {}).get(
self.field_id
)
if (
my_entity_update
and my_entity_update.received_at > self._ttn_value.received_at
):
_LOGGER.debug(
"Received update for %s: %s", self.unique_id, my_entity_update
)
# Check that the type of an entity has not changed since the creation
assert isinstance(my_entity_update, type(self._ttn_value))
self._ttn_value = my_entity_update
self.async_write_ha_state()
@property
def device_id(self) -> str:
"""Return device_id."""
return str(self._ttn_value.device_id)
@property
def field_id(self) -> str:
"""Return field_id."""
return str(self._ttn_value.field_id)

7
homeassistant/components/thethingsnetwork/manifest.json
View File

@ -1,7 +1,10 @@
{
"domain": "thethingsnetwork",
"name": "The Things Network",
"codeowners": ["@fabaff"],
"codeowners": ["@angelnu"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/thethingsnetwork",
"iot_class": "local_push"
"integration_type": "hub",
"iot_class": "cloud_polling",
"requirements": ["ttn_client==0.0.4"]
}

179
homeassistant/components/thethingsnetwork/sensor.py
View File

@ -1,165 +1,56 @@
"""Support for The Things Network's Data storage integration."""
"""The Things Network's integration sensors."""
from __future__ import annotations
import asyncio
from http import HTTPStatus
import logging
import aiohttp
from aiohttp.hdrs import ACCEPT, AUTHORIZATION
import voluptuous as vol
from ttn_client import TTNSensorValue
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.const import (
ATTR_DEVICE_ID,
ATTR_TIME,
CONF_DEVICE_ID,
CONTENT_TYPE_JSON,
)
from homeassistant.components.sensor import SensorEntity, StateType
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DATA_TTN, TTN_ACCESS_KEY, TTN_APP_ID, TTN_DATA_STORAGE_URL
from .const import CONF_APP_ID, DOMAIN
from .entity import TTNEntity
_LOGGER = logging.getLogger(__name__)
ATTR_RAW = "raw"
DEFAULT_TIMEOUT = 10
CONF_VALUES = "values"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_DEVICE_ID): cv.string,
vol.Required(CONF_VALUES): {cv.string: cv.string},
}
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up The Things Network Data storage sensors."""
ttn = hass.data[DATA_TTN]
device_id = config[CONF_DEVICE_ID]
values = config[CONF_VALUES]
app_id = ttn.get(TTN_APP_ID)
access_key = ttn.get(TTN_ACCESS_KEY)
"""Add entities for TTN."""
ttn_data_storage = TtnDataStorage(hass, app_id, device_id, access_key, values)
success = await ttn_data_storage.async_update()
coordinator = hass.data[DOMAIN][entry.entry_id]
if not success:
return
sensors: set[tuple[str, str]] = set()
devices = []
for value, unit_of_measurement in values.items():
devices.append(
TtnDataSensor(ttn_data_storage, device_id, value, unit_of_measurement)
)
async_add_entities(devices, True)
def _async_measurement_listener() -> None:
data = coordinator.data
new_sensors = {
(device_id, field_id): TtnDataSensor(
coordinator,
entry.data[CONF_APP_ID],
ttn_value,
)
for device_id, device_uplinks in data.items()
for field_id, ttn_value in device_uplinks.items()
if (device_id, field_id) not in sensors
and isinstance(ttn_value, TTNSensorValue)
}
if len(new_sensors):
async_add_entities(new_sensors.values())
sensors.update(new_sensors.keys())
entry.async_on_unload(coordinator.async_add_listener(_async_measurement_listener))
_async_measurement_listener()
class TtnDataSensor(SensorEntity):
"""Representation of a The Things Network Data Storage sensor."""
class TtnDataSensor(TTNEntity, SensorEntity):
"""Represents a TTN Home Assistant Sensor."""
def __init__(self, ttn_data_storage, device_id, value, unit_of_measurement):
"""Initialize a The Things Network Data Storage sensor."""
self._ttn_data_storage = ttn_data_storage
self._state = None
self._device_id = device_id
self._unit_of_measurement = unit_of_measurement
self._value = value
self._name = f"{self._device_id} {self._value}"
_ttn_value: TTNSensorValue
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def native_value(self):
def native_value(self) -> StateType:
"""Return the state of the entity."""
if self._ttn_data_storage.data is not None:
try:
return self._state[self._value]
except KeyError:
return None
return None
@property
def native_unit_of_measurement(self):
"""Return the unit this state is expressed in."""
return self._unit_of_measurement
@property
def extra_state_attributes(self):
"""Return the state attributes of the sensor."""
if self._ttn_data_storage.data is not None:
return {
ATTR_DEVICE_ID: self._device_id,
ATTR_RAW: self._state["raw"],
ATTR_TIME: self._state["time"],
}
async def async_update(self) -> None:
"""Get the current state."""
await self._ttn_data_storage.async_update()
self._state = self._ttn_data_storage.data
class TtnDataStorage:
"""Get the latest data from The Things Network Data Storage."""
def __init__(self, hass, app_id, device_id, access_key, values):
"""Initialize the data object."""
self.data = None
self._hass = hass
self._app_id = app_id
self._device_id = device_id
self._values = values
self._url = TTN_DATA_STORAGE_URL.format(
app_id=app_id, endpoint="api/v2/query", device_id=device_id
)
self._headers = {ACCEPT: CONTENT_TYPE_JSON, AUTHORIZATION: f"key {access_key}"}
async def async_update(self):
"""Get the current state from The Things Network Data Storage."""
try:
session = async_get_clientsession(self._hass)
async with asyncio.timeout(DEFAULT_TIMEOUT):
response = await session.get(self._url, headers=self._headers)
except (TimeoutError, aiohttp.ClientError):
_LOGGER.error("Error while accessing: %s", self._url)
return None
status = response.status
if status == HTTPStatus.NO_CONTENT:
_LOGGER.error("The device is not available: %s", self._device_id)
return None
if status == HTTPStatus.UNAUTHORIZED:
_LOGGER.error("Not authorized for Application ID: %s", self._app_id)
return None
if status == HTTPStatus.NOT_FOUND:
_LOGGER.error("Application ID is not available: %s", self._app_id)
return None
data = await response.json()
self.data = data[-1]
for value in self._values.items():
if value[0] not in self.data:
_LOGGER.warning("Value not available: %s", value[0])
return response
return self._ttn_value.value

32
homeassistant/components/thethingsnetwork/strings.json Normal file
View File

@ -0,0 +1,32 @@
{
"config": {
"step": {
"user": {
"title": "Connect to The Things Network v3 App",
"description": "Enter the API hostname, app id and API key for your TTN application.\n\nYou can find your API key in the [The Things Network console](https://console.thethingsnetwork.org) -> Applications -> application_id -> API keys.",
"data": {
"hostname": "[%key:common::config_flow::data::host%]",
"app_id": "Application ID",
"access_key": "[%key:common::config_flow::data::api_key%]"
}
},
"reauth_confirm": {
"description": "The Things Network application could not be connected.\n\nPlease check your credentials."
}
},
"abort": {
"already_configured": "Application ID is already configured",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
},
"issues": {
"manual_migration": {
"description": "Configuring {domain} using YAML was removed as part of migrating to [The Things Network v3]({v2_v3_migration_url}). [The Things Network v2 has shutted down]({v2_deprecation_url}).\n\nPlease remove the {domain} entry from the configuration.yaml and add re-add the integration using the config_flow",
"title": "The {domain} YAML configuration is not supported"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -551,6 +551,7 @@ FLOWS = {
"tessie",
"thermobeacon",
"thermopro",
"thethingsnetwork",
"thread",
"tibber",
"tile",

4
homeassistant/generated/integrations.json
View File

@ -6146,8 +6146,8 @@
"thethingsnetwork": {
"name": "The Things Network",
"integration_type": "hub",
"config_flow": false,
"iot_class": "local_push"
"config_flow": true,
"iot_class": "cloud_polling"
},
"thingspeak": {
"name": "ThingSpeak",

10
mypy.ini
View File

@ -4044,6 +4044,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.thethingsnetwork.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.threshold.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt
View File

@ -2763,6 +2763,9 @@ transmission-rpc==7.0.3
# homeassistant.components.twinkly
ttls==1.5.1
# homeassistant.components.thethingsnetwork
ttn_client==0.0.4
# homeassistant.components.tuya
tuya-device-sharing-sdk==0.1.9

3
requirements_test_all.txt
View File

@ -2137,6 +2137,9 @@ transmission-rpc==7.0.3
# homeassistant.components.twinkly
ttls==1.5.1
# homeassistant.components.thethingsnetwork
ttn_client==0.0.4
# homeassistant.components.tuya
tuya-device-sharing-sdk==0.1.9

10
tests/components/thethingsnetwork/__init__.py Normal file
View File

@ -0,0 +1,10 @@
"""Define tests for the The Things Network."""
from homeassistant.core import HomeAssistant
async def init_integration(hass: HomeAssistant, config_entry) -> None:
"""Mock TTNClient."""
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

95
tests/components/thethingsnetwork/conftest.py Normal file
View File

@ -0,0 +1,95 @@
"""Define fixtures for the The Things Network tests."""
from unittest.mock import AsyncMock, patch
import pytest
from ttn_client import TTNSensorValue
from homeassistant.components.thethingsnetwork.const import (
CONF_APP_ID,
DOMAIN,
TTN_API_HOST,
)
from homeassistant.const import CONF_API_KEY, CONF_HOST
from tests.common import MockConfigEntry
HOST = "example.com"
APP_ID = "my_app"
API_KEY = "my_api_key"
DEVICE_ID = "my_device"
DEVICE_ID_2 = "my_device_2"
DEVICE_FIELD = "a_field"
DEVICE_FIELD_2 = "a_field_2"
DEVICE_FIELD_VALUE = 42
DATA = {
DEVICE_ID: {
DEVICE_FIELD: TTNSensorValue(
{
"end_device_ids": {"device_id": DEVICE_ID},
"received_at": "2024-03-11T08:49:11.153738893Z",
},
DEVICE_FIELD,
DEVICE_FIELD_VALUE,
)
}
}
DATA_UPDATE = {
DEVICE_ID: {
DEVICE_FIELD: TTNSensorValue(
{
"end_device_ids": {"device_id": DEVICE_ID},
"received_at": "2024-03-12T08:49:11.153738893Z",
},
DEVICE_FIELD,
DEVICE_FIELD_VALUE,
)
},
DEVICE_ID_2: {
DEVICE_FIELD_2: TTNSensorValue(
{
"end_device_ids": {"device_id": DEVICE_ID_2},
"received_at": "2024-03-12T08:49:11.153738893Z",
},
DEVICE_FIELD_2,
DEVICE_FIELD_VALUE,
)
},
}
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Mock a config entry."""
return MockConfigEntry(
domain=DOMAIN,
unique_id=APP_ID,
title=APP_ID,
data={
CONF_APP_ID: APP_ID,
CONF_HOST: TTN_API_HOST,
CONF_API_KEY: API_KEY,
},
)
@pytest.fixture
def mock_ttnclient():
"""Mock TTNClient."""
with (
patch(
"homeassistant.components.thethingsnetwork.coordinator.TTNClient",
autospec=True,
) as ttn_client,
patch(
"homeassistant.components.thethingsnetwork.config_flow.TTNClient",
new=ttn_client,
),
):
instance = ttn_client.return_value
instance.fetch_data = AsyncMock(return_value=DATA)
yield ttn_client

132
tests/components/thethingsnetwork/test_config_flow.py Normal file
View File

@ -0,0 +1,132 @@
"""Define tests for the The Things Network onfig flows."""
import pytest
from ttn_client import TTNAuthError
from homeassistant.components.thethingsnetwork.const import CONF_APP_ID, DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import CONF_API_KEY, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import init_integration
from .conftest import API_KEY, APP_ID, HOST
USER_DATA = {CONF_HOST: HOST, CONF_APP_ID: APP_ID, CONF_API_KEY: API_KEY}
async def test_user(hass: HomeAssistant, mock_ttnclient) -> None:
"""Test user config."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=USER_DATA,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == APP_ID
assert result["data"][CONF_HOST] == HOST
assert result["data"][CONF_APP_ID] == APP_ID
assert result["data"][CONF_API_KEY] == API_KEY
@pytest.mark.parametrize(
("fetch_data_exception", "base_error"),
[(TTNAuthError, "invalid_auth"), (Exception, "unknown")],
)
async def test_user_errors(
hass: HomeAssistant, fetch_data_exception, base_error, mock_ttnclient
) -> None:
"""Test user config errors."""
# Test error
mock_ttnclient.return_value.fetch_data.side_effect = fetch_data_exception
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=USER_DATA,
)
assert result["type"] is FlowResultType.FORM
assert base_error in result["errors"]["base"]
# Recover
mock_ttnclient.return_value.fetch_data.side_effect = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=USER_DATA,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_duplicate_entry(
hass: HomeAssistant, mock_ttnclient, mock_config_entry
) -> None:
"""Test that duplicate entries are caught."""
await init_integration(hass, mock_config_entry)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data=USER_DATA,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_step_reauth(
hass: HomeAssistant, mock_ttnclient, mock_config_entry
) -> None:
"""Test that the reauth step works."""
await init_integration(hass, mock_config_entry)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": APP_ID,
"entry_id": mock_config_entry.entry_id,
},
data=USER_DATA,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert not result["errors"]
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert not result["errors"]
new_api_key = "1234"
new_user_input = dict(USER_DATA)
new_user_input[CONF_API_KEY] = new_api_key
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input=new_user_input
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert len(hass.config_entries.async_entries()) == 1
assert hass.config_entries.async_entries()[0].data[CONF_API_KEY] == new_api_key

33
tests/components/thethingsnetwork/test_init.py Normal file
View File

@ -0,0 +1,33 @@
"""Define tests for the The Things Network init."""
import pytest
from ttn_client import TTNAuthError
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from homeassistant.setup import async_setup_component
from .conftest import DOMAIN
async def test_error_configuration(
hass: HomeAssistant,
issue_registry: ir.IssueRegistry,
) -> None:
"""Test issue is logged when deprecated configuration is used."""
await async_setup_component(
hass, DOMAIN, {DOMAIN: {"app_id": "123", "access_key": "42"}}
)
await hass.async_block_till_done()
assert issue_registry.async_get_issue(DOMAIN, "manual_migration")
@pytest.mark.parametrize(("exception_class"), [TTNAuthError, Exception])
async def test_init_exceptions(
hass: HomeAssistant, mock_ttnclient, exception_class, mock_config_entry
) -> None:
"""Test TTN Exceptions."""
mock_ttnclient.return_value.fetch_data.side_effect = exception_class
mock_config_entry.add_to_hass(hass)
assert not await hass.config_entries.async_setup(mock_config_entry.entry_id)

43
tests/components/thethingsnetwork/test_sensor.py Normal file
View File

@ -0,0 +1,43 @@
"""Define tests for the The Things Network sensor."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import init_integration
from .conftest import (
APP_ID,
DATA_UPDATE,
DEVICE_FIELD,
DEVICE_FIELD_2,
DEVICE_ID,
DEVICE_ID_2,
DOMAIN,
)
async def test_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_ttnclient,
mock_config_entry,
) -> None:
"""Test a working configurations."""
await init_integration(hass, mock_config_entry)
# Check devices
assert (
device_registry.async_get_device(
identifiers={(DOMAIN, f"{APP_ID}_{DEVICE_ID}")}
).name
== DEVICE_ID
)
# Check entities
assert entity_registry.async_get(f"sensor.{DEVICE_ID}_{DEVICE_FIELD}")
assert not entity_registry.async_get(f"sensor.{DEVICE_ID_2}_{DEVICE_FIELD}")
push_callback = mock_ttnclient.call_args.kwargs["push_callback"]
await push_callback(DATA_UPDATE)
assert entity_registry.async_get(f"sensor.{DEVICE_ID_2}_{DEVICE_FIELD_2}")