mirror of
https://github.com/home-assistant/core.git
synced 2025-07-24 05:37:44 +00:00
Co2signal configflow (#53193)
Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Daniel Hjelseth Høyer
GitHub
parent
562aa74c77
commit
f0b28c90bf
@ -152,7 +152,6 @@ omit =
|
||||
homeassistant/components/clicksend/notify.py
|
||||
homeassistant/components/clicksend_tts/notify.py
|
||||
homeassistant/components/cmus/media_player.py
|
||||
homeassistant/components/co2signal/*
|
||||
homeassistant/components/coinbase/sensor.py
|
||||
homeassistant/components/comed_hourly_pricing/sensor.py
|
||||
homeassistant/components/comfoconnect/fan.py
|
||||
|
||||
@ -1 +1,20 @@
|
||||
"""The co2signal component."""
|
||||
"""The CO2 Signal integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import DOMAIN # noqa: F401
|
||||
|
||||
PLATFORMS = ["sensor"]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Set up CO2 Signal from a config entry."""
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
249
homeassistant/components/co2signal/config_flow.py
Normal file
249
homeassistant/components/co2signal/config_flow.py
Normal file
@ -0,0 +1,249 @@
|
||||
"""Config flow for Co2signal integration."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import CO2Signal
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_TOKEN
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from .const import CONF_COUNTRY_CODE, DOMAIN
|
||||
from .util import get_extra_name
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
TYPE_USE_HOME = "Use home location"
|
||||
TYPE_SPECIFY_COORDINATES = "Specify coordinates"
|
||||
TYPE_SPECIFY_COUNTRY = "Specify country code"
|
||||
|
||||
|
||||
def _get_entry_type(config: dict) -> str:
|
||||
"""Get entry type from the configuration."""
|
||||
if CONF_LATITUDE in config:
|
||||
return TYPE_SPECIFY_COORDINATES
|
||||
|
||||
if CONF_COUNTRY_CODE in config:
|
||||
return TYPE_SPECIFY_COUNTRY
|
||||
|
||||
return TYPE_USE_HOME
|
||||
|
||||
|
||||
def _validate_info(hass, config: dict) -> dict:
|
||||
"""Validate the passed in info."""
|
||||
if CONF_COUNTRY_CODE in config:
|
||||
latitude = None
|
||||
longitude = None
|
||||
else:
|
||||
latitude = config.get(CONF_LATITUDE, hass.config.latitude)
|
||||
longitude = config.get(CONF_LONGITUDE, hass.config.longitude)
|
||||
|
||||
try:
|
||||
data = CO2Signal.get_latest(
|
||||
config[CONF_API_KEY],
|
||||
config.get(CONF_COUNTRY_CODE),
|
||||
latitude,
|
||||
longitude,
|
||||
wait=False,
|
||||
)
|
||||
|
||||
except ValueError as err:
|
||||
err_str = str(err)
|
||||
|
||||
if "Invalid authentication credentials" in err_str:
|
||||
raise InvalidAuth from err
|
||||
if "API rate limit exceeded." in err_str:
|
||||
raise APIRatelimitExceeded from err
|
||||
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
raise UnknownError from err
|
||||
except Exception as err: # pylint: disable=broad-except
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
raise UnknownError from err
|
||||
|
||||
else:
|
||||
if data.get("status") != "ok":
|
||||
_LOGGER.exception("Unexpected response: %s", data)
|
||||
raise UnknownError
|
||||
|
||||
return data
|
||||
|
||||
|
||||
class CO2Error(HomeAssistantError):
|
||||
"""Base error."""
|
||||
|
||||
|
||||
class InvalidAuth(CO2Error):
|
||||
"""Raised when invalid authentication credentials are provided."""
|
||||
|
||||
|
||||
class APIRatelimitExceeded(CO2Error):
|
||||
"""Raised when the API rate limit is exceeded."""
|
||||
|
||||
|
||||
class UnknownError(CO2Error):
|
||||
"""Raised when an unknown error occurs."""
|
||||
|
||||
|
||||
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Co2signal."""
|
||||
|
||||
VERSION = 1
|
||||
_data: dict | None
|
||||
|
||||
async def async_step_import(self, import_info):
|
||||
"""Set the config entry up from yaml."""
|
||||
data = {CONF_API_KEY: import_info[CONF_TOKEN]}
|
||||
|
||||
if CONF_COUNTRY_CODE in import_info:
|
||||
data[CONF_COUNTRY_CODE] = import_info[CONF_COUNTRY_CODE]
|
||||
new_entry_type = TYPE_SPECIFY_COUNTRY
|
||||
elif (
|
||||
CONF_LATITUDE in import_info
|
||||
and import_info[CONF_LATITUDE] != self.hass.config.latitude
|
||||
and import_info[CONF_LONGITUDE] != self.hass.config.longitude
|
||||
):
|
||||
data[CONF_LATITUDE] = import_info[CONF_LATITUDE]
|
||||
data[CONF_LONGITUDE] = import_info[CONF_LONGITUDE]
|
||||
new_entry_type = TYPE_SPECIFY_COORDINATES
|
||||
else:
|
||||
new_entry_type = TYPE_USE_HOME
|
||||
|
||||
for entry in self._async_current_entries(include_ignore=True):
|
||||
if entry.source == config_entries.SOURCE_IGNORE:
|
||||
continue
|
||||
|
||||
if (cur_entry_type := _get_entry_type(entry.data)) != new_entry_type:
|
||||
continue
|
||||
|
||||
if cur_entry_type == TYPE_USE_HOME and new_entry_type == TYPE_USE_HOME:
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
if (
|
||||
cur_entry_type == TYPE_SPECIFY_COUNTRY
|
||||
and data[CONF_COUNTRY_CODE] == entry.data[CONF_COUNTRY_CODE]
|
||||
):
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
if (
|
||||
cur_entry_type == TYPE_SPECIFY_COORDINATES
|
||||
and data[CONF_LATITUDE] == entry.data[CONF_LATITUDE]
|
||||
and data[CONF_LONGITUDE] == entry.data[CONF_LONGITUDE]
|
||||
):
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
try:
|
||||
await self.hass.async_add_executor_job(_validate_info, self.hass, data)
|
||||
except CO2Error:
|
||||
return self.async_abort(reason="unknown")
|
||||
|
||||
return self.async_create_entry(
|
||||
title=get_extra_name(self.hass, data) or "CO2 Signal", data=data
|
||||
)
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required("location", default=TYPE_USE_HOME): vol.In(
|
||||
(
|
||||
TYPE_USE_HOME,
|
||||
TYPE_SPECIFY_COORDINATES,
|
||||
TYPE_SPECIFY_COUNTRY,
|
||||
)
|
||||
),
|
||||
vol.Required(CONF_API_KEY): cv.string,
|
||||
}
|
||||
)
|
||||
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=data_schema,
|
||||
)
|
||||
|
||||
data = {CONF_API_KEY: user_input[CONF_API_KEY]}
|
||||
|
||||
if user_input["location"] == TYPE_SPECIFY_COORDINATES:
|
||||
self._data = data
|
||||
return await self.async_step_coordinates()
|
||||
|
||||
if user_input["location"] == TYPE_SPECIFY_COUNTRY:
|
||||
self._data = data
|
||||
return await self.async_step_country()
|
||||
|
||||
return await self._validate_and_create("user", data_schema, data)
|
||||
|
||||
async def async_step_coordinates(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Validate coordinates."""
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required(
|
||||
CONF_LATITUDE,
|
||||
): cv.latitude,
|
||||
vol.Required(
|
||||
CONF_LONGITUDE,
|
||||
): cv.longitude,
|
||||
}
|
||||
)
|
||||
if user_input is None:
|
||||
return self.async_show_form(step_id="coordinates", data_schema=data_schema)
|
||||
|
||||
assert self._data is not None
|
||||
|
||||
return await self._validate_and_create(
|
||||
"coordinates", data_schema, {**self._data, **user_input}
|
||||
)
|
||||
|
||||
async def async_step_country(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Validate country."""
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_COUNTRY_CODE): cv.string,
|
||||
}
|
||||
)
|
||||
if user_input is None:
|
||||
return self.async_show_form(step_id="country", data_schema=data_schema)
|
||||
|
||||
assert self._data is not None
|
||||
|
||||
return await self._validate_and_create(
|
||||
"country", data_schema, {**self._data, **user_input}
|
||||
)
|
||||
|
||||
async def _validate_and_create(
|
||||
self, step_id: str, data_schema: vol.Schema, data: dict
|
||||
) -> FlowResult:
|
||||
"""Validate data and show form if it is invalid."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
try:
|
||||
await self.hass.async_add_executor_job(_validate_info, self.hass, data)
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except APIRatelimitExceeded:
|
||||
errors["base"] = "api_ratelimit"
|
||||
except UnknownError:
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
return self.async_create_entry(
|
||||
title=get_extra_name(self.hass, data) or "CO2 Signal",
|
||||
data=data,
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id=step_id,
|
||||
data_schema=data_schema,
|
||||
errors=errors,
|
||||
)
|
||||
11
homeassistant/components/co2signal/const.py
Normal file
11
homeassistant/components/co2signal/const.py
Normal file
@ -0,0 +1,11 @@
|
||||
"""Constants for the Co2signal integration."""
|
||||
|
||||
|
||||
DOMAIN = "co2signal"
|
||||
CONF_COUNTRY_CODE = "country_code"
|
||||
ATTRIBUTION = "Data provided by CO2signal"
|
||||
MSG_LOCATION = (
|
||||
"Please use either coordinates or the country code. "
|
||||
"For the coordinates, "
|
||||
"you need to use both latitude and longitude."
|
||||
)
|
||||
@ -2,7 +2,10 @@
|
||||
"domain": "co2signal",
|
||||
"name": "CO2 Signal",
|
||||
"documentation": "https://www.home-assistant.io/integrations/co2signal",
|
||||
"requirements": ["co2signal==0.4.2"],
|
||||
"requirements": [
|
||||
"co2signal==0.4.2"
|
||||
],
|
||||
"codeowners": [],
|
||||
"iot_class": "cloud_polling"
|
||||
}
|
||||
"iot_class": "cloud_polling",
|
||||
"config_flow": true
|
||||
}
|
||||
@ -5,9 +5,14 @@ import logging
|
||||
import CO2Signal
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
||||
from homeassistant.const import (
|
||||
ATTR_ATTRIBUTION,
|
||||
ATTR_IDENTIFIERS,
|
||||
ATTR_MANUFACTURER,
|
||||
ATTR_NAME,
|
||||
CONF_API_KEY,
|
||||
CONF_LATITUDE,
|
||||
CONF_LONGITUDE,
|
||||
CONF_TOKEN,
|
||||
@ -15,18 +20,12 @@ from homeassistant.const import (
|
||||
)
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
CONF_COUNTRY_CODE = "country_code"
|
||||
from .const import ATTRIBUTION, CONF_COUNTRY_CODE, DOMAIN, MSG_LOCATION
|
||||
from .util import get_extra_name
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
SCAN_INTERVAL = timedelta(minutes=3)
|
||||
|
||||
ATTRIBUTION = "Data provided by CO2signal"
|
||||
|
||||
MSG_LOCATION = (
|
||||
"Please use either coordinates or the country code. "
|
||||
"For the coordinates, "
|
||||
"you need to use both latitude and longitude."
|
||||
)
|
||||
CO2_INTENSITY_UNIT = f"CO2eq/{ENERGY_KILO_WATT_HOUR}"
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
@ -38,16 +37,31 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
async def async_setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
"""Set up the CO2signal sensor."""
|
||||
token = config[CONF_TOKEN]
|
||||
lat = config.get(CONF_LATITUDE, hass.config.latitude)
|
||||
lon = config.get(CONF_LONGITUDE, hass.config.longitude)
|
||||
country_code = config.get(CONF_COUNTRY_CODE)
|
||||
await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_IMPORT},
|
||||
data=config,
|
||||
)
|
||||
|
||||
_LOGGER.debug("Setting up the sensor using the %s", country_code)
|
||||
|
||||
add_entities([CO2Sensor(token, country_code, lat, lon)], True)
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
"""Set up the CO2signal sensor."""
|
||||
name = "CO2 intensity"
|
||||
if extra_name := get_extra_name(hass, entry.data):
|
||||
name += f" - {extra_name}"
|
||||
|
||||
async_add_entities(
|
||||
[
|
||||
CO2Sensor(
|
||||
name,
|
||||
entry.data,
|
||||
entry_id=entry.entry_id,
|
||||
)
|
||||
],
|
||||
True,
|
||||
)
|
||||
|
||||
|
||||
class CO2Sensor(SensorEntity):
|
||||
@ -56,33 +70,37 @@ class CO2Sensor(SensorEntity):
|
||||
_attr_icon = "mdi:molecule-co2"
|
||||
_attr_unit_of_measurement = CO2_INTENSITY_UNIT
|
||||
|
||||
def __init__(self, token, country_code, lat, lon):
|
||||
def __init__(self, name, config, entry_id):
|
||||
"""Initialize the sensor."""
|
||||
self._token = token
|
||||
self._country_code = country_code
|
||||
self._latitude = lat
|
||||
self._longitude = lon
|
||||
|
||||
if country_code is not None:
|
||||
device_name = country_code
|
||||
else:
|
||||
device_name = f"{round(self._latitude, 2)}/{round(self._longitude, 2)}"
|
||||
|
||||
self._attr_name = f"CO2 intensity - {device_name}"
|
||||
self._config = config
|
||||
self._attr_name = name
|
||||
self._attr_extra_state_attributes = {ATTR_ATTRIBUTION: ATTRIBUTION}
|
||||
self._attr_device_info = {
|
||||
ATTR_IDENTIFIERS: {(DOMAIN, entry_id)},
|
||||
ATTR_NAME: "CO2 signal",
|
||||
ATTR_MANUFACTURER: "Tmrow.com",
|
||||
"entry_type": "service",
|
||||
}
|
||||
self._attr_unique_id = f"{entry_id}_co2intensity"
|
||||
|
||||
def update(self):
|
||||
"""Get the latest data and updates the states."""
|
||||
|
||||
_LOGGER.debug("Update data for %s", self.name)
|
||||
|
||||
if self._country_code is not None:
|
||||
data = CO2Signal.get_latest_carbon_intensity(
|
||||
self._token, country_code=self._country_code
|
||||
)
|
||||
if CONF_COUNTRY_CODE in self._config:
|
||||
kwargs = {"country_code": self._config[CONF_COUNTRY_CODE]}
|
||||
elif CONF_LATITUDE in self._config:
|
||||
kwargs = {
|
||||
"latitude": self._config[CONF_LATITUDE],
|
||||
"longitude": self._config[CONF_LONGITUDE],
|
||||
}
|
||||
else:
|
||||
data = CO2Signal.get_latest_carbon_intensity(
|
||||
self._token, latitude=self._latitude, longitude=self._longitude
|
||||
)
|
||||
kwargs = {
|
||||
"latitude": self.hass.config.latitude,
|
||||
"longitude": self.hass.config.longitude,
|
||||
}
|
||||
|
||||
self._attr_state = round(data, 2)
|
||||
self._attr_state = round(
|
||||
CO2Signal.get_latest_carbon_intensity(self._config[CONF_API_KEY], **kwargs),
|
||||
2,
|
||||
)
|
||||
|
||||
34
homeassistant/components/co2signal/strings.json
Normal file
34
homeassistant/components/co2signal/strings.json
Normal file
@ -0,0 +1,34 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"location": "Get data for",
|
||||
"api_key": "[%key:common::config_flow::data::access_token%]"
|
||||
},
|
||||
"description": "Visit https://co2signal.com/ to request a token."
|
||||
},
|
||||
"coordinates": {
|
||||
"data": {
|
||||
"latitude": "[%key:common::config_flow::data::latitude%]",
|
||||
"longitude": "[%key:common::config_flow::data::longitude%]"
|
||||
}
|
||||
},
|
||||
"country": {
|
||||
"data": {
|
||||
"country_code": "Country code"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]",
|
||||
"api_ratelimit": "API Ratelimit exceeded"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]",
|
||||
"api_ratelimit": "API Ratelimit exceeded"
|
||||
}
|
||||
}
|
||||
}
|
||||
34
homeassistant/components/co2signal/translations/en.json
Normal file
34
homeassistant/components/co2signal/translations/en.json
Normal file
@ -0,0 +1,34 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Device is already configured",
|
||||
"api_ratelimit": "API Ratelimit exceeded",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"error": {
|
||||
"api_ratelimit": "API Ratelimit exceeded",
|
||||
"invalid_auth": "Invalid authentication",
|
||||
"unknown": "Unexpected error"
|
||||
},
|
||||
"step": {
|
||||
"coordinates": {
|
||||
"data": {
|
||||
"latitude": "Latitude",
|
||||
"longitude": "Longitude"
|
||||
}
|
||||
},
|
||||
"country": {
|
||||
"data": {
|
||||
"country_code": "Country code"
|
||||
}
|
||||
},
|
||||
"user": {
|
||||
"data": {
|
||||
"api_key": "Access Token",
|
||||
"location": "Get data for"
|
||||
},
|
||||
"description": "Visit https://co2signal.com/ to request a token."
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
18
homeassistant/components/co2signal/util.py
Normal file
18
homeassistant/components/co2signal/util.py
Normal file
@ -0,0 +1,18 @@
|
||||
"""Utils for CO2 signal."""
|
||||
from __future__ import annotations
|
||||
|
||||
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import CONF_COUNTRY_CODE
|
||||
|
||||
|
||||
def get_extra_name(hass: HomeAssistant, config: dict) -> str | None:
|
||||
"""Return the extra name describing the location if not home."""
|
||||
if CONF_COUNTRY_CODE in config:
|
||||
return config[CONF_COUNTRY_CODE]
|
||||
|
||||
if CONF_LATITUDE in config:
|
||||
return f"{round(config[CONF_LATITUDE], 2)}, {round(config[CONF_LONGITUDE], 2)}"
|
||||
|
||||
return None
|
||||
@ -45,6 +45,7 @@ FLOWS = [
|
||||
"cert_expiry",
|
||||
"climacell",
|
||||
"cloudflare",
|
||||
"co2signal",
|
||||
"coinbase",
|
||||
"control4",
|
||||
"coolmaster",
|
||||
|
||||
@ -249,6 +249,9 @@ buienradar==1.0.4
|
||||
# homeassistant.components.caldav
|
||||
caldav==0.7.1
|
||||
|
||||
# homeassistant.components.co2signal
|
||||
co2signal==0.4.2
|
||||
|
||||
# homeassistant.components.coinbase
|
||||
coinbase==2.1.0
|
||||
|
||||
|
||||
11
tests/components/co2signal/__init__.py
Normal file
11
tests/components/co2signal/__init__.py
Normal file
@ -0,0 +1,11 @@
|
||||
"""Tests for the CO2 Signal integration."""
|
||||
|
||||
VALID_PAYLOAD = {
|
||||
"status": "ok",
|
||||
"countryCode": "FR",
|
||||
"data": {
|
||||
"carbonIntensity": 45.98623190095805,
|
||||
"fossilFuelPercentage": 5.461182741937103,
|
||||
},
|
||||
"units": {"carbonIntensity": "gCO2eq/kWh"},
|
||||
}
|
||||
299
tests/components/co2signal/test_config_flow.py
Normal file
299
tests/components/co2signal/test_config_flow.py
Normal file
@ -0,0 +1,299 @@
|
||||
"""Test the CO2 Signal config flow."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries, setup
|
||||
from homeassistant.components.co2signal import DOMAIN, config_flow
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import RESULT_TYPE_CREATE_ENTRY, RESULT_TYPE_FORM
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from . import VALID_PAYLOAD
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_form_home(hass: HomeAssistant) -> None:
|
||||
"""Test we get the form."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] is None
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
), patch(
|
||||
"homeassistant.components.co2signal.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"location": config_flow.TYPE_USE_HOME,
|
||||
"api_key": "api_key",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result2["title"] == "CO2 Signal"
|
||||
assert result2["data"] == {
|
||||
"api_key": "api_key",
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_coordinates(hass: HomeAssistant) -> None:
|
||||
"""Test we get the form."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] is None
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"location": config_flow.TYPE_SPECIFY_COORDINATES,
|
||||
"api_key": "api_key",
|
||||
},
|
||||
)
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
), patch(
|
||||
"homeassistant.components.co2signal.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result2["flow_id"],
|
||||
{
|
||||
"latitude": 12.3,
|
||||
"longitude": 45.6,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result3["title"] == "12.3, 45.6"
|
||||
assert result3["data"] == {
|
||||
"latitude": 12.3,
|
||||
"longitude": 45.6,
|
||||
"api_key": "api_key",
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_form_country(hass: HomeAssistant) -> None:
|
||||
"""Test we get the form."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == RESULT_TYPE_FORM
|
||||
assert result["errors"] is None
|
||||
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"location": config_flow.TYPE_SPECIFY_COUNTRY,
|
||||
"api_key": "api_key",
|
||||
},
|
||||
)
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
), patch(
|
||||
"homeassistant.components.co2signal.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry:
|
||||
result3 = await hass.config_entries.flow.async_configure(
|
||||
result2["flow_id"],
|
||||
{
|
||||
"country_code": "fr",
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result3["type"] == RESULT_TYPE_CREATE_ENTRY
|
||||
assert result3["title"] == "fr"
|
||||
assert result3["data"] == {
|
||||
"country_code": "fr",
|
||||
"api_key": "api_key",
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"err_str,err_code",
|
||||
[
|
||||
("Invalid authentication credentials", "invalid_auth"),
|
||||
("API rate limit exceeded.", "api_ratelimit"),
|
||||
("Something else", "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_form_error_handling(hass: HomeAssistant, err_str, err_code) -> None:
|
||||
"""Test we handle expected errors."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
side_effect=ValueError(err_str),
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"location": config_flow.TYPE_USE_HOME,
|
||||
"api_key": "api_key",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": err_code}
|
||||
|
||||
|
||||
async def test_form_error_unexpected_error(hass: HomeAssistant) -> None:
|
||||
"""Test we handle unexpected error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
side_effect=Exception("Boom"),
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"location": config_flow.TYPE_USE_HOME,
|
||||
"api_key": "api_key",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_form_error_unexpected_data(hass: HomeAssistant) -> None:
|
||||
"""Test we handle unexpected data."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value={"status": "error"},
|
||||
):
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
"location": config_flow.TYPE_USE_HOME,
|
||||
"api_key": "api_key",
|
||||
},
|
||||
)
|
||||
|
||||
assert result2["type"] == RESULT_TYPE_FORM
|
||||
assert result2["errors"] == {"base": "unknown"}
|
||||
|
||||
|
||||
async def test_import(hass: HomeAssistant) -> None:
|
||||
"""Test we import correctly."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass, "sensor", {"sensor": {"platform": "co2signal", "token": "1234"}}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries("co2signal")) == 1
|
||||
state = hass.states.get("sensor.co2_intensity")
|
||||
assert state is not None
|
||||
assert state.state == "45.99"
|
||||
assert state.name == "CO2 intensity"
|
||||
|
||||
|
||||
async def test_import_abort_existing_home(hass: HomeAssistant) -> None:
|
||||
"""Test we abort if home entry found."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
MockConfigEntry(domain="co2signal", data={"api_key": "abcd"}).add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass, "sensor", {"sensor": {"platform": "co2signal", "token": "1234"}}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries("co2signal")) == 1
|
||||
|
||||
|
||||
async def test_import_abort_existing_country(hass: HomeAssistant) -> None:
|
||||
"""Test we abort if existing country found."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
MockConfigEntry(
|
||||
domain="co2signal", data={"api_key": "abcd", "country_code": "nl"}
|
||||
).add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{
|
||||
"sensor": {
|
||||
"platform": "co2signal",
|
||||
"token": "1234",
|
||||
"country_code": "nl",
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries("co2signal")) == 1
|
||||
|
||||
|
||||
async def test_import_abort_existing_coordinates(hass: HomeAssistant) -> None:
|
||||
"""Test we abort if existing coordinates found."""
|
||||
await setup.async_setup_component(hass, "persistent_notification", {})
|
||||
MockConfigEntry(
|
||||
domain="co2signal", data={"api_key": "abcd", "latitude": 1, "longitude": 2}
|
||||
).add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.co2signal.config_flow.CO2Signal.get_latest",
|
||||
return_value=VALID_PAYLOAD,
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"sensor",
|
||||
{
|
||||
"sensor": {
|
||||
"platform": "co2signal",
|
||||
"token": "1234",
|
||||
"latitude": 1,
|
||||
"longitude": 2,
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.config_entries.async_entries("co2signal")) == 1
|
||||
Loading…
x
Reference in New Issue
Block a user