Add FreeDNS component (#13526)

* Add FreeDNS component

* Implement review changes in FreeDNS component

* Implement review changes in FreeDNS component

* Implement review changes in FreeDNS component
This commit is contained in:
Beat 2018-03-30 21:33:30 +02:00 committed by Pascal Vizeli
parent 9cfcd38c1e
commit 8fad97a47a
2 changed files with 172 additions and 0 deletions

View File

@ -0,0 +1,103 @@
"""
Integrate with FreeDNS Dynamic DNS service at freedns.afraid.org.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/freedns/
"""
import asyncio
from datetime import timedelta
import logging
import aiohttp
import async_timeout
import voluptuous as vol
from homeassistant.const import (CONF_URL, CONF_ACCESS_TOKEN)
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'freedns'
DEFAULT_INTERVAL = timedelta(minutes=10)
TIMEOUT = 10
UPDATE_URL = 'https://freedns.afraid.org/dynamic/update.php'
CONF_UPDATE_INTERVAL = 'update_interval'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Exclusive(CONF_URL, DOMAIN): cv.string,
vol.Exclusive(CONF_ACCESS_TOKEN, DOMAIN): cv.string,
vol.Optional(CONF_UPDATE_INTERVAL, default=DEFAULT_INTERVAL): vol.All(
cv.time_period, cv.positive_timedelta),
})
}, extra=vol.ALLOW_EXTRA)
@asyncio.coroutine
def async_setup(hass, config):
"""Initialize the FreeDNS component."""
url = config[DOMAIN].get(CONF_URL)
auth_token = config[DOMAIN].get(CONF_ACCESS_TOKEN)
update_interval = config[DOMAIN].get(CONF_UPDATE_INTERVAL)
session = hass.helpers.aiohttp_client.async_get_clientsession()
result = yield from _update_freedns(
hass, session, url, auth_token)
if result is False:
return False
@asyncio.coroutine
def update_domain_callback(now):
"""Update the FreeDNS entry."""
yield from _update_freedns(hass, session, url, auth_token)
hass.helpers.event.async_track_time_interval(
update_domain_callback, update_interval)
return True
@asyncio.coroutine
def _update_freedns(hass, session, url, auth_token):
"""Update FreeDNS."""
params = None
if url is None:
url = UPDATE_URL
if auth_token is not None:
params = {}
params[auth_token] = ""
try:
with async_timeout.timeout(TIMEOUT, loop=hass.loop):
resp = yield from session.get(url, params=params)
body = yield from resp.text()
if "has not changed" in body:
# IP has not changed.
_LOGGER.debug("FreeDNS update skipped: IP has not changed")
return True
if "ERROR" not in body:
_LOGGER.debug("Updating FreeDNS was successful: %s", body)
return True
if "Invalid update URL" in body:
_LOGGER.error("FreeDNS update token is invalid")
else:
_LOGGER.warning("Updating FreeDNS failed: %s", body)
except aiohttp.ClientError:
_LOGGER.warning("Can't connect to FreeDNS API")
except asyncio.TimeoutError:
_LOGGER.warning("Timeout from FreeDNS API at %s", url)
return False

View File

@ -0,0 +1,69 @@
"""Test the FreeDNS component."""
import asyncio
import pytest
from homeassistant.setup import async_setup_component
from homeassistant.components import freedns
from homeassistant.util.dt import utcnow
from tests.common import async_fire_time_changed
ACCESS_TOKEN = 'test_token'
UPDATE_INTERVAL = freedns.DEFAULT_INTERVAL
UPDATE_URL = freedns.UPDATE_URL
@pytest.fixture
def setup_freedns(hass, aioclient_mock):
"""Fixture that sets up FreeDNS."""
params = {}
params[ACCESS_TOKEN] = ""
aioclient_mock.get(
UPDATE_URL, params=params, text='Successfully updated 1 domains.')
hass.loop.run_until_complete(async_setup_component(hass, freedns.DOMAIN, {
freedns.DOMAIN: {
'access_token': ACCESS_TOKEN,
'update_interval': UPDATE_INTERVAL,
}
}))
@asyncio.coroutine
def test_setup(hass, aioclient_mock):
"""Test setup works if update passes."""
params = {}
params[ACCESS_TOKEN] = ""
aioclient_mock.get(
UPDATE_URL, params=params, text='ERROR: Address has not changed.')
result = yield from async_setup_component(hass, freedns.DOMAIN, {
freedns.DOMAIN: {
'access_token': ACCESS_TOKEN,
'update_interval': UPDATE_INTERVAL,
}
})
assert result
assert aioclient_mock.call_count == 1
async_fire_time_changed(hass, utcnow() + UPDATE_INTERVAL)
yield from hass.async_block_till_done()
assert aioclient_mock.call_count == 2
@asyncio.coroutine
def test_setup_fails_if_wrong_token(hass, aioclient_mock):
"""Test setup fails if first update fails through wrong token."""
params = {}
params[ACCESS_TOKEN] = ""
aioclient_mock.get(
UPDATE_URL, params=params, text='ERROR: Invalid update URL (2)')
result = yield from async_setup_component(hass, freedns.DOMAIN, {
freedns.DOMAIN: {
'access_token': ACCESS_TOKEN,
'update_interval': UPDATE_INTERVAL,
}
})
assert not result
assert aioclient_mock.call_count == 1