github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

GeoNet NZ Volcanic Alert Level sensor (#26901) 

* first version of new integration

* moved icon to shared consts

* added unit tests

* transformed from geolocation to sensor integration

* alert level is now the state of the sensor

* adopted unit tests

* fixed comment

* keep all sensors registered even if the feed update fails intermittently

* bumped version of upstream library

* bumped version of integration library

* regenerated requirements

* bumped version of integration library

* bumped version of integration library

* fixed generated file

* removed commented out code

* regenerated config flow file

* update to latest integration library version

* simplified code

* removed debug log statement

* simplified code structure

* defined constant

* use core interfaces

* moved test and fixture

* sorted imports

* simplified patching

* moved fixture to central config file
This commit is contained in:
Malte Franken 2019-11-20 00:00:59 +11:00 committed by Martin Hjelmare
parent 8439e597ce
commit 4e9e9efa43
16 changed files with 844 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CODEOWNERS
View File

@ -112,6 +112,7 @@ homeassistant/components/gearbest/* @HerrHofrat
homeassistant/components/geniushub/* @zxdavb
homeassistant/components/geo_rss_events/* @exxamalte
homeassistant/components/geonetnz_quakes/* @exxamalte
homeassistant/components/geonetnz_volcano/* @exxamalte
homeassistant/components/gitter/* @fabaff
homeassistant/components/glances/* @fabaff @engrbm87
homeassistant/components/gntp/* @robbiet480

16
homeassistant/components/geonetnz_volcano/.translations/en.json Normal file
View File

@ -0,0 +1,16 @@
{
"config": {
"error": {
"identifier_exists": "Location already registered"
},
"step": {
"user": {
"data": {
"radius": "Radius"
},
"title": "Fill in your filter details."
}
},
"title": "GeoNet NZ Volcano"
}
}

205
homeassistant/components/geonetnz_volcano/__init__.py Normal file
View File

@ -0,0 +1,205 @@
"""The GeoNet NZ Volcano integration."""
import asyncio
import logging
from datetime import timedelta, datetime
from typing import Optional
import voluptuous as vol
from aio_geojson_geonetnz_volcano import GeonetnzVolcanoFeedManager
from homeassistant.core import callback
from homeassistant.util.unit_system import METRIC_SYSTEM
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_RADIUS,
CONF_SCAN_INTERVAL,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM,
LENGTH_MILES,
)
from homeassistant.helpers import config_validation as cv, aiohttp_client
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from .config_flow import configured_instances
from .const import (
DEFAULT_RADIUS,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
FEED,
SIGNAL_NEW_SENSOR,
SIGNAL_UPDATE_ENTITY,
)
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_RADIUS, default=DEFAULT_RADIUS): vol.Coerce(float),
vol.Optional(
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
): cv.time_period,
}
)
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass, config):
"""Set up the GeoNet NZ Volcano component."""
if DOMAIN not in config:
return True
conf = config[DOMAIN]
latitude = conf.get(CONF_LATITUDE, hass.config.latitude)
longitude = conf.get(CONF_LONGITUDE, hass.config.longitude)
scan_interval = conf[CONF_SCAN_INTERVAL]
identifier = f"{latitude}, {longitude}"
if identifier in configured_instances(hass):
return True
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_LATITUDE: latitude,
CONF_LONGITUDE: longitude,
CONF_RADIUS: conf[CONF_RADIUS],
CONF_SCAN_INTERVAL: scan_interval,
},
)
)
return True
async def async_setup_entry(hass, config_entry):
"""Set up the GeoNet NZ Volcano component as config entry."""
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN].setdefault(FEED, {})
radius = config_entry.data[CONF_RADIUS]
unit_system = config_entry.data[CONF_UNIT_SYSTEM]
if unit_system == CONF_UNIT_SYSTEM_IMPERIAL:
radius = METRIC_SYSTEM.length(radius, LENGTH_MILES)
# Create feed entity manager for all platforms.
manager = GeonetnzVolcanoFeedEntityManager(hass, config_entry, radius, unit_system)
hass.data[DOMAIN][FEED][config_entry.entry_id] = manager
_LOGGER.debug("Feed entity manager added for %s", config_entry.entry_id)
await manager.async_init()
return True
async def async_unload_entry(hass, config_entry):
"""Unload an GeoNet NZ Volcano component config entry."""
manager = hass.data[DOMAIN][FEED].pop(config_entry.entry_id)
await manager.async_stop()
await asyncio.wait(
[hass.config_entries.async_forward_entry_unload(config_entry, "sensor")]
)
return True
class GeonetnzVolcanoFeedEntityManager:
"""Feed Entity Manager for GeoNet NZ Volcano feed."""
def __init__(self, hass, config_entry, radius_in_km, unit_system):
"""Initialize the Feed Entity Manager."""
self._hass = hass
self._config_entry = config_entry
coordinates = (
config_entry.data[CONF_LATITUDE],
config_entry.data[CONF_LONGITUDE],
)
websession = aiohttp_client.async_get_clientsession(hass)
self._feed_manager = GeonetnzVolcanoFeedManager(
websession,
self._generate_entity,
self._update_entity,
self._remove_entity,
coordinates,
filter_radius=radius_in_km,
)
self._config_entry_id = config_entry.entry_id
self._scan_interval = timedelta(seconds=config_entry.data[CONF_SCAN_INTERVAL])
self._unit_system = unit_system
self._track_time_remove_callback = None
self.listeners = []
async def async_init(self):
"""Schedule initial and regular updates based on configured time interval."""
self._hass.async_create_task(
self._hass.config_entries.async_forward_entry_setup(
self._config_entry, "sensor"
)
)
async def update(event_time):
"""Update."""
await self.async_update()
# Trigger updates at regular intervals.
self._track_time_remove_callback = async_track_time_interval(
self._hass, update, self._scan_interval
)
_LOGGER.debug("Feed entity manager initialized")
async def async_update(self):
"""Refresh data."""
await self._feed_manager.update()
_LOGGER.debug("Feed entity manager updated")
async def async_stop(self):
"""Stop this feed entity manager from refreshing."""
for unsub_dispatcher in self.listeners:
unsub_dispatcher()
self.listeners = []
if self._track_time_remove_callback:
self._track_time_remove_callback()
_LOGGER.debug("Feed entity manager stopped")
@callback
def async_event_new_entity(self):
"""Return manager specific event to signal new entity."""
return SIGNAL_NEW_SENSOR.format(self._config_entry_id)
def get_entry(self, external_id):
"""Get feed entry by external id."""
return self._feed_manager.feed_entries.get(external_id)
def last_update(self) -> Optional[datetime]:
"""Return the last update of this feed."""
return self._feed_manager.last_update
def last_update_successful(self) -> Optional[datetime]:
"""Return the last successful update of this feed."""
return self._feed_manager.last_update_successful
async def _generate_entity(self, external_id):
"""Generate new entity."""
async_dispatcher_send(
self._hass,
self.async_event_new_entity(),
self,
external_id,
self._unit_system,
)
async def _update_entity(self, external_id):
"""Update entity."""
async_dispatcher_send(self._hass, SIGNAL_UPDATE_ENTITY.format(external_id))
async def _remove_entity(self, external_id):
"""Ignore removing entity."""

74
homeassistant/components/geonetnz_volcano/config_flow.py Normal file
View File

@ -0,0 +1,74 @@
"""Config flow to configure the GeoNet NZ Volcano integration."""
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_RADIUS,
CONF_SCAN_INTERVAL,
CONF_UNIT_SYSTEM,
CONF_UNIT_SYSTEM_IMPERIAL,
CONF_UNIT_SYSTEM_METRIC,
)
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv
from .const import DEFAULT_RADIUS, DEFAULT_SCAN_INTERVAL, DOMAIN
_LOGGER = logging.getLogger(__name__)
@callback
def configured_instances(hass):
"""Return a set of configured GeoNet NZ Volcano instances."""
return set(
f"{entry.data[CONF_LATITUDE]}, {entry.data[CONF_LONGITUDE]}"
for entry in hass.config_entries.async_entries(DOMAIN)
)
class GeonetnzVolcanoFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a GeoNet NZ Volcano config flow."""
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
async def _show_form(self, errors=None):
"""Show the form to the user."""
data_schema = vol.Schema(
{vol.Optional(CONF_RADIUS, default=DEFAULT_RADIUS): cv.positive_int}
)
return self.async_show_form(
step_id="user", data_schema=data_schema, errors=errors or {}
)
async def async_step_import(self, import_config):
"""Import a config entry from configuration.yaml."""
return await self.async_step_user(import_config)
async def async_step_user(self, user_input=None):
"""Handle the start of the config flow."""
if not user_input:
return await self._show_form()
latitude = user_input.get(CONF_LATITUDE, self.hass.config.latitude)
user_input[CONF_LATITUDE] = latitude
longitude = user_input.get(CONF_LONGITUDE, self.hass.config.longitude)
user_input[CONF_LONGITUDE] = longitude
identifier = f"{user_input[CONF_LATITUDE]}, {user_input[CONF_LONGITUDE]}"
if identifier in configured_instances(self.hass):
return await self._show_form({"base": "identifier_exists"})
if self.hass.config.units.name == CONF_UNIT_SYSTEM_IMPERIAL:
user_input[CONF_UNIT_SYSTEM] = CONF_UNIT_SYSTEM_IMPERIAL
else:
user_input[CONF_UNIT_SYSTEM] = CONF_UNIT_SYSTEM_METRIC
scan_interval = user_input.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
user_input[CONF_SCAN_INTERVAL] = scan_interval.seconds
return self.async_create_entry(title=identifier, data=user_input)

19
homeassistant/components/geonetnz_volcano/const.py Normal file
View File

@ -0,0 +1,19 @@
"""Define constants for the GeoNet NZ Volcano integration."""
from datetime import timedelta
DOMAIN = "geonetnz_volcano"
FEED = "feed"
ATTR_ACTIVITY = "activity"
ATTR_DISTANCE = "distance"
ATTR_EXTERNAL_ID = "external_id"
ATTR_HAZARDS = "hazards"
# Icon alias "mdi:mountain" not working.
DEFAULT_ICON = "mdi:image-filter-hdr"
DEFAULT_RADIUS = 50.0
DEFAULT_SCAN_INTERVAL = timedelta(minutes=5)
SIGNAL_NEW_SENSOR = "geonetnz_volcano_new_sensor_{}"
SIGNAL_UPDATE_ENTITY = "geonetnz_volcano_update_{}"

13
homeassistant/components/geonetnz_volcano/manifest.json Normal file
View File

@ -0,0 +1,13 @@
{
"domain": "geonetnz_volcano",
"name": "GeoNet NZ Volcano",
"config_flow": true,
"documentation": "https://www.home-assistant.io/components/geonetnz_volcano",
"requirements": [
"aio_geojson_geonetnz_volcano==0.5"
],
"dependencies": [],
"codeowners": [
"@exxamalte"
]
}

169
homeassistant/components/geonetnz_volcano/sensor.py Normal file
View File

@ -0,0 +1,169 @@
"""Feed Entity Manager Sensor support for GeoNet NZ Volcano Feeds."""
import logging
from typing import Optional
from homeassistant.const import (
CONF_UNIT_SYSTEM_IMPERIAL,
LENGTH_KILOMETERS,
ATTR_ATTRIBUTION,
ATTR_LONGITUDE,
ATTR_LATITUDE,
)
from homeassistant.core import callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from homeassistant.util import dt
from homeassistant.util.unit_system import IMPERIAL_SYSTEM
from .const import (
ATTR_ACTIVITY,
ATTR_DISTANCE,
ATTR_EXTERNAL_ID,
ATTR_HAZARDS,
DEFAULT_ICON,
DOMAIN,
FEED,
SIGNAL_UPDATE_ENTITY,
)
_LOGGER = logging.getLogger(__name__)
ATTR_LAST_UPDATE = "feed_last_update"
ATTR_LAST_UPDATE_SUCCESSFUL = "feed_last_update_successful"
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up the GeoNet NZ Volcano Feed platform."""
manager = hass.data[DOMAIN][FEED][entry.entry_id]
@callback
def async_add_sensor(feed_manager, external_id, unit_system):
"""Add sensor entity from feed."""
new_entity = GeonetnzVolcanoSensor(
entry.entry_id, feed_manager, external_id, unit_system
)
_LOGGER.debug("Adding sensor %s", new_entity)
async_add_entities([new_entity], True)
manager.listeners.append(
async_dispatcher_connect(
hass, manager.async_event_new_entity(), async_add_sensor
)
)
hass.async_create_task(manager.async_update())
_LOGGER.debug("Sensor setup done")
class GeonetnzVolcanoSensor(Entity):
"""This represents an external event with GeoNet NZ Volcano feed data."""
def __init__(self, config_entry_id, feed_manager, external_id, unit_system):
"""Initialize entity with data from feed entry."""
self._config_entry_id = config_entry_id
self._feed_manager = feed_manager
self._external_id = external_id
self._unit_system = unit_system
self._title = None
self._distance = None
self._latitude = None
self._longitude = None
self._attribution = None
self._alert_level = None
self._activity = None
self._hazards = None
self._feed_last_update = None
self._feed_last_update_successful = None
self._remove_signal_update = None
async def async_added_to_hass(self):
"""Call when entity is added to hass."""
self._remove_signal_update = async_dispatcher_connect(
self.hass,
SIGNAL_UPDATE_ENTITY.format(self._external_id),
self._update_callback,
)
async def async_will_remove_from_hass(self) -> None:
"""Call when entity will be removed from hass."""
if self._remove_signal_update:
self._remove_signal_update()
@callback
def _update_callback(self):
"""Call update method."""
self.async_schedule_update_ha_state(True)
@property
def should_poll(self):
"""No polling needed for GeoNet NZ Volcano feed location events."""
return False
async def async_update(self):
"""Update this entity from the data held in the feed manager."""
_LOGGER.debug("Updating %s", self._external_id)
feed_entry = self._feed_manager.get_entry(self._external_id)
last_update = self._feed_manager.last_update()
last_update_successful = self._feed_manager.last_update_successful()
if feed_entry:
self._update_from_feed(feed_entry, last_update, last_update_successful)
def _update_from_feed(self, feed_entry, last_update, last_update_successful):
"""Update the internal state from the provided feed entry."""
self._title = feed_entry.title
# Convert distance if not metric system.
if self._unit_system == CONF_UNIT_SYSTEM_IMPERIAL:
self._distance = round(
IMPERIAL_SYSTEM.length(feed_entry.distance_to_home, LENGTH_KILOMETERS),
1,
)
else:
self._distance = round(feed_entry.distance_to_home, 1)
self._latitude = round(feed_entry.coordinates[0], 5)
self._longitude = round(feed_entry.coordinates[1], 5)
self._attribution = feed_entry.attribution
self._alert_level = feed_entry.alert_level
self._activity = feed_entry.activity
self._hazards = feed_entry.hazards
self._feed_last_update = dt.as_utc(last_update) if last_update else None
self._feed_last_update_successful = (
dt.as_utc(last_update_successful) if last_update_successful else None
)
@property
def state(self):
"""Return the state of the sensor."""
return self._alert_level
@property
def icon(self):
"""Return the icon to use in the frontend, if any."""
return DEFAULT_ICON
@property
def name(self) -> Optional[str]:
"""Return the name of the entity."""
return f"Volcano {self._title}"
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
return "alert level"
@property
def device_state_attributes(self):
"""Return the device state attributes."""
attributes = {}
for key, value in (
(ATTR_EXTERNAL_ID, self._external_id),
(ATTR_ATTRIBUTION, self._attribution),
(ATTR_ACTIVITY, self._activity),
(ATTR_HAZARDS, self._hazards),
(ATTR_LONGITUDE, self._longitude),
(ATTR_LATITUDE, self._latitude),
(ATTR_DISTANCE, self._distance),
(ATTR_LAST_UPDATE, self._feed_last_update),
(ATTR_LAST_UPDATE_SUCCESSFUL, self._feed_last_update_successful),
):
if value or isinstance(value, bool):
attributes[key] = value
return attributes

16
homeassistant/components/geonetnz_volcano/strings.json Normal file
View File

@ -0,0 +1,16 @@
{
"config": {
"title": "GeoNet NZ Volcano",
"step": {
"user": {
"title": "Fill in your filter details.",
"data": {
"radius": "Radius"
}
}
},
"error": {
"identifier_exists": "Location already registered"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -24,6 +24,7 @@ FLOWS = [
"esphome",
"geofency",
"geonetnz_quakes",
"geonetnz_volcano",
"glances",
"gpslogger",
"hangouts",

3
requirements_all.txt
View File

@ -126,6 +126,9 @@ afsapi==0.0.4
# homeassistant.components.geonetnz_quakes
aio_geojson_geonetnz_quakes==0.11
# homeassistant.components.geonetnz_volcano
aio_geojson_geonetnz_volcano==0.5
# homeassistant.components.ambient_station
aioambient==0.3.2

3
requirements_test_all.txt
View File

@ -37,6 +37,9 @@ adguardhome==0.3.0
# homeassistant.components.geonetnz_quakes
aio_geojson_geonetnz_quakes==0.11
# homeassistant.components.geonetnz_volcano
aio_geojson_geonetnz_volcano==0.5
# homeassistant.components.ambient_station
aioambient==0.3.2

25
tests/components/geonetnz_volcano/__init__.py Normal file
View File

@ -0,0 +1,25 @@
"""The tests for the GeoNet NZ Volcano Feed integration."""
from unittest.mock import MagicMock
def _generate_mock_feed_entry(
external_id,
title,
alert_level,
distance_to_home,
coordinates,
attribution=None,
activity=None,
hazards=None,
):
"""Construct a mock feed entry for testing purposes."""
feed_entry = MagicMock()
feed_entry.external_id = external_id
feed_entry.title = title
feed_entry.alert_level = alert_level
feed_entry.distance_to_home = distance_to_home
feed_entry.coordinates = coordinates
feed_entry.attribution = attribution
feed_entry.activity = activity
feed_entry.hazards = hazards
return feed_entry

28
tests/components/geonetnz_volcano/conftest.py Normal file
View File

@ -0,0 +1,28 @@
"""Configuration for GeoNet NZ Volcano tests."""
import pytest
from homeassistant.components.geonetnz_volcano import DOMAIN
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_RADIUS,
CONF_UNIT_SYSTEM,
CONF_SCAN_INTERVAL,
)
from tests.common import MockConfigEntry
@pytest.fixture
def config_entry():
"""Create a mock GeoNet NZ Volcano config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_LATITUDE: -41.2,
CONF_LONGITUDE: 174.7,
CONF_RADIUS: 25,
CONF_UNIT_SYSTEM: "metric",
CONF_SCAN_INTERVAL: 300.0,
},
title="-41.2, 174.7",
)

81
tests/components/geonetnz_volcano/test_config_flow.py Normal file
View File

@ -0,0 +1,81 @@
"""Define tests for the GeoNet NZ Volcano config flow."""
from datetime import timedelta
from homeassistant import data_entry_flow
from homeassistant.components.geonetnz_volcano import config_flow
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_RADIUS,
CONF_SCAN_INTERVAL,
CONF_UNIT_SYSTEM,
)
async def test_duplicate_error(hass, config_entry):
"""Test that errors are shown when duplicates are added."""
conf = {CONF_LATITUDE: -41.2, CONF_LONGITUDE: 174.7, CONF_RADIUS: 25}
config_entry.add_to_hass(hass)
flow = config_flow.GeonetnzVolcanoFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=conf)
assert result["errors"] == {"base": "identifier_exists"}
async def test_show_form(hass):
"""Test that the form is served with no input."""
flow = config_flow.GeonetnzVolcanoFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=None)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_step_import(hass):
"""Test that the import step works."""
conf = {
CONF_LATITUDE: -41.2,
CONF_LONGITUDE: 174.7,
CONF_RADIUS: 25,
CONF_UNIT_SYSTEM: "metric",
CONF_SCAN_INTERVAL: timedelta(minutes=4),
}
flow = config_flow.GeonetnzVolcanoFlowHandler()
flow.hass = hass
result = await flow.async_step_import(import_config=conf)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "-41.2, 174.7"
assert result["data"] == {
CONF_LATITUDE: -41.2,
CONF_LONGITUDE: 174.7,
CONF_RADIUS: 25,
CONF_UNIT_SYSTEM: "metric",
CONF_SCAN_INTERVAL: 240.0,
}
async def test_step_user(hass):
"""Test that the user step works."""
hass.config.latitude = -41.2
hass.config.longitude = 174.7
conf = {CONF_RADIUS: 25}
flow = config_flow.GeonetnzVolcanoFlowHandler()
flow.hass = hass
result = await flow.async_step_user(user_input=conf)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "-41.2, 174.7"
assert result["data"] == {
CONF_LATITUDE: -41.2,
CONF_LONGITUDE: 174.7,
CONF_RADIUS: 25,
CONF_UNIT_SYSTEM: "metric",
CONF_SCAN_INTERVAL: 300.0,
}

22
tests/components/geonetnz_volcano/test_init.py Normal file
View File

@ -0,0 +1,22 @@
"""Define tests for the GeoNet NZ Volcano general setup."""
from asynctest import CoroutineMock, patch
from homeassistant.components.geonetnz_volcano import DOMAIN, FEED
async def test_component_unload_config_entry(hass, config_entry):
"""Test that loading and unloading of a config entry works."""
config_entry.add_to_hass(hass)
with patch(
"aio_geojson_geonetnz_volcano.GeonetnzVolcanoFeedManager.update",
new_callable=CoroutineMock,
) as mock_feed_manager_update:
# Load config entry.
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert mock_feed_manager_update.call_count == 1
assert hass.data[DOMAIN][FEED][config_entry.entry_id] is not None
# Unload config entry.
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[DOMAIN][FEED].get(config_entry.entry_id) is None

168
tests/components/geonetnz_volcano/test_sensor.py Normal file
View File

@ -0,0 +1,168 @@
"""The tests for the GeoNet NZ Volcano Feed integration."""
from asynctest import CoroutineMock, patch
from homeassistant.components import geonetnz_volcano
from homeassistant.components.geo_location import ATTR_DISTANCE
from homeassistant.components.geonetnz_volcano import DEFAULT_SCAN_INTERVAL
from homeassistant.components.geonetnz_volcano.const import (
ATTR_ACTIVITY,
ATTR_EXTERNAL_ID,
ATTR_HAZARDS,
)
from homeassistant.const import (
ATTR_ATTRIBUTION,
ATTR_FRIENDLY_NAME,
ATTR_ICON,
ATTR_LATITUDE,
ATTR_LONGITUDE,
ATTR_UNIT_OF_MEASUREMENT,
CONF_RADIUS,
EVENT_HOMEASSISTANT_START,
)
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from homeassistant.util.unit_system import IMPERIAL_SYSTEM
from tests.common import async_fire_time_changed
from tests.components.geonetnz_volcano import _generate_mock_feed_entry
CONFIG = {geonetnz_volcano.DOMAIN: {CONF_RADIUS: 200}}
async def test_setup(hass):
"""Test the general setup of the integration."""
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry(
"1234",
"Title 1",
1,
15.5,
(38.0, -3.0),
attribution="Attribution 1",
activity="Activity 1",
hazards="Hazards 1",
)
mock_entry_2 = _generate_mock_feed_entry("2345", "Title 2", 0, 20.5, (38.1, -3.1))
mock_entry_3 = _generate_mock_feed_entry("3456", "Title 3", 2, 25.5, (38.2, -3.2))
mock_entry_4 = _generate_mock_feed_entry("4567", "Title 4", 1, 12.5, (38.3, -3.3))
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"aio_geojson_client.feed.GeoJsonFeed.update", new_callable=CoroutineMock
) as mock_feed_update:
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_2, mock_entry_3]
assert await async_setup_component(hass, geonetnz_volcano.DOMAIN, CONFIG)
# Artificially trigger update and collect events.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
all_states = hass.states.async_all()
# 3 sensor entities
assert len(all_states) == 3
state = hass.states.get("sensor.volcano_title_1")
assert state is not None
assert state.name == "Volcano Title 1"
assert int(state.state) == 1
assert state.attributes[ATTR_EXTERNAL_ID] == "1234"
assert state.attributes[ATTR_LATITUDE] == 38.0
assert state.attributes[ATTR_LONGITUDE] == -3.0
assert state.attributes[ATTR_DISTANCE] == 15.5
assert state.attributes[ATTR_FRIENDLY_NAME] == "Volcano Title 1"
assert state.attributes[ATTR_ATTRIBUTION] == "Attribution 1"
assert state.attributes[ATTR_ACTIVITY] == "Activity 1"
assert state.attributes[ATTR_HAZARDS] == "Hazards 1"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "alert level"
assert state.attributes[ATTR_ICON] == "mdi:image-filter-hdr"
state = hass.states.get("sensor.volcano_title_2")
assert state is not None
assert state.name == "Volcano Title 2"
assert int(state.state) == 0
assert state.attributes[ATTR_EXTERNAL_ID] == "2345"
assert state.attributes[ATTR_LATITUDE] == 38.1
assert state.attributes[ATTR_LONGITUDE] == -3.1
assert state.attributes[ATTR_DISTANCE] == 20.5
assert state.attributes[ATTR_FRIENDLY_NAME] == "Volcano Title 2"
state = hass.states.get("sensor.volcano_title_3")
assert state is not None
assert state.name == "Volcano Title 3"
assert int(state.state) == 2
assert state.attributes[ATTR_EXTERNAL_ID] == "3456"
assert state.attributes[ATTR_LATITUDE] == 38.2
assert state.attributes[ATTR_LONGITUDE] == -3.2
assert state.attributes[ATTR_DISTANCE] == 25.5
assert state.attributes[ATTR_FRIENDLY_NAME] == "Volcano Title 3"
# Simulate an update - two existing, one new entry, one outdated entry
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_4, mock_entry_3]
async_fire_time_changed(hass, utcnow + DEFAULT_SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 4
# Simulate an update - empty data, but successful update,
# so no changes to entities.
mock_feed_update.return_value = "OK_NO_DATA", None
async_fire_time_changed(hass, utcnow + 2 * DEFAULT_SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 4
# Simulate an update - empty data, keep all entities
mock_feed_update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + 3 * DEFAULT_SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 4
# Simulate an update - regular data for 3 entries
mock_feed_update.return_value = "OK", [mock_entry_1, mock_entry_2, mock_entry_3]
async_fire_time_changed(hass, utcnow + 4 * DEFAULT_SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 4
async def test_setup_imperial(hass):
"""Test the setup of the integration using imperial unit system."""
hass.config.units = IMPERIAL_SYSTEM
# Set up some mock feed entries for this test.
mock_entry_1 = _generate_mock_feed_entry("1234", "Title 1", 1, 15.5, (38.0, -3.0))
# Patching 'utcnow' to gain more control over the timed update.
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"aio_geojson_client.feed.GeoJsonFeed.update", new_callable=CoroutineMock
) as mock_feed_update, patch(
"aio_geojson_client.feed.GeoJsonFeed.__init__"
) as mock_feed_init:
mock_feed_update.return_value = "OK", [mock_entry_1]
assert await async_setup_component(hass, geonetnz_volcano.DOMAIN, CONFIG)
# Artificially trigger update and collect events.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
# Test conversion of 200 miles to kilometers.
assert mock_feed_init.call_args[1].get("filter_radius") == 321.8688
state = hass.states.get("sensor.volcano_title_1")
assert state is not None
assert state.name == "Volcano Title 1"
assert int(state.state) == 1
assert state.attributes[ATTR_EXTERNAL_ID] == "1234"
assert state.attributes[ATTR_LATITUDE] == 38.0
assert state.attributes[ATTR_LONGITUDE] == -3.0
assert state.attributes[ATTR_DISTANCE] == 9.6
assert state.attributes[ATTR_FRIENDLY_NAME] == "Volcano Title 1"
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "alert level"
assert state.attributes[ATTR_ICON] == "mdi:image-filter-hdr"