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

Refactor WAQI tests (#148968)

Browse Source
This commit is contained in:
Joost Lekkerkerker 2025-07-17 20:30:40 +02:00 committed by GitHub
parent 9802441fea
commit 0d819f2389
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 735 additions and 358 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
homeassistant/components/waqi/config_flow.py
View File

@ -66,24 +66,22 @@ class WAQIConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
async with WAQIClient(
session=async_get_clientsession(self.hass)
) as waqi_client:
waqi_client.authenticate(user_input[CONF_API_KEY])
try:
await waqi_client.get_by_ip()
except WAQIAuthenticationError:
errors["base"] = "invalid_auth"
except WAQIConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self.data = user_input
if user_input[CONF_METHOD] == CONF_MAP:
return await self.async_step_map()
return await self.async_step_station_number()
client = WAQIClient(session=async_get_clientsession(self.hass))
client.authenticate(user_input[CONF_API_KEY])
try:
await client.get_by_ip()
except WAQIAuthenticationError:
errors["base"] = "invalid_auth"
except WAQIConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self.data = user_input
if user_input[CONF_METHOD] == CONF_MAP:
return await self.async_step_map()
return await self.async_step_station_number()
return self.async_show_form(
step_id="user",
@ -107,22 +105,20 @@ class WAQIConfigFlow(ConfigFlow, domain=DOMAIN):
"""Add measuring station via map."""
errors: dict[str, str] = {}
if user_input is not None:
async with WAQIClient(
session=async_get_clientsession(self.hass)
) as waqi_client:
waqi_client.authenticate(self.data[CONF_API_KEY])
try:
measuring_station = await waqi_client.get_by_coordinates(
user_input[CONF_LOCATION][CONF_LATITUDE],
user_input[CONF_LOCATION][CONF_LONGITUDE],
)
except WAQIConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return await self._async_create_entry(measuring_station)
client = WAQIClient(session=async_get_clientsession(self.hass))
client.authenticate(self.data[CONF_API_KEY])
try:
measuring_station = await client.get_by_coordinates(
user_input[CONF_LOCATION][CONF_LATITUDE],
user_input[CONF_LOCATION][CONF_LONGITUDE],
)
except WAQIConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return await self._async_create_entry(measuring_station)
return self.async_show_form(
step_id=CONF_MAP,
data_schema=self.add_suggested_values_to_schema(
@ -149,21 +145,19 @@ class WAQIConfigFlow(ConfigFlow, domain=DOMAIN):
"""Add measuring station via station number."""
errors: dict[str, str] = {}
if user_input is not None:
async with WAQIClient(
session=async_get_clientsession(self.hass)
) as waqi_client:
waqi_client.authenticate(self.data[CONF_API_KEY])
station_number = user_input[CONF_STATION_NUMBER]
measuring_station, errors = await get_by_station_number(
waqi_client, abs(station_number)
client = WAQIClient(session=async_get_clientsession(self.hass))
client.authenticate(self.data[CONF_API_KEY])
station_number = user_input[CONF_STATION_NUMBER]
measuring_station, errors = await get_by_station_number(
client, abs(station_number)
)
if not measuring_station:
measuring_station, _ = await get_by_station_number(
client,
abs(station_number) - station_number - station_number,
)
if not measuring_station:
measuring_station, _ = await get_by_station_number(
waqi_client,
abs(station_number) - station_number - station_number,
)
if measuring_station:
return await self._async_create_entry(measuring_station)
if measuring_station:
return await self._async_create_entry(measuring_station)
return self.async_show_form(
step_id=CONF_STATION_NUMBER,
data_schema=vol.Schema(

12
tests/components/waqi/__init__.py
View File

@ -1 +1,13 @@
"""Tests for the World Air Quality Index (WAQI) integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Fixture for setting up the component."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

31
tests/components/waqi/conftest.py
View File

@ -1,14 +1,16 @@
"""Common fixtures for the World Air Quality Index (WAQI) tests."""
from collections.abc import Generator
from collections.abc import AsyncGenerator, Generator
from unittest.mock import AsyncMock, patch
from aiowaqi import WAQIAirQuality
import pytest
from homeassistant.components.waqi.const import CONF_STATION_NUMBER, DOMAIN
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_load_json_object_fixture
@pytest.fixture
@ -29,3 +31,28 @@ def mock_config_entry() -> MockConfigEntry:
title="de Jongweg, Utrecht",
data={CONF_API_KEY: "asd", CONF_STATION_NUMBER: 4584},
)
@pytest.fixture
async def mock_waqi(hass: HomeAssistant) -> AsyncGenerator[AsyncMock]:
"""Mock WAQI client."""
with (
patch(
"homeassistant.components.waqi.WAQIClient",
autospec=True,
) as mock_waqi,
patch(
"homeassistant.components.waqi.config_flow.WAQIClient",
new=mock_waqi,
),
):
client = mock_waqi.return_value
air_quality = WAQIAirQuality.from_dict(
await async_load_json_object_fixture(
hass, "air_quality_sensor.json", DOMAIN
)
)
client.get_by_station_number.return_value = air_quality
client.get_by_ip.return_value = air_quality
client.get_by_coordinates.return_value = air_quality
yield client

666
tests/components/waqi/snapshots/test_sensor.ambr
View File

@ -1,5 +1,42 @@
# serializer version: 1
# name: test_sensor
# name: test_sensor[sensor.de_jongweg_utrecht_air_quality_index-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_air_quality_index',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.AQI: 'aqi'>,
'original_icon': None,
'original_name': 'Air quality index',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '4584_air_quality',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_air_quality_index-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
@ -15,39 +52,104 @@
'state': '29',
})
# ---
# name: test_sensor.1
# name: test_sensor[sensor.de_jongweg_utrecht_carbon_monoxide-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_carbon_monoxide',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Carbon monoxide',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'carbon_monoxide',
'unique_id': '4584_carbon_monoxide',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_carbon_monoxide-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'device_class': 'humidity',
'friendly_name': 'de Jongweg, Utrecht Humidity',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_humidity',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80',
})
# ---
# name: test_sensor.10
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Visibility using nephelometry',
'friendly_name': 'de Jongweg, Utrecht Carbon monoxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_visibility_using_nephelometry',
'entity_id': 'sensor.de_jongweg_utrecht_carbon_monoxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80',
'state': '2.3',
})
# ---
# name: test_sensor.11
# name: test_sensor[sensor.de_jongweg_utrecht_dominant_pollutant-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
'co',
'no2',
'o3',
'so2',
'pm10',
'pm25',
'neph',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_dominant_pollutant',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
'original_icon': None,
'original_name': 'Dominant pollutant',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'dominant_pollutant',
'unique_id': '4584_dominant_pollutant',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_dominant_pollutant-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
@ -71,7 +173,309 @@
'state': 'o3',
})
# ---
# name: test_sensor.2
# name: test_sensor[sensor.de_jongweg_utrecht_humidity-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_humidity',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.HUMIDITY: 'humidity'>,
'original_icon': None,
'original_name': 'Humidity',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '4584_humidity',
'unit_of_measurement': '%',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_humidity-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'device_class': 'humidity',
'friendly_name': 'de Jongweg, Utrecht Humidity',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_humidity',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_nitrogen_dioxide-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_nitrogen_dioxide',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Nitrogen dioxide',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'nitrogen_dioxide',
'unique_id': '4584_nitrogen_dioxide',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_nitrogen_dioxide-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Nitrogen dioxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_nitrogen_dioxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.3',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_ozone-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_ozone',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Ozone',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'ozone',
'unique_id': '4584_ozone',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_ozone-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Ozone',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_ozone',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '29.4',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_pm10-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_pm10',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'PM10',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'pm10',
'unique_id': '4584_pm10',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_pm10-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht PM10',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_pm10',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '12',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_pm2_5-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_pm2_5',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'PM2.5',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'pm25',
'unique_id': '4584_pm25',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_pm2_5-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht PM2.5',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_pm2_5',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '17',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_pressure-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_pressure',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.PRESSURE: 'pressure'>,
'original_icon': None,
'original_name': 'Pressure',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '4584_pressure',
'unit_of_measurement': <UnitOfPressure.HPA: 'hPa'>,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_pressure-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
@ -88,7 +492,99 @@
'state': '1008.8',
})
# ---
# name: test_sensor.3
# name: test_sensor[sensor.de_jongweg_utrecht_sulphur_dioxide-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_sulphur_dioxide',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sulphur dioxide',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sulphur_dioxide',
'unique_id': '4584_sulphur_dioxide',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_sulphur_dioxide-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Sulphur dioxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_sulphur_dioxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.3',
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '4584_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
@ -105,93 +601,55 @@
'state': '16',
})
# ---
# name: test_sensor.4
# name: test_sensor[sensor.de_jongweg_utrecht_visibility_using_nephelometry-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.de_jongweg_utrecht_visibility_using_nephelometry',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Visibility using nephelometry',
'platform': 'waqi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'neph',
'unique_id': '4584_neph',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[sensor.de_jongweg_utrecht_visibility_using_nephelometry-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Carbon monoxide',
'friendly_name': 'de Jongweg, Utrecht Visibility using nephelometry',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_carbon_monoxide',
'entity_id': 'sensor.de_jongweg_utrecht_visibility_using_nephelometry',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.3',
})
# ---
# name: test_sensor.5
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Nitrogen dioxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_nitrogen_dioxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.3',
})
# ---
# name: test_sensor.6
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Ozone',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_ozone',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '29.4',
})
# ---
# name: test_sensor.7
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht Sulphur dioxide',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_sulphur_dioxide',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2.3',
})
# ---
# name: test_sensor.8
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht PM10',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_pm10',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '12',
})
# ---
# name: test_sensor.9
StateSnapshot({
'attributes': ReadOnlyDict({
'attribution': 'RIVM - Rijksinstituut voor Volksgezondheid en Milieum, Landelijk Meetnet Luchtkwaliteit and World Air Quality Index Project',
'friendly_name': 'de Jongweg, Utrecht PM2.5',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.de_jongweg_utrecht_pm2_5',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '17',
'state': '80',
})
# ---

222
tests/components/waqi/test_config_flow.py
View File

@ -1,15 +1,14 @@
"""Test the World Air Quality Index (WAQI) config flow."""
import json
from typing import Any
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
from aiowaqi import WAQIAirQuality, WAQIAuthenticationError, WAQIConnectionError
from aiowaqi import WAQIAuthenticationError, WAQIConnectionError
import pytest
from homeassistant import config_entries
from homeassistant.components.waqi.config_flow import CONF_MAP
from homeassistant.components.waqi.const import CONF_STATION_NUMBER, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import (
CONF_API_KEY,
CONF_LATITUDE,
@ -20,10 +19,6 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import async_load_fixture
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
@pytest.mark.parametrize(
("method", "payload"),
@ -45,63 +40,28 @@ pytestmark = pytest.mark.usefixtures("mock_setup_entry")
async def test_full_map_flow(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_waqi: AsyncMock,
method: str,
payload: dict[str, Any],
) -> None:
"""Test we get the form."""
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
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_ip",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: method},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: method},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == method
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_coordinates",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
patch(
"aiowaqi.WAQIClient.get_by_station_number",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
payload,
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
payload,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "de Jongweg, Utrecht"
@ -109,6 +69,7 @@ async def test_full_map_flow(
CONF_API_KEY: "asd",
CONF_STATION_NUMBER: 4584,
}
assert result["result"].unique_id == "4584"
assert len(mock_setup_entry.mock_calls) == 1
@ -121,73 +82,43 @@ async def test_full_map_flow(
],
)
async def test_flow_errors(
hass: HomeAssistant, exception: Exception, error: str
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_waqi: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test we handle errors during configuration."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_ip",
side_effect=exception,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: CONF_MAP},
)
await hass.async_block_till_done()
mock_waqi.get_by_ip.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: CONF_MAP},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_ip",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: CONF_MAP},
)
await hass.async_block_till_done()
mock_waqi.get_by_ip.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: CONF_MAP},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "map"
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_coordinates",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LOCATION: {CONF_LATITUDE: 50.0, CONF_LONGITUDE: 10.0},
},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LOCATION: {CONF_LATITUDE: 50.0, CONF_LONGITUDE: 10.0},
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
@ -232,6 +163,7 @@ async def test_flow_errors(
async def test_error_in_second_step(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_waqi: AsyncMock,
method: str,
payload: dict[str, Any],
exception: Exception,
@ -239,74 +171,36 @@ async def test_error_in_second_step(
) -> None:
"""Test we get the form."""
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
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_ip",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: method},
)
await hass.async_block_till_done()
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_API_KEY: "asd", CONF_METHOD: method},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == method
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch("aiowaqi.WAQIClient.get_by_coordinates", side_effect=exception),
patch("aiowaqi.WAQIClient.get_by_station_number", side_effect=exception),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
payload,
)
await hass.async_block_till_done()
mock_waqi.get_by_coordinates.side_effect = exception
mock_waqi.get_by_station_number.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
payload,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
with (
patch(
"aiowaqi.WAQIClient.authenticate",
),
patch(
"aiowaqi.WAQIClient.get_by_coordinates",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
patch(
"aiowaqi.WAQIClient.get_by_station_number",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
payload,
)
await hass.async_block_till_done()
mock_waqi.get_by_coordinates.side_effect = None
mock_waqi.get_by_station_number.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
payload,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "de Jongweg, Utrecht"

24
tests/components/waqi/test_init.py Normal file
View File

@ -0,0 +1,24 @@
"""Test the World Air Quality Index (WAQI) initialization."""
from unittest.mock import AsyncMock
from aiowaqi import WAQIError
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_setup_failed(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_waqi: AsyncMock,
) -> None:
"""Test setup failure due to API error."""
mock_waqi.get_by_station_number.side_effect = WAQIError("API error")
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY

48
tests/components/waqi/test_sensor.py
View File

@ -1,59 +1,27 @@
"""Test the World Air Quality Index (WAQI) sensor."""
import json
from unittest.mock import patch
from unittest.mock import AsyncMock
from aiowaqi import WAQIAirQuality, WAQIError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.waqi.const import DOMAIN
from homeassistant.components.waqi.sensor import SENSORS
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, async_load_fixture
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensor(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_waqi: AsyncMock,
mock_config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
) -> None:
"""Test failed update."""
mock_config_entry.add_to_hass(hass)
with patch(
"aiowaqi.WAQIClient.get_by_station_number",
return_value=WAQIAirQuality.from_dict(
json.loads(
await async_load_fixture(hass, "air_quality_sensor.json", DOMAIN)
)
),
):
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
for sensor in SENSORS:
entity_id = entity_registry.async_get_entity_id(
SENSOR_DOMAIN, DOMAIN, f"4584_{sensor.key}"
)
assert hass.states.get(entity_id) == snapshot
"""Test the World Air Quality Index (WAQI) sensor."""
await setup_integration(hass, mock_config_entry)
async def test_updating_failed(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test failed update."""
mock_config_entry.add_to_hass(hass)
with patch(
"aiowaqi.WAQIClient.get_by_station_number",
side_effect=WAQIError(),
):
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)