Fix buienradar component and add smoke tests (#27965)

* Fixes the buienradar component and add smoke tests

  * Fix errors due to circular imports after imports were moved.
  * Add smoke test so this situation will be caught in the future.

* Add buienradar.util to coveragerc

* Refactor tests to standalone pytest test function style

* Add __init__ to buienradar tests
This commit is contained in:
Ties de Kock 2019-10-21 13:52:25 +02:00 committed by cgtobi
parent ea1401d0b6
commit a05144bb8b
8 changed files with 294 additions and 211 deletions

View file

@ -100,6 +100,7 @@ omit =
homeassistant/components/bt_home_hub_5/device_tracker.py
homeassistant/components/bt_smarthub/device_tracker.py
homeassistant/components/buienradar/sensor.py
homeassistant/components/buienradar/util.py
homeassistant/components/buienradar/weather.py
homeassistant/components/caldav/calendar.py
homeassistant/components/canary/alarm_control_panel.py

View file

@ -0,0 +1,7 @@
"""Constants for buienradar component."""
DEFAULT_TIMEFRAME = 60
"""Schedule next call after (minutes)."""
SCHEDULE_OK = 10
"""When an error occurred, new call after (minutes)."""
SCHEDULE_NOK = 2

View file

@ -1,38 +1,23 @@
"""Support for Buienradar.nl weather service."""
import asyncio
from datetime import datetime, timedelta
import logging
import aiohttp
import async_timeout
from buienradar.buienradar import parse_data
from buienradar.constants import (
ATTRIBUTION,
CONDCODE,
CONDITION,
CONTENT,
DATA,
DETAILED,
EXACT,
EXACTNL,
FORECAST,
HUMIDITY,
IMAGE,
MEASURED,
MESSAGE,
PRECIPITATION_FORECAST,
PRESSURE,
STATIONNAME,
STATUS_CODE,
SUCCESS,
TEMPERATURE,
TIMEFRAME,
VISIBILITY,
WINDAZIMUTH,
WINDGUST,
WINDSPEED,
)
from buienradar.urls import JSON_FEED_URL, json_precipitation_forecast_url
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
@ -44,13 +29,14 @@ from homeassistant.const import (
CONF_NAME,
TEMP_CELSIUS,
)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util import dt as dt_util
from .weather import DEFAULT_TIMEFRAME
from .const import DEFAULT_TIMEFRAME
from .util import BrData
_LOGGER = logging.getLogger(__name__)
@ -465,194 +451,3 @@ class BrSensor(Entity):
def force_update(self):
"""Return true for continuous sensors, false for discrete sensors."""
return self._force_update
class BrData:
"""Get the latest data and updates the states."""
def __init__(self, hass, coordinates, timeframe, devices):
"""Initialize the data object."""
self.devices = devices
self.data = {}
self.hass = hass
self.coordinates = coordinates
self.timeframe = timeframe
async def update_devices(self):
"""Update all devices/sensors."""
if self.devices:
tasks = []
# Update all devices
for dev in self.devices:
if dev.load_data(self.data):
tasks.append(dev.async_update_ha_state())
if tasks:
await asyncio.wait(tasks)
async def schedule_update(self, minute=1):
"""Schedule an update after minute minutes."""
_LOGGER.debug("Scheduling next update in %s minutes.", minute)
nxt = dt_util.utcnow() + timedelta(minutes=minute)
async_track_point_in_utc_time(self.hass, self.async_update, nxt)
async def get_data(self, url):
"""Load data from specified url."""
_LOGGER.debug("Calling url: %s...", url)
result = {SUCCESS: False, MESSAGE: None}
resp = None
try:
websession = async_get_clientsession(self.hass)
with async_timeout.timeout(10):
resp = await websession.get(url)
result[STATUS_CODE] = resp.status
result[CONTENT] = await resp.text()
if resp.status == 200:
result[SUCCESS] = True
else:
result[MESSAGE] = "Got http statuscode: %d" % (resp.status)
return result
except (asyncio.TimeoutError, aiohttp.ClientError) as err:
result[MESSAGE] = "%s" % err
return result
finally:
if resp is not None:
await resp.release()
async def async_update(self, *_):
"""Update the data from buienradar."""
content = await self.get_data(JSON_FEED_URL)
if content.get(SUCCESS) is not True:
# unable to get the data
_LOGGER.warning(
"Unable to retrieve json data from Buienradar."
"(Msg: %s, status: %s,)",
content.get(MESSAGE),
content.get(STATUS_CODE),
)
# schedule new call
await self.schedule_update(SCHEDULE_NOK)
return
# rounding coordinates prevents unnecessary redirects/calls
lat = self.coordinates[CONF_LATITUDE]
lon = self.coordinates[CONF_LONGITUDE]
rainurl = json_precipitation_forecast_url(lat, lon)
raincontent = await self.get_data(rainurl)
if raincontent.get(SUCCESS) is not True:
# unable to get the data
_LOGGER.warning(
"Unable to retrieve raindata from Buienradar." "(Msg: %s, status: %s,)",
raincontent.get(MESSAGE),
raincontent.get(STATUS_CODE),
)
# schedule new call
await self.schedule_update(SCHEDULE_NOK)
return
result = parse_data(
content.get(CONTENT),
raincontent.get(CONTENT),
self.coordinates[CONF_LATITUDE],
self.coordinates[CONF_LONGITUDE],
self.timeframe,
False,
)
_LOGGER.debug("Buienradar parsed data: %s", result)
if result.get(SUCCESS) is not True:
if int(datetime.now().strftime("%H")) > 0:
_LOGGER.warning(
"Unable to parse data from Buienradar." "(Msg: %s)",
result.get(MESSAGE),
)
await self.schedule_update(SCHEDULE_NOK)
return
self.data = result.get(DATA)
await self.update_devices()
await self.schedule_update(SCHEDULE_OK)
@property
def attribution(self):
"""Return the attribution."""
return self.data.get(ATTRIBUTION)
@property
def stationname(self):
"""Return the name of the selected weatherstation."""
return self.data.get(STATIONNAME)
@property
def condition(self):
"""Return the condition."""
return self.data.get(CONDITION)
@property
def temperature(self):
"""Return the temperature, or None."""
try:
return float(self.data.get(TEMPERATURE))
except (ValueError, TypeError):
return None
@property
def pressure(self):
"""Return the pressure, or None."""
try:
return float(self.data.get(PRESSURE))
except (ValueError, TypeError):
return None
@property
def humidity(self):
"""Return the humidity, or None."""
try:
return int(self.data.get(HUMIDITY))
except (ValueError, TypeError):
return None
@property
def visibility(self):
"""Return the visibility, or None."""
try:
return int(self.data.get(VISIBILITY))
except (ValueError, TypeError):
return None
@property
def wind_speed(self):
"""Return the windspeed, or None."""
try:
return float(self.data.get(WINDSPEED))
except (ValueError, TypeError):
return None
@property
def wind_bearing(self):
"""Return the wind bearing, or None."""
try:
return int(self.data.get(WINDAZIMUTH))
except (ValueError, TypeError):
return None
@property
def forecast(self):
"""Return the forecast data."""
return self.data.get(FORECAST)

View file

@ -0,0 +1,228 @@
"""Shared utilities for different supported platforms."""
import asyncio
from datetime import datetime, timedelta
import logging
import aiohttp
import async_timeout
from buienradar.buienradar import parse_data
from buienradar.constants import (
ATTRIBUTION,
CONDITION,
CONTENT,
DATA,
FORECAST,
HUMIDITY,
MESSAGE,
PRESSURE,
STATIONNAME,
STATUS_CODE,
SUCCESS,
TEMPERATURE,
VISIBILITY,
WINDAZIMUTH,
WINDSPEED,
)
from buienradar.urls import JSON_FEED_URL, json_precipitation_forecast_url
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util import dt as dt_util
from .const import SCHEDULE_OK, SCHEDULE_NOK
_LOGGER = logging.getLogger(__name__)
class BrData:
"""Get the latest data and updates the states."""
def __init__(self, hass, coordinates, timeframe, devices):
"""Initialize the data object."""
self.devices = devices
self.data = {}
self.hass = hass
self.coordinates = coordinates
self.timeframe = timeframe
async def update_devices(self):
"""Update all devices/sensors."""
if self.devices:
tasks = []
# Update all devices
for dev in self.devices:
if dev.load_data(self.data):
tasks.append(dev.async_update_ha_state())
if tasks:
await asyncio.wait(tasks)
async def schedule_update(self, minute=1):
"""Schedule an update after minute minutes."""
_LOGGER.debug("Scheduling next update in %s minutes.", minute)
nxt = dt_util.utcnow() + timedelta(minutes=minute)
async_track_point_in_utc_time(self.hass, self.async_update, nxt)
async def get_data(self, url):
"""Load data from specified url."""
_LOGGER.debug("Calling url: %s...", url)
result = {SUCCESS: False, MESSAGE: None}
resp = None
try:
websession = async_get_clientsession(self.hass)
with async_timeout.timeout(10):
resp = await websession.get(url)
result[STATUS_CODE] = resp.status
result[CONTENT] = await resp.text()
if resp.status == 200:
result[SUCCESS] = True
else:
result[MESSAGE] = "Got http statuscode: %d" % (resp.status)
return result
except (asyncio.TimeoutError, aiohttp.ClientError) as err:
result[MESSAGE] = "%s" % err
return result
finally:
if resp is not None:
await resp.release()
async def async_update(self, *_):
"""Update the data from buienradar."""
content = await self.get_data(JSON_FEED_URL)
if content.get(SUCCESS) is not True:
# unable to get the data
_LOGGER.warning(
"Unable to retrieve json data from Buienradar."
"(Msg: %s, status: %s,)",
content.get(MESSAGE),
content.get(STATUS_CODE),
)
# schedule new call
await self.schedule_update(SCHEDULE_NOK)
return
# rounding coordinates prevents unnecessary redirects/calls
lat = self.coordinates[CONF_LATITUDE]
lon = self.coordinates[CONF_LONGITUDE]
rainurl = json_precipitation_forecast_url(lat, lon)
raincontent = await self.get_data(rainurl)
if raincontent.get(SUCCESS) is not True:
# unable to get the data
_LOGGER.warning(
"Unable to retrieve raindata from Buienradar." "(Msg: %s, status: %s,)",
raincontent.get(MESSAGE),
raincontent.get(STATUS_CODE),
)
# schedule new call
await self.schedule_update(SCHEDULE_NOK)
return
result = parse_data(
content.get(CONTENT),
raincontent.get(CONTENT),
self.coordinates[CONF_LATITUDE],
self.coordinates[CONF_LONGITUDE],
self.timeframe,
False,
)
_LOGGER.debug("Buienradar parsed data: %s", result)
if result.get(SUCCESS) is not True:
if int(datetime.now().strftime("%H")) > 0:
_LOGGER.warning(
"Unable to parse data from Buienradar." "(Msg: %s)",
result.get(MESSAGE),
)
await self.schedule_update(SCHEDULE_NOK)
return
self.data = result.get(DATA)
await self.update_devices()
await self.schedule_update(SCHEDULE_OK)
@property
def attribution(self):
"""Return the attribution."""
return self.data.get(ATTRIBUTION)
@property
def stationname(self):
"""Return the name of the selected weatherstation."""
return self.data.get(STATIONNAME)
@property
def condition(self):
"""Return the condition."""
return self.data.get(CONDITION)
@property
def temperature(self):
"""Return the temperature, or None."""
try:
return float(self.data.get(TEMPERATURE))
except (ValueError, TypeError):
return None
@property
def pressure(self):
"""Return the pressure, or None."""
try:
return float(self.data.get(PRESSURE))
except (ValueError, TypeError):
return None
@property
def humidity(self):
"""Return the humidity, or None."""
try:
return int(self.data.get(HUMIDITY))
except (ValueError, TypeError):
return None
@property
def visibility(self):
"""Return the visibility, or None."""
try:
return int(self.data.get(VISIBILITY))
except (ValueError, TypeError):
return None
@property
def wind_speed(self):
"""Return the windspeed, or None."""
try:
return float(self.data.get(WINDSPEED))
except (ValueError, TypeError):
return None
@property
def wind_bearing(self):
"""Return the wind bearing, or None."""
try:
return int(self.data.get(WINDAZIMUTH))
except (ValueError, TypeError):
return None
@property
def forecast(self):
"""Return the forecast data."""
return self.data.get(FORECAST)

View file

@ -28,13 +28,13 @@ from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME, TEMP_C
from homeassistant.helpers import config_validation as cv
# Reuse data and API logic from the sensor implementation
from .sensor import BrData
from .util import BrData
from .const import DEFAULT_TIMEFRAME
_LOGGER = logging.getLogger(__name__)
DATA_CONDITION = "buienradar_condition"
DEFAULT_TIMEFRAME = 60
CONF_FORECAST = "forecast"

View file

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

View file

@ -0,0 +1,26 @@
"""The tests for the Buienradar sensor platform."""
from homeassistant.setup import async_setup_component
from homeassistant.components import sensor
CONDITIONS = ["stationname", "temperature"]
BASE_CONFIG = {
"sensor": [
{
"platform": "buienradar",
"name": "volkel",
"latitude": 51.65,
"longitude": 5.7,
"monitored_conditions": CONDITIONS,
}
]
}
async def test_smoke_test_setup_component(hass):
"""Smoke test for successfully set-up with default config."""
assert await async_setup_component(hass, sensor.DOMAIN, BASE_CONFIG)
for cond in CONDITIONS:
state = hass.states.get(f"sensor.volkel_{cond}")
assert state.state == "unknown"

View file

@ -0,0 +1,25 @@
"""The tests for the buienradar weather component."""
from homeassistant.components import weather
from homeassistant.setup import async_setup_component
# Example config snippet from documentation.
BASE_CONFIG = {
"weather": [
{
"platform": "buienradar",
"name": "volkel",
"latitude": 51.65,
"longitude": 5.7,
"forecast": True,
}
]
}
async def test_smoke_test_setup_component(hass):
"""Smoke test for successfully set-up with default config."""
assert await async_setup_component(hass, weather.DOMAIN, BASE_CONFIG)
state = hass.states.get("weather.volkel")
assert state.state == "unknown"