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

GeoNet NZ Quakes feed integration (#25736) 

* initial working version

* support configured unit system and convert distance automatically

* properly unloading integration when removed and stopping refreshes

* ran isort

* fixed pylint

* introduced time filter of seven days into past

* adding unit tests

* fixed lint

* removed unused code

* added test case

* added test case for config flow

* fixed lint

* fixed comment

* removed unused test code

* increased test coverage

* fixed filtering by time

* changed wording in config flow

* reformatted with black

* removed unused logger

* fixed black

* changed default mmi

* reduced the options in the config flow form; fixed a few schema options and processing of data

* moved unsubscribing signals

* fixed minimum magnitude and modified tests

* fixed radius in imperial unit system

* increased test coverage

* simplified code

* fixed lint

* changed string formatting; simplified code

* removed unused strings

* added translation
pull/25848/head
Malte Franken 2019-08-10 20:50:27 +10:00 committed by Martin Hjelmare
parent e685f077ae
commit 9e6732e530
15 changed files with 908 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CODEOWNERS
View File

@ -102,6 +102,7 @@ homeassistant/components/fronius/* @nielstron
homeassistant/components/frontend/* @home-assistant/frontend
homeassistant/components/gearbest/* @HerrHofrat
homeassistant/components/geniushub/* @zxdavb
homeassistant/components/geonetnz_quakes/* @exxamalte
homeassistant/components/gitter/* @fabaff
homeassistant/components/glances/* @fabaff
homeassistant/components/gntp/* @robbiet480

17
homeassistant/components/geonetnz_quakes/.translations/en.json Normal file
View File

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

101
homeassistant/components/geonetnz_quakes/__init__.py Normal file
View File

@ -0,0 +1,101 @@
"""The GeoNet NZ Quakes integration."""
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_RADIUS,
CONF_SCAN_INTERVAL,
)
from homeassistant.helpers import config_validation as cv
from .config_flow import configured_instances
from .const import (
CONF_MINIMUM_MAGNITUDE,
CONF_MMI,
DEFAULT_MINIMUM_MAGNITUDE,
DEFAULT_MMI,
DEFAULT_RADIUS,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
FEED,
)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_MMI, default=DEFAULT_MMI): vol.All(
vol.Coerce(int), vol.Range(min=-1, max=8)
),
vol.Optional(CONF_RADIUS, default=DEFAULT_RADIUS): vol.Coerce(float),
vol.Optional(
CONF_MINIMUM_MAGNITUDE, default=DEFAULT_MINIMUM_MAGNITUDE
): vol.All(vol.Coerce(float), vol.Range(min=0)),
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 Quakes 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)
mmi = conf[CONF_MMI]
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_MINIMUM_MAGNITUDE: conf[CONF_MINIMUM_MAGNITUDE],
CONF_MMI: mmi,
CONF_SCAN_INTERVAL: scan_interval,
},
)
)
return True
async def async_setup_entry(hass, config_entry):
"""Set up the GeoNet NZ Quakes component as config entry."""
hass.data[DOMAIN] = {}
hass.data[DOMAIN][FEED] = {}
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(config_entry, "geo_location")
)
return True
async def async_unload_entry(hass, config_entry):
"""Unload an GeoNet NZ Quakes component config entry."""
manager = hass.data[DOMAIN][FEED].pop(config_entry.entry_id)
await manager.async_stop()
await hass.config_entries.async_forward_entry_unload(config_entry, "geo_location")
return True

94
homeassistant/components/geonetnz_quakes/config_flow.py Normal file
View File

@ -0,0 +1,94 @@
"""Config flow to configure the GeoNet NZ Quakes 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 (
CONF_MMI,
DEFAULT_MMI,
DEFAULT_RADIUS,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
DEFAULT_MINIMUM_MAGNITUDE,
CONF_MINIMUM_MAGNITUDE,
)
_LOGGER = logging.getLogger(__name__)
@callback
def configured_instances(hass):
"""Return a set of configured GeoNet NZ Quakes instances."""
return set(
f"{entry.data[CONF_LATITUDE]}, {entry.data[CONF_LONGITUDE]}"
for entry in hass.config_entries.async_entries(DOMAIN)
)
@config_entries.HANDLERS.register(DOMAIN)
class GeonetnzQuakesFlowHandler(config_entries.ConfigFlow):
"""Handle a GeoNet NZ Quakes 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_MMI, default=DEFAULT_MMI): vol.All(
vol.Coerce(int), vol.Range(min=-1, max=8)
),
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."""
_LOGGER.debug("User input: %s", user_input)
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
minimum_magnitude = user_input.get(
CONF_MINIMUM_MAGNITUDE, DEFAULT_MINIMUM_MAGNITUDE
)
user_input[CONF_MINIMUM_MAGNITUDE] = minimum_magnitude
return self.async_create_entry(title=identifier, data=user_input)

14
homeassistant/components/geonetnz_quakes/const.py Normal file
View File

@ -0,0 +1,14 @@
"""Define constants for the GeoNet NZ Quakes integration."""
from datetime import timedelta
DOMAIN = "geonetnz_quakes"
CONF_MINIMUM_MAGNITUDE = "minimum_magnitude"
CONF_MMI = "mmi"
FEED = "feed"
DEFAULT_MINIMUM_MAGNITUDE = 0.0
DEFAULT_MMI = 3
DEFAULT_RADIUS = 50.0
DEFAULT_SCAN_INTERVAL = timedelta(minutes=5)

284
homeassistant/components/geonetnz_quakes/geo_location.py Normal file
View File

@ -0,0 +1,284 @@
"""Geolocation support for GeoNet NZ Quakes Feeds."""
from datetime import timedelta
import logging
from typing import Optional
from aio_geojson_geonetnz_quakes import GeonetnzQuakesFeedManager
from homeassistant.components.geo_location import GeolocationEvent
from homeassistant.const import (
ATTR_ATTRIBUTION,
CONF_LATITUDE,
CONF_LONGITUDE,
CONF_RADIUS,
CONF_SCAN_INTERVAL,
CONF_UNIT_SYSTEM,
CONF_UNIT_SYSTEM_IMPERIAL,
LENGTH_KILOMETERS,
LENGTH_MILES,
ATTR_TIME,
)
from homeassistant.core import callback
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.util.unit_system import IMPERIAL_SYSTEM, METRIC_SYSTEM
from .const import CONF_MINIMUM_MAGNITUDE, CONF_MMI, DOMAIN, FEED
_LOGGER = logging.getLogger(__name__)
ATTR_DEPTH = "depth"
ATTR_EXTERNAL_ID = "external_id"
ATTR_LOCALITY = "locality"
ATTR_MAGNITUDE = "magnitude"
ATTR_MMI = "mmi"
ATTR_PUBLICATION_DATE = "publication_date"
ATTR_QUALITY = "quality"
DEFAULT_FILTER_TIME_INTERVAL = timedelta(days=7)
SIGNAL_DELETE_ENTITY = "geonetnz_quakes_delete_{}"
SIGNAL_UPDATE_ENTITY = "geonetnz_quakes_update_{}"
SOURCE = "geonetnz_quakes"
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up the GeoNet NZ Quakes Feed platform."""
radius = entry.data[CONF_RADIUS]
unit_system = entry.data[CONF_UNIT_SYSTEM]
if unit_system == CONF_UNIT_SYSTEM_IMPERIAL:
radius = METRIC_SYSTEM.length(radius, LENGTH_MILES)
manager = GeonetnzQuakesFeedEntityManager(
hass,
async_add_entities,
entry.data[CONF_SCAN_INTERVAL],
entry.data[CONF_LATITUDE],
entry.data[CONF_LONGITUDE],
entry.data[CONF_MMI],
radius,
unit_system,
entry.data[CONF_MINIMUM_MAGNITUDE],
)
hass.data[DOMAIN][FEED][entry.entry_id] = manager
await manager.async_init()
class GeonetnzQuakesFeedEntityManager:
"""Feed Entity Manager for GeoNet NZ Quakes feed."""
def __init__(
self,
hass,
async_add_entities,
scan_interval,
latitude,
longitude,
mmi,
radius_in_km,
unit_system,
minimum_magnitude,
):
"""Initialize the Feed Entity Manager."""
self._hass = hass
coordinates = (latitude, longitude)
websession = aiohttp_client.async_get_clientsession(hass)
self._feed_manager = GeonetnzQuakesFeedManager(
websession,
self._generate_entity,
self._update_entity,
self._remove_entity,
coordinates,
mmi=mmi,
filter_radius=radius_in_km,
filter_minimum_magnitude=minimum_magnitude,
filter_time=DEFAULT_FILTER_TIME_INTERVAL,
)
self._async_add_entities = async_add_entities
self._scan_interval = timedelta(seconds=scan_interval)
self._unit_system = unit_system
self._track_time_remove_callback = None
async def async_init(self):
"""Schedule regular updates based on configured time interval."""
async def update(event_time):
"""Update."""
await self.async_update()
await self.async_update()
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."""
if self._track_time_remove_callback:
self._track_time_remove_callback()
_LOGGER.debug("Feed entity manager stopped")
def get_entry(self, external_id):
"""Get feed entry by external id."""
return self._feed_manager.feed_entries.get(external_id)
async def _generate_entity(self, external_id):
"""Generate new entity."""
new_entity = GeonetnzQuakesEvent(self, external_id, self._unit_system)
# Add new entities to HA.
self._async_add_entities([new_entity], True)
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):
"""Remove entity."""
async_dispatcher_send(self._hass, SIGNAL_DELETE_ENTITY.format(external_id))
class GeonetnzQuakesEvent(GeolocationEvent):
"""This represents an external event with GeoNet NZ Quakes feed data."""
def __init__(self, feed_manager, external_id, unit_system):
"""Initialize entity with data from feed entry."""
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._depth = None
self._locality = None
self._magnitude = None
self._mmi = None
self._quality = None
self._time = None
self._remove_signal_delete = None
self._remove_signal_update = None
async def async_added_to_hass(self):
"""Call when entity is added to hass."""
self._remove_signal_delete = async_dispatcher_connect(
self.hass,
SIGNAL_DELETE_ENTITY.format(self._external_id),
self._delete_callback,
)
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."""
self._remove_signal_delete()
self._remove_signal_update()
@callback
def _delete_callback(self):
"""Remove this entity."""
self.hass.async_create_task(self.async_remove())
@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 Quakes 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)
if feed_entry:
self._update_from_feed(feed_entry)
def _update_from_feed(self, feed_entry):
"""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 = IMPERIAL_SYSTEM.length(
feed_entry.distance_to_home, LENGTH_KILOMETERS
)
else:
self._distance = feed_entry.distance_to_home
self._latitude = feed_entry.coordinates[0]
self._longitude = feed_entry.coordinates[1]
self._attribution = feed_entry.attribution
self._depth = feed_entry.depth
self._locality = feed_entry.locality
self._magnitude = feed_entry.magnitude
self._mmi = feed_entry.mmi
self._quality = feed_entry.quality
self._time = feed_entry.time
@property
def icon(self):
"""Return the icon to use in the frontend, if any."""
return "mdi:pulse"
@property
def source(self) -> str:
"""Return source value of this external event."""
return SOURCE
@property
def name(self) -> Optional[str]:
"""Return the name of the entity."""
return self._title
@property
def distance(self) -> Optional[float]:
"""Return distance value of this external event."""
return self._distance
@property
def latitude(self) -> Optional[float]:
"""Return latitude value of this external event."""
return self._latitude
@property
def longitude(self) -> Optional[float]:
"""Return longitude value of this external event."""
return self._longitude
@property
def unit_of_measurement(self):
"""Return the unit of measurement."""
if self._unit_system == CONF_UNIT_SYSTEM_IMPERIAL:
return LENGTH_MILES
return LENGTH_KILOMETERS
@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_DEPTH, self._depth),
(ATTR_LOCALITY, self._locality),
(ATTR_MAGNITUDE, self._magnitude),
(ATTR_MMI, self._mmi),
(ATTR_QUALITY, self._quality),
(ATTR_TIME, self._time),
):
if value or isinstance(value, bool):
attributes[key] = value
return attributes

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

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

17
homeassistant/components/geonetnz_quakes/strings.json Normal file
View File

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

1
homeassistant/generated/config_flows.py
View File

@ -16,6 +16,7 @@ FLOWS = [
"emulated_roku",
"esphome",
"geofency",
"geonetnz_quakes",
"gpslogger",
"hangouts",
"heos",

3
requirements_all.txt
View File

@ -114,6 +114,9 @@ adguardhome==0.2.1
# homeassistant.components.frontier_silicon
afsapi==0.0.4
# homeassistant.components.geonetnz_quakes
aio_geojson_geonetnz_quakes==0.5
# homeassistant.components.ambient_station
aioambient==0.3.1

3
requirements_test_all.txt
View File

@ -42,6 +42,9 @@ YesssSMS==0.2.3
# homeassistant.components.adguard
adguardhome==0.2.1
# homeassistant.components.geonetnz_quakes
aio_geojson_geonetnz_quakes==0.5
# homeassistant.components.ambient_station
aioambient==0.3.1

1
script/gen_requirements_all.py
View File

@ -43,6 +43,7 @@ COMMENT_REQUIREMENTS = (
TEST_REQUIREMENTS = (
"adguardhome",
"ambiclimate",
"aio_geojson_geonetnz_quakes",
"aioambient",
"aioautomatic",
"aiobotocore",

1
tests/components/geonetnz_quakes/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the geonetnz_quakes component."""

135
tests/components/geonetnz_quakes/test_config_flow.py Normal file
View File

@ -0,0 +1,135 @@
"""Define tests for the GeoNet NZ Quakes config flow."""
from datetime import timedelta
import pytest
from asynctest import patch, CoroutineMock
from homeassistant import data_entry_flow
from homeassistant.components.geonetnz_quakes import (
async_setup_entry,
config_flow,
CONF_MMI,
CONF_MINIMUM_MAGNITUDE,
DOMAIN,
async_unload_entry,
FEED,
)
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 Quakes 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,
CONF_MMI: 4,
CONF_MINIMUM_MAGNITUDE: 0.0,
},
title="-41.2, 174.7",
)
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.GeonetnzQuakesFlowHandler()
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.GeonetnzQuakesFlowHandler()
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_MMI: 2,
CONF_SCAN_INTERVAL: timedelta(minutes=4),
CONF_MINIMUM_MAGNITUDE: 2.5,
}
flow = config_flow.GeonetnzQuakesFlowHandler()
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_MMI: 2,
CONF_UNIT_SYSTEM: "metric",
CONF_SCAN_INTERVAL: 240.0,
CONF_MINIMUM_MAGNITUDE: 2.5,
}
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, CONF_MMI: 4}
flow = config_flow.GeonetnzQuakesFlowHandler()
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_MMI: 4,
CONF_UNIT_SYSTEM: "metric",
CONF_SCAN_INTERVAL: 300.0,
CONF_MINIMUM_MAGNITUDE: 0.0,
}
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_quakes.GeonetnzQuakesFeedManager.update",
new_callable=CoroutineMock,
) as mock_feed_manager_update:
# Load config entry.
assert await async_setup_entry(hass, config_entry)
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 async_unload_entry(hass, config_entry)
await hass.async_block_till_done()
assert hass.data[DOMAIN][FEED].get(config_entry.entry_id) is None

223
tests/components/geonetnz_quakes/test_geo_location.py Normal file
View File

@ -0,0 +1,223 @@
"""The tests for the GeoNet NZ Quakes Feed integration."""
import datetime
from unittest.mock import MagicMock
from asynctest import patch, CoroutineMock
from homeassistant.components import geonetnz_quakes
from homeassistant.components.geo_location import ATTR_SOURCE
from homeassistant.components.geonetnz_quakes import DEFAULT_SCAN_INTERVAL
from homeassistant.components.geonetnz_quakes.geo_location import (
ATTR_EXTERNAL_ID,
ATTR_MAGNITUDE,
ATTR_LOCALITY,
ATTR_MMI,
ATTR_DEPTH,
ATTR_QUALITY,
)
from homeassistant.const import (
EVENT_HOMEASSISTANT_START,
CONF_RADIUS,
ATTR_LATITUDE,
ATTR_LONGITUDE,
ATTR_FRIENDLY_NAME,
ATTR_UNIT_OF_MEASUREMENT,
ATTR_ATTRIBUTION,
ATTR_TIME,
ATTR_ICON,
)
from homeassistant.setup import async_setup_component
from homeassistant.util.unit_system import IMPERIAL_SYSTEM
from tests.common import async_fire_time_changed
import homeassistant.util.dt as dt_util
CONFIG = {geonetnz_quakes.DOMAIN: {CONF_RADIUS: 200}}
def _generate_mock_feed_entry(
external_id,
title,
distance_to_home,
coordinates,
attribution=None,
depth=None,
magnitude=None,
mmi=None,
locality=None,
quality=None,
time=None,
):
"""Construct a mock feed entry for testing purposes."""
feed_entry = MagicMock()
feed_entry.external_id = external_id
feed_entry.title = title
feed_entry.distance_to_home = distance_to_home
feed_entry.coordinates = coordinates
feed_entry.attribution = attribution
feed_entry.depth = depth
feed_entry.magnitude = magnitude
feed_entry.mmi = mmi
feed_entry.locality = locality
feed_entry.quality = quality
feed_entry.time = time
return feed_entry
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",
15.5,
(38.0, -3.0),
locality="Locality 1",
attribution="Attribution 1",
time=datetime.datetime(2018, 9, 22, 8, 0, tzinfo=datetime.timezone.utc),
magnitude=5.7,
mmi=5,
depth=10.5,
quality="best",
)
mock_entry_2 = _generate_mock_feed_entry(
"2345", "Title 2", 20.5, (38.1, -3.1), magnitude=4.6
)
mock_entry_3 = _generate_mock_feed_entry(
"3456", "Title 3", 25.5, (38.2, -3.2), locality="Locality 3"
)
mock_entry_4 = _generate_mock_feed_entry("4567", "Title 4", 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_quakes.DOMAIN, CONFIG)
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 3
state = hass.states.get("geo_location.title_1")
assert state is not None
assert state.name == "Title 1"
assert state.attributes == {
ATTR_EXTERNAL_ID: "1234",
ATTR_LATITUDE: 38.0,
ATTR_LONGITUDE: -3.0,
ATTR_FRIENDLY_NAME: "Title 1",
ATTR_LOCALITY: "Locality 1",
ATTR_ATTRIBUTION: "Attribution 1",
ATTR_TIME: datetime.datetime(
2018, 9, 22, 8, 0, tzinfo=datetime.timezone.utc
),
ATTR_MAGNITUDE: 5.7,
ATTR_DEPTH: 10.5,
ATTR_MMI: 5,
ATTR_QUALITY: "best",
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: "geonetnz_quakes",
ATTR_ICON: "mdi:pulse",
}
assert float(state.state) == 15.5
state = hass.states.get("geo_location.title_2")
assert state is not None
assert state.name == "Title 2"
assert state.attributes == {
ATTR_EXTERNAL_ID: "2345",
ATTR_LATITUDE: 38.1,
ATTR_LONGITUDE: -3.1,
ATTR_FRIENDLY_NAME: "Title 2",
ATTR_MAGNITUDE: 4.6,
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: "geonetnz_quakes",
ATTR_ICON: "mdi:pulse",
}
assert float(state.state) == 20.5
state = hass.states.get("geo_location.title_3")
assert state is not None
assert state.name == "Title 3"
assert state.attributes == {
ATTR_EXTERNAL_ID: "3456",
ATTR_LATITUDE: 38.2,
ATTR_LONGITUDE: -3.2,
ATTR_FRIENDLY_NAME: "Title 3",
ATTR_LOCALITY: "Locality 3",
ATTR_UNIT_OF_MEASUREMENT: "km",
ATTR_SOURCE: "geonetnz_quakes",
ATTR_ICON: "mdi:pulse",
}
assert float(state.state) == 25.5
# Simulate an update - one 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) == 3
# 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) == 3
# Simulate an update - empty data, removes 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) == 0
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", 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__", new_callable=CoroutineMock
) as mock_feed_init:
mock_feed_update.return_value = "OK", [mock_entry_1]
assert await async_setup_component(hass, geonetnz_quakes.DOMAIN, CONFIG)
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
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("geo_location.title_1")
assert state is not None
assert state.name == "Title 1"
assert state.attributes == {
ATTR_EXTERNAL_ID: "1234",
ATTR_LATITUDE: 38.0,
ATTR_LONGITUDE: -3.0,
ATTR_FRIENDLY_NAME: "Title 1",
ATTR_UNIT_OF_MEASUREMENT: "mi",
ATTR_SOURCE: "geonetnz_quakes",
ATTR_ICON: "mdi:pulse",
}
assert float(state.state) == 9.6