Hassio api v3 (#7323)

* HassIO rest API v3

* fix content type

* fix lint

* Update comment

* fix content type

* change proxy handling

* fix handling

* fix register

* fix addons

* fix routing

* Update hassio to just proxy

* Fix tests

* Lint
This commit is contained in:
Paulus Schoutsen 2017-04-26 22:36:48 -07:00 committed by GitHub
parent 3374169c74
commit b14c07a60c
2 changed files with 261 additions and 807 deletions

View file

@ -7,70 +7,52 @@ https://home-assistant.io/components/hassio/
import asyncio
import logging
import os
import re
import aiohttp
from aiohttp import web
from aiohttp.web_exceptions import HTTPBadGateway
from aiohttp.web_exceptions import (
HTTPBadGateway, HTTPNotFound, HTTPMethodNotAllowed)
from aiohttp.hdrs import CONTENT_TYPE
import async_timeout
import voluptuous as vol
from homeassistant.config import load_yaml_config_file
from homeassistant.const import CONTENT_TYPE_TEXT_PLAIN
from homeassistant.components.http import HomeAssistantView
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
DOMAIN = 'hassio'
DEPENDENCIES = ['http']
_LOGGER = logging.getLogger(__name__)
LONG_TASK_TIMEOUT = 900
DEFAULT_TIMEOUT = 10
TIMEOUT = 10
SERVICE_HOST_SHUTDOWN = 'host_shutdown'
SERVICE_HOST_REBOOT = 'host_reboot'
HASSIO_REST_COMMANDS = {
'host/shutdown': ['POST'],
'host/reboot': ['POST'],
'host/update': ['GET'],
'host/info': ['GET'],
'supervisor/info': ['GET'],
'supervisor/update': ['POST'],
'supervisor/options': ['POST'],
'supervisor/reload': ['POST'],
'supervisor/logs': ['GET'],
'homeassistant/info': ['GET'],
'homeassistant/update': ['POST'],
'homeassistant/logs': ['GET'],
'network/info': ['GET'],
'network/options': ['GET'],
}
SERVICE_HOST_UPDATE = 'host_update'
SERVICE_HOMEASSISTANT_UPDATE = 'homeassistant_update'
SERVICE_SUPERVISOR_UPDATE = 'supervisor_update'
SERVICE_SUPERVISOR_RELOAD = 'supervisor_reload'
SERVICE_ADDON_INSTALL = 'addon_install'
SERVICE_ADDON_UNINSTALL = 'addon_uninstall'
SERVICE_ADDON_UPDATE = 'addon_update'
SERVICE_ADDON_START = 'addon_start'
SERVICE_ADDON_STOP = 'addon_stop'
ATTR_ADDON = 'addon'
ATTR_VERSION = 'version'
SCHEMA_SERVICE_UPDATE = vol.Schema({
vol.Optional(ATTR_VERSION): cv.string,
})
SCHEMA_SERVICE_ADDONS = vol.Schema({
vol.Required(ATTR_ADDON): cv.slug,
})
SCHEMA_SERVICE_ADDONS_VERSION = SCHEMA_SERVICE_ADDONS.extend({
vol.Optional(ATTR_VERSION): cv.string,
})
SERVICE_MAP = {
SERVICE_HOST_SHUTDOWN: None,
SERVICE_HOST_REBOOT: None,
SERVICE_HOST_UPDATE: SCHEMA_SERVICE_UPDATE,
SERVICE_HOMEASSISTANT_UPDATE: SCHEMA_SERVICE_UPDATE,
SERVICE_SUPERVISOR_UPDATE: SCHEMA_SERVICE_UPDATE,
SERVICE_SUPERVISOR_RELOAD: None,
SERVICE_ADDON_INSTALL: SCHEMA_SERVICE_ADDONS_VERSION,
SERVICE_ADDON_UNINSTALL: SCHEMA_SERVICE_ADDONS,
SERVICE_ADDON_START: SCHEMA_SERVICE_ADDONS,
SERVICE_ADDON_STOP: SCHEMA_SERVICE_ADDONS,
SERVICE_ADDON_UPDATE: SCHEMA_SERVICE_ADDONS_VERSION,
ADDON_REST_COMMANDS = {
'install': ['POST'],
'uninstall': ['POST'],
'start': ['POST'],
'stop': ['POST'],
'update': ['POST'],
'options': ['POST'],
'info': ['GET'],
'logs': ['GET'],
}
@ -91,71 +73,7 @@ def async_setup(hass, config):
_LOGGER.error("Not connected with HassIO!")
return False
# register base api views
for base in ('host', 'homeassistant'):
hass.http.register_view(HassIOBaseView(hassio, base))
for base in ('supervisor', 'network'):
hass.http.register_view(HassIOBaseEditView(hassio, base))
for base in ('supervisor', 'homeassistant'):
hass.http.register_view(HassIOBaseLogsView(hassio, base))
# register view for addons
hass.http.register_view(HassIOAddonsView(hassio))
hass.http.register_view(HassIOAddonsLogsView(hassio))
@asyncio.coroutine
def async_service_handler(service):
"""Handle HassIO service calls."""
addon = service.data.get(ATTR_ADDON)
if ATTR_VERSION in service.data:
version = {ATTR_VERSION: service.data[ATTR_VERSION]}
else:
version = None
# map to api call
if service.service == SERVICE_HOST_UPDATE:
yield from hassio.send_command(
"/host/update", payload=version)
elif service.service == SERVICE_HOST_REBOOT:
yield from hassio.send_command("/host/reboot")
elif service.service == SERVICE_HOST_SHUTDOWN:
yield from hassio.send_command("/host/shutdown")
elif service.service == SERVICE_SUPERVISOR_UPDATE:
yield from hassio.send_command(
"/supervisor/update", payload=version,
timeout=LONG_TASK_TIMEOUT)
elif service.service == SERVICE_SUPERVISOR_RELOAD:
yield from hassio.send_command(
"/supervisor/reload", timeout=LONG_TASK_TIMEOUT)
elif service.service == SERVICE_HOMEASSISTANT_UPDATE:
yield from hassio.send_command(
"/homeassistant/update", payload=version,
timeout=LONG_TASK_TIMEOUT)
elif service.service == SERVICE_ADDON_INSTALL:
yield from hassio.send_command(
"/addons/{}/install".format(addon), payload=version,
timeout=LONG_TASK_TIMEOUT)
elif service.service == SERVICE_ADDON_UNINSTALL:
yield from hassio.send_command(
"/addons/{}/uninstall".format(addon))
elif service.service == SERVICE_ADDON_START:
yield from hassio.send_command("/addons/{}/start".format(addon))
elif service.service == SERVICE_ADDON_STOP:
yield from hassio.send_command(
"/addons/{}/stop".format(addon), timeout=LONG_TASK_TIMEOUT)
elif service.service == SERVICE_ADDON_UPDATE:
yield from hassio.send_command(
"/addons/{}/update".format(addon), payload=version,
timeout=LONG_TASK_TIMEOUT)
descriptions = yield from hass.loop.run_in_executor(
None, load_yaml_config_file, os.path.join(
os.path.dirname(__file__), 'services.yaml'))
for service, schema in SERVICE_MAP.items():
hass.services.async_register(
DOMAIN, service, async_service_handler,
descriptions[DOMAIN][service], schema=schema)
hass.http.register_view(HassIOView(hassio))
return True
@ -169,165 +87,122 @@ class HassIO(object):
self.websession = websession
self._ip = ip
@asyncio.coroutine
def is_connected(self):
"""Return True if it connected to HassIO supervisor.
Return a coroutine.
This method is a coroutine.
"""
return self.send_command("/supervisor/ping")
try:
with async_timeout.timeout(TIMEOUT, loop=self.loop):
request = yield from self.websession.get(
"http://{}{}".format(self._ip, "/supervisor/ping")
)
@asyncio.coroutine
def send_command(self, cmd, payload=None, timeout=DEFAULT_TIMEOUT):
"""Send request to API."""
answer = yield from self.send_raw(
cmd, payload=payload, timeout=timeout
)
if answer and answer['result'] == 'ok':
return answer['data'] if answer['data'] else True
elif answer:
_LOGGER.error("%s return error %s.", cmd, answer['message'])
if request.status != 200:
_LOGGER.error("Ping return code %d.", request.status)
return False
answer = yield from request.json()
return answer and answer['result'] == 'ok'
except asyncio.TimeoutError:
_LOGGER.error("Timeout on ping request")
except aiohttp.ClientError as err:
_LOGGER.error("Client error on ping request %s", err)
return False
@asyncio.coroutine
def send_raw(self, cmd, payload=None, timeout=DEFAULT_TIMEOUT, json=True):
"""Send raw request to API."""
def command_proxy(self, path, request):
"""Return a client request with proxy origin for HassIO supervisor.
This method is a coroutine.
"""
try:
with async_timeout.timeout(timeout, loop=self.loop):
request = yield from self.websession.get(
"http://{}{}".format(self._ip, cmd),
timeout=None, json=payload
)
data = None
headers = None
with async_timeout.timeout(TIMEOUT, loop=self.loop):
data = yield from request.read()
if data:
headers = {CONTENT_TYPE: request.content_type}
else:
data = None
if request.status != 200:
_LOGGER.error("%s return code %d.", cmd, request.status)
return
method = getattr(self.websession, request.method.lower())
client = yield from method(
"http://{}/{}".format(self._ip, path), data=data,
headers=headers
)
if json:
return (yield from request.json())
return client
# get raw output
return (yield from request.read())
except aiohttp.ClientError as err:
_LOGGER.error("Client error on api %s request %s.", path, err)
except asyncio.TimeoutError:
_LOGGER.error("Timeout on api request %s.", cmd)
_LOGGER.error("Client timeout error on api request %s.", path)
except aiohttp.ClientError:
_LOGGER.error("Client error on api request %s.", cmd)
raise HTTPBadGateway()
class HassIOBaseView(HomeAssistantView):
class HassIOView(HomeAssistantView):
"""HassIO view to handle base part."""
name = "api:hassio"
url = "/api/hassio/{path:.+}"
requires_auth = True
def __init__(self, hassio, base):
"""Initialize a hassio base view."""
self.hassio = hassio
self._url_info = "/{}/info".format(base)
self.url = "/api/hassio/{}".format(base)
self.name = "api:hassio:{}".format(base)
@asyncio.coroutine
def get(self, request):
"""Get base data."""
data = yield from self.hassio.send_command(self._url_info)
if not data:
raise HTTPBadGateway()
return web.json_response(data)
class HassIOBaseEditView(HassIOBaseView):
"""HassIO view to handle base with options support."""
def __init__(self, hassio, base):
"""Initialize a hassio base edit view."""
super().__init__(hassio, base)
self._url_options = "/{}/options".format(base)
@asyncio.coroutine
def post(self, request):
"""Set options on host."""
data = yield from request.json()
response = yield from self.hassio.send_raw(
self._url_options, payload=data)
if not response:
raise HTTPBadGateway()
return web.json_response(response)
class HassIOBaseLogsView(HomeAssistantView):
"""HassIO view to handle base logs part."""
requires_auth = True
def __init__(self, hassio, base):
"""Initialize a hassio base view."""
self.hassio = hassio
self._url_logs = "/{}/logs".format(base)
self.url = "/api/hassio/logs/{}".format(base)
self.name = "api:hassio:logs:{}".format(base)
@asyncio.coroutine
def get(self, request):
"""Get logs."""
data = yield from self.hassio.send_raw(self._url_logs, json=False)
if not data:
raise HTTPBadGateway()
return web.Response(body=data)
class HassIOAddonsView(HomeAssistantView):
"""HassIO view to handle addons part."""
requires_auth = True
url = "/api/hassio/addons/{addon}"
name = "api:hassio:addons"
def __init__(self, hassio):
"""Initialize a hassio addon view."""
"""Initialize a hassio base view."""
self.hassio = hassio
@asyncio.coroutine
def get(self, request, addon):
"""Get addon data."""
data = yield from self.hassio.send_command(
"/addons/{}/info".format(addon))
if not data:
raise HTTPBadGateway()
return web.json_response(data)
def _handle(self, request, path):
"""Route data to hassio."""
if path.startswith('addons/'):
parts = path.split('/')
@asyncio.coroutine
def post(self, request, addon):
"""Set options on host."""
data = yield from request.json()
if len(parts) != 3:
raise HTTPNotFound()
response = yield from self.hassio.send_raw(
"/addons/{}/options".format(addon), payload=data)
if not response:
raise HTTPBadGateway()
return web.json_response(response)
allowed_methods = ADDON_REST_COMMANDS.get(parts[-1])
else:
allowed_methods = HASSIO_REST_COMMANDS.get(path)
if allowed_methods is None:
raise HTTPNotFound()
if request.method not in allowed_methods:
raise HTTPMethodNotAllowed(request.method, allowed_methods)
client = yield from self.hassio.command_proxy(path, request)
data = yield from client.read()
if path.endswith('/logs'):
return _create_response_log(client, data)
return _create_response(client, data)
get = _handle
post = _handle
class HassIOAddonsLogsView(HomeAssistantView):
"""HassIO view to handle addons logs part."""
def _create_response(client, data):
"""Convert a response from client request."""
return web.Response(
body=data,
status=client.status,
content_type=client.content_type,
)
requires_auth = True
url = "/api/hassio/logs/addons/{addon}"
name = "api:hassio:logs:addons"
def __init__(self, hassio):
"""Initialize a hassio addon view."""
self.hassio = hassio
def _create_response_log(client, data):
"""Convert a response from client request."""
# Remove color codes
log = re.sub(r"\x1b(\[.*?[@-~]|\].*?(\x07|\x1b\\))", "", data.decode())
@asyncio.coroutine
def get(self, request, addon):
"""Get addon data."""
data = yield from self.hassio.send_raw(
"/addons/{}/logs".format(addon), json=False)
if not data:
raise HTTPBadGateway()
return web.Response(body=data)
return web.Response(
text=log,
status=client.status,
content_type=CONTENT_TYPE_TEXT_PLAIN,
)

View file

@ -1,605 +1,184 @@
"""The tests for the hassio component."""
import asyncio
import os
from unittest.mock import patch
from unittest.mock import patch, Mock, MagicMock
import aiohttp
import pytest
import homeassistant.components.hassio as ho
from homeassistant.setup import setup_component, async_setup_component
from homeassistant.setup import async_setup_component
from tests.common import (
get_test_home_assistant, assert_setup_component)
from tests.common import mock_coro, mock_http_component_app
@pytest.fixture
def hassio_env():
"""Fixture to inject hassio env."""
with patch.dict(os.environ, {'HASSIO': "127.0.0.1"}) as env_mock:
yield env_mock
class TestHassIOSetup(object):
"""Test the hassio component."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.config = {
ho.DOMAIN: {},
}
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component(self, aioclient_mock, hassio_env):
"""Test setup component."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
def test_setup_component_bad(self, aioclient_mock):
"""Test setup component bad."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
with assert_setup_component(0, ho.DOMAIN):
assert not setup_component(self.hass, ho.DOMAIN, self.config)
assert len(aioclient_mock.mock_calls) == 0
def test_setup_component_test_service(self, aioclient_mock, hassio_env):
"""Test setup component and check if service exits."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_HOST_REBOOT)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_HOST_SHUTDOWN)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_HOST_UPDATE)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_SUPERVISOR_UPDATE)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_SUPERVISOR_RELOAD)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_ADDON_INSTALL)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_ADDON_UNINSTALL)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_ADDON_UPDATE)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_ADDON_START)
assert self.hass.services.has_service(
ho.DOMAIN, ho.SERVICE_ADDON_STOP)
class TestHassIOComponent(object):
"""Test the HassIO component."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.config = {
ho.DOMAIN: {},
}
self.url = "http://127.0.0.1/{}"
self.error_msg = {
'result': 'error',
'message': 'Test error',
}
self.ok_msg = {
'result': 'ok',
'data': {},
}
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_rest_command_timeout(self, aioclient_mock, hassio_env):
"""Call a hassio with timeout."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/update"), exc=asyncio.TimeoutError())
self.hass.services.call(ho.DOMAIN, ho.SERVICE_HOST_UPDATE, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_aiohttp_error(self, aioclient_mock, hassio_env):
"""Call a hassio with aiohttp exception."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/update"), exc=aiohttp.ClientError())
self.hass.services.call(ho.DOMAIN, ho.SERVICE_HOST_UPDATE, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_error(self, aioclient_mock, hassio_env):
"""Call a hassio with status code 503."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/update"), status=503)
self.hass.services.call(ho.DOMAIN, ho.SERVICE_HOST_UPDATE, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_error_api(self, aioclient_mock, hassio_env):
"""Call a hassio with status code 503."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/update"), json=self.error_msg)
self.hass.services.call(ho.DOMAIN, ho.SERVICE_HOST_UPDATE, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_host_reboot(self, aioclient_mock, hassio_env):
"""Call a hassio for host reboot."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/reboot"), json=self.ok_msg)
self.hass.services.call(ho.DOMAIN, ho.SERVICE_HOST_REBOOT, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_host_shutdown(self, aioclient_mock, hassio_env):
"""Call a hassio for host shutdown."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/shutdown"), json=self.ok_msg)
self.hass.services.call(ho.DOMAIN, ho.SERVICE_HOST_SHUTDOWN, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_host_update(self, aioclient_mock, hassio_env):
"""Call a hassio for host update."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("host/update"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_HOST_UPDATE, {'version': '0.4'})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert aioclient_mock.mock_calls[-1][2]['version'] == '0.4'
def test_rest_command_http_supervisor_update(self, aioclient_mock,
hassio_env):
"""Call a hassio for supervisor update."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("supervisor/update"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_SUPERVISOR_UPDATE, {'version': '0.4'})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert aioclient_mock.mock_calls[-1][2]['version'] == '0.4'
def test_rest_command_http_supervisor_reload(self, aioclient_mock,
hassio_env):
"""Call a hassio for supervisor reload."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("supervisor/reload"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_SUPERVISOR_RELOAD, {})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_homeassistant_update(self, aioclient_mock,
hassio_env):
"""Call a hassio for homeassistant update."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("homeassistant/update"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_HOMEASSISTANT_UPDATE, {'version': '0.4'})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert aioclient_mock.mock_calls[-1][2]['version'] == '0.4'
def test_rest_command_http_addon_install(self, aioclient_mock, hassio_env):
"""Call a hassio for addon install."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("addons/smb_config/install"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_ADDON_INSTALL, {
'addon': 'smb_config',
'version': '0.4'
})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert aioclient_mock.mock_calls[-1][2]['version'] == '0.4'
def test_rest_command_http_addon_uninstall(self, aioclient_mock,
hassio_env):
"""Call a hassio for addon uninstall."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("addons/smb_config/uninstall"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_ADDON_UNINSTALL, {
'addon': 'smb_config'
})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_addon_update(self, aioclient_mock, hassio_env):
"""Call a hassio for addon update."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("addons/smb_config/update"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_ADDON_UPDATE, {
'addon': 'smb_config',
'version': '0.4'
})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert aioclient_mock.mock_calls[-1][2]['version'] == '0.4'
def test_rest_command_http_addon_start(self, aioclient_mock, hassio_env):
"""Call a hassio for addon start."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("addons/smb_config/start"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_ADDON_START, {
'addon': 'smb_config',
})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
def test_rest_command_http_addon_stop(self, aioclient_mock, hassio_env):
"""Call a hassio for addon stop."""
aioclient_mock.get(
"http://127.0.0.1/supervisor/ping", json=self.ok_msg)
with assert_setup_component(0, ho.DOMAIN):
setup_component(self.hass, ho.DOMAIN, self.config)
aioclient_mock.get(
self.url.format("addons/smb_config/stop"), json=self.ok_msg)
self.hass.services.call(
ho.DOMAIN, ho.SERVICE_ADDON_STOP, {
'addon': 'smb_config'
})
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
with patch.dict(os.environ, {'HASSIO': "127.0.0.1"}), \
patch('homeassistant.components.hassio.HassIO.is_connected',
Mock(return_value=mock_coro(True))):
yield
@pytest.fixture
def hassio_client(hassio_env, hass, test_client):
"""Create mock hassio http client."""
app = mock_http_component_app(hass)
hass.loop.run_until_complete(async_setup_component(hass, 'hassio', {}))
hass.http.views['api:hassio'].register(app.router)
yield hass.loop.run_until_complete(test_client(app))
@asyncio.coroutine
def test_async_hassio_host_view(aioclient_mock, hass, test_client, hassio_env):
"""Test that it fetches the given url."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
result = yield from async_setup_component(hass, ho.DOMAIN, {ho.DOMAIN: {}})
assert result, 'Failed to setup hasio'
client = yield from test_client(hass.http.app)
aioclient_mock.get('http://127.0.0.1/host/info', json={
'result': 'ok',
'data': {
'os': 'resinos',
'version': '0.3',
'current': '0.4',
'level': 16,
'hostname': 'test',
}
})
resp = yield from client.get('/api/hassio/host')
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 2
assert resp.status == 200
assert data['os'] == 'resinos'
assert data['version'] == '0.3'
assert data['current'] == '0.4'
assert data['level'] == 16
assert data['hostname'] == 'test'
def test_fail_setup_without_environ_var(hass):
"""Fail setup if no environ variable set."""
with patch.dict(os.environ, {}, clear=True):
result = yield from async_setup_component(hass, 'hassio', {})
assert not result
@asyncio.coroutine
def test_async_hassio_homeassistant_view(aioclient_mock, hass, test_client,
hassio_env):
"""Test that it fetches the given url."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
result = yield from async_setup_component(hass, ho.DOMAIN, {ho.DOMAIN: {}})
assert result, 'Failed to setup hasio'
client = yield from test_client(hass.http.app)
aioclient_mock.get('http://127.0.0.1/homeassistant/info', json={
'result': 'ok',
'data': {
'version': '0.41',
'current': '0.41.1',
}
})
resp = yield from client.get('/api/hassio/homeassistant')
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 2
assert resp.status == 200
assert data['version'] == '0.41'
assert data['current'] == '0.41.1'
aioclient_mock.get('http://127.0.0.1/homeassistant/logs',
content=b"That is a test log")
resp = yield from client.get('/api/hassio/logs/homeassistant')
data = yield from resp.read()
assert len(aioclient_mock.mock_calls) == 3
assert resp.status == 200
assert data == b"That is a test log"
def test_fail_setup_cannot_connect(hass):
"""Fail setup if cannot connect."""
with patch.dict(os.environ, {'HASSIO': "127.0.0.1"}), \
patch('homeassistant.components.hassio.HassIO.is_connected',
Mock(return_value=mock_coro(False))):
result = yield from async_setup_component(hass, 'hassio', {})
assert not result
@asyncio.coroutine
def test_async_hassio_supervisor_view(aioclient_mock, hass, test_client,
hassio_env):
"""Test that it fetches the given url."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
result = yield from async_setup_component(hass, ho.DOMAIN, {ho.DOMAIN: {}})
assert result, 'Failed to setup hasio'
def test_invalid_path(hassio_client):
"""Test requesting invalid path."""
with patch.dict(ho.HASSIO_REST_COMMANDS, {}, clear=True):
resp = yield from hassio_client.post('/api/hassio/beer')
client = yield from test_client(hass.http.app)
aioclient_mock.get('http://127.0.0.1/supervisor/info', json={
'result': 'ok',
'data': {
'version': '0.3',
'current': '0.4',
'beta': False,
}
})
resp = yield from client.get('/api/hassio/supervisor')
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 2
assert resp.status == 200
assert data['version'] == '0.3'
assert data['current'] == '0.4'
assert not data['beta']
aioclient_mock.get('http://127.0.0.1/supervisor/options', json={
'result': 'ok',
'data': {},
})
resp = yield from client.post('/api/hassio/supervisor', json={
'beta': True,
})
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 3
assert resp.status == 200
assert aioclient_mock.mock_calls[-1][2]['beta']
aioclient_mock.get('http://127.0.0.1/supervisor/logs',
content=b"That is a test log")
resp = yield from client.get('/api/hassio/logs/supervisor')
data = yield from resp.read()
assert len(aioclient_mock.mock_calls) == 4
assert resp.status == 200
assert data == b"That is a test log"
assert resp.status == 404
@asyncio.coroutine
def test_async_hassio_network_view(aioclient_mock, hass, test_client,
hassio_env):
"""Test that it fetches the given url."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
result = yield from async_setup_component(hass, ho.DOMAIN, {ho.DOMAIN: {}})
assert result, 'Failed to setup hasio'
def test_invalid_method(hassio_client):
"""Test requesting path with invalid method."""
with patch.dict(ho.HASSIO_REST_COMMANDS, {'beer': ['POST']}):
resp = yield from hassio_client.get('/api/hassio/beer')
client = yield from test_client(hass.http.app)
aioclient_mock.get('http://127.0.0.1/network/info', json={
'result': 'ok',
'data': {
'mode': 'dhcp',
'ssid': 'my_wlan',
'password': '123456',
}
})
resp = yield from client.get('/api/hassio/network')
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 2
assert resp.status == 200
assert data['mode'] == 'dhcp'
assert data['ssid'] == 'my_wlan'
assert data['password'] == '123456'
aioclient_mock.get('http://127.0.0.1/network/options', json={
'result': 'ok',
'data': {},
})
resp = yield from client.post('/api/hassio/network', json={
'mode': 'dhcp',
'ssid': 'my_wlan2',
'password': '654321',
})
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 3
assert resp.status == 200
assert aioclient_mock.mock_calls[-1][2]['ssid'] == 'my_wlan2'
assert aioclient_mock.mock_calls[-1][2]['password'] == '654321'
assert resp.status == 405
@asyncio.coroutine
def test_async_hassio_addon_view(aioclient_mock, hass, test_client,
hassio_env):
"""Test that it fetches the given url."""
aioclient_mock.get("http://127.0.0.1/supervisor/ping", json={
'result': 'ok', 'data': {}
})
result = yield from async_setup_component(hass, ho.DOMAIN, {ho.DOMAIN: {}})
assert result, 'Failed to setup hasio'
def test_forward_normal_path(hassio_client):
"""Test fetching normal path."""
response = MagicMock()
response.read.return_value = mock_coro('data')
client = yield from test_client(hass.http.app)
with patch.dict(ho.HASSIO_REST_COMMANDS, {'beer': ['POST']}), \
patch('homeassistant.components.hassio.HassIO.command_proxy',
Mock(return_value=mock_coro(response))), \
patch('homeassistant.components.hassio._create_response') as mresp:
mresp.return_value = 'response'
resp = yield from hassio_client.post('/api/hassio/beer')
aioclient_mock.get('http://127.0.0.1/addons/smb_config/info', json={
'result': 'ok',
'data': {
'name': 'SMB Config',
'state': 'running',
'boot': 'auto',
'options': {
'bla': False,
}
}
})
resp = yield from client.get('/api/hassio/addons/smb_config')
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 2
# Check we got right response
assert resp.status == 200
assert data['name'] == 'SMB Config'
assert data['state'] == 'running'
assert data['boot'] == 'auto'
assert not data['options']['bla']
body = yield from resp.text()
assert body == 'response'
aioclient_mock.get('http://127.0.0.1/addons/smb_config/options', json={
'result': 'ok',
'data': {},
})
# Check we forwarded command
assert len(mresp.mock_calls) == 1
assert mresp.mock_calls[0][1] == (response, 'data')
resp = yield from client.post('/api/hassio/addons/smb_config', json={
'boot': 'manual',
'options': {
'bla': True,
}
})
data = yield from resp.json()
assert len(aioclient_mock.mock_calls) == 3
@asyncio.coroutine
def test_forward_normal_log_path(hassio_client):
"""Test fetching normal log path."""
response = MagicMock()
response.read.return_value = mock_coro('data')
with patch.dict(ho.HASSIO_REST_COMMANDS, {'beer/logs': ['GET']}), \
patch('homeassistant.components.hassio.HassIO.command_proxy',
Mock(return_value=mock_coro(response))), \
patch('homeassistant.components.hassio.'
'_create_response_log') as mresp:
mresp.return_value = 'response'
resp = yield from hassio_client.get('/api/hassio/beer/logs')
# Check we got right response
assert resp.status == 200
assert aioclient_mock.mock_calls[-1][2]['boot'] == 'manual'
assert aioclient_mock.mock_calls[-1][2]['options']['bla']
body = yield from resp.text()
assert body == 'response'
aioclient_mock.get('http://127.0.0.1/addons/smb_config/logs',
content=b"That is a test log")
# Check we forwarded command
assert len(mresp.mock_calls) == 1
assert mresp.mock_calls[0][1] == (response, 'data')
resp = yield from client.get('/api/hassio/logs/addons/smb_config')
data = yield from resp.read()
assert len(aioclient_mock.mock_calls) == 4
@asyncio.coroutine
def test_forward_addon_path(hassio_client):
"""Test fetching addon path."""
response = MagicMock()
response.read.return_value = mock_coro('data')
with patch.dict(ho.ADDON_REST_COMMANDS, {'install': ['POST']}), \
patch('homeassistant.components.hassio.'
'HassIO.command_proxy') as proxy_command, \
patch('homeassistant.components.hassio._create_response') as mresp:
proxy_command.return_value = mock_coro(response)
mresp.return_value = 'response'
resp = yield from hassio_client.post('/api/hassio/addons/beer/install')
# Check we got right response
assert resp.status == 200
assert data == b"That is a test log"
body = yield from resp.text()
assert body == 'response'
assert proxy_command.mock_calls[0][1][0] == 'addons/beer/install'
# Check we forwarded command
assert len(mresp.mock_calls) == 1
assert mresp.mock_calls[0][1] == (response, 'data')
@asyncio.coroutine
def test_forward_addon_log_path(hassio_client):
"""Test fetching addon log path."""
response = MagicMock()
response.read.return_value = mock_coro('data')
with patch.dict(ho.ADDON_REST_COMMANDS, {'logs': ['GET']}), \
patch('homeassistant.components.hassio.'
'HassIO.command_proxy') as proxy_command, \
patch('homeassistant.components.hassio.'
'_create_response_log') as mresp:
proxy_command.return_value = mock_coro(response)
mresp.return_value = 'response'
resp = yield from hassio_client.get('/api/hassio/addons/beer/logs')
# Check we got right response
assert resp.status == 200
body = yield from resp.text()
assert body == 'response'
assert proxy_command.mock_calls[0][1][0] == 'addons/beer/logs'
# Check we forwarded command
assert len(mresp.mock_calls) == 1
assert mresp.mock_calls[0][1] == (response, 'data')
@asyncio.coroutine
def test_bad_request_when_wrong_addon_url(hassio_client):
"""Test we cannot mess with addon url."""
resp = yield from hassio_client.get('/api/hassio/addons/../../info')
assert resp.status == 404
resp = yield from hassio_client.get('/api/hassio/addons/info')
assert resp.status == 404
@asyncio.coroutine
def test_bad_gateway_when_cannot_find_supervisor(hassio_client):
"""Test we get a bad gateway error if we can't find supervisor."""
with patch('homeassistant.components.hassio.async_timeout.timeout',
side_effect=asyncio.TimeoutError):
resp = yield from hassio_client.get('/api/hassio/addons/test/info')
assert resp.status == 502