Fix tests (#7659)
* Remove global hass * Http.auth test no longer spin up server * Remove server usage from http.ban test * Remove setupModule from test device_sun_light_trigger * Update common.py
This commit is contained in:
Paulus Schoutsen
GitHub
parent
5aa72562a7
commit
d369d70ca5
6 changed files with 499 additions and 564 deletions
|
|
@ -150,14 +150,14 @@ def async_setup(hass: HomeAssistantType, config: ConfigType):
|
|||
scanner = yield from platform.async_get_scanner(
|
||||
hass, {DOMAIN: p_config})
|
||||
elif hasattr(platform, 'get_scanner'):
|
||||
scanner = yield from hass.loop.run_in_executor(
|
||||
None, platform.get_scanner, hass, {DOMAIN: p_config})
|
||||
scanner = yield from hass.async_add_job(
|
||||
platform.get_scanner, hass, {DOMAIN: p_config})
|
||||
elif hasattr(platform, 'async_setup_scanner'):
|
||||
setup = yield from platform.async_setup_scanner(
|
||||
hass, p_config, tracker.async_see, disc_info)
|
||||
elif hasattr(platform, 'setup_scanner'):
|
||||
setup = yield from hass.loop.run_in_executor(
|
||||
None, platform.setup_scanner, hass, p_config, tracker.see,
|
||||
setup = yield from hass.async_add_job(
|
||||
platform.setup_scanner, hass, p_config, tracker.see,
|
||||
disc_info)
|
||||
else:
|
||||
raise HomeAssistantError("Invalid device_tracker platform.")
|
||||
|
|
@ -209,8 +209,8 @@ def async_setup(hass: HomeAssistantType, config: ConfigType):
|
|||
ATTR_GPS, ATTR_GPS_ACCURACY, ATTR_BATTERY, ATTR_ATTRIBUTES)}
|
||||
yield from tracker.async_see(**args)
|
||||
|
||||
descriptions = yield from hass.loop.run_in_executor(
|
||||
None, load_yaml_config_file,
|
||||
descriptions = yield from hass.async_add_job(
|
||||
load_yaml_config_file,
|
||||
os.path.join(os.path.dirname(__file__), 'services.yaml')
|
||||
)
|
||||
hass.services.async_register(
|
||||
|
|
@ -322,8 +322,8 @@ class DeviceTracker(object):
|
|||
This method is a coroutine.
|
||||
"""
|
||||
with (yield from self._is_updating):
|
||||
yield from self.hass.loop.run_in_executor(
|
||||
None, update_config, self.hass.config.path(YAML_DEVICES),
|
||||
yield from self.hass.async_add_job(
|
||||
update_config, self.hass.config.path(YAML_DEVICES),
|
||||
dev_id, device)
|
||||
|
||||
@asyncio.coroutine
|
||||
|
|
@ -608,7 +608,7 @@ class DeviceScanner(object):
|
|||
|
||||
This method must be run in the event loop and returns a coroutine.
|
||||
"""
|
||||
return self.hass.loop.run_in_executor(None, self.scan_devices)
|
||||
return self.hass.async_add_job(self.scan_devices)
|
||||
|
||||
def get_device_name(self, mac: str) -> str:
|
||||
"""Get device name from mac."""
|
||||
|
|
@ -619,7 +619,7 @@ class DeviceScanner(object):
|
|||
|
||||
This method must be run in the event loop and returns a coroutine.
|
||||
"""
|
||||
return self.hass.loop.run_in_executor(None, self.get_device_name, mac)
|
||||
return self.hass.async_add_job(self.get_device_name, mac)
|
||||
|
||||
|
||||
def load_config(path: str, hass: HomeAssistantType, consider_home: timedelta):
|
||||
|
|
@ -650,8 +650,8 @@ def async_load_config(path: str, hass: HomeAssistantType,
|
|||
try:
|
||||
result = []
|
||||
try:
|
||||
devices = yield from hass.loop.run_in_executor(
|
||||
None, load_yaml_config_file, path)
|
||||
devices = yield from hass.async_add_job(
|
||||
load_yaml_config_file, path)
|
||||
except HomeAssistantError as err:
|
||||
_LOGGER.error("Unable to load %s: %s", path, str(err))
|
||||
return []
|
||||
|
|
|
|||
|
|
@ -1,26 +1,19 @@
|
|||
"""The tests for the Home Assistant HTTP component."""
|
||||
# pylint: disable=protected-access
|
||||
import logging
|
||||
import asyncio
|
||||
from ipaddress import ip_address, ip_network
|
||||
from unittest.mock import patch
|
||||
|
||||
import requests
|
||||
import pytest
|
||||
|
||||
from homeassistant import setup, const
|
||||
from homeassistant import const
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.components.http as http
|
||||
from homeassistant.components.http.const import (
|
||||
KEY_TRUSTED_NETWORKS, KEY_USE_X_FORWARDED_FOR, HTTP_HEADER_X_FORWARDED_FOR)
|
||||
|
||||
from tests.common import get_test_instance_port, get_test_home_assistant
|
||||
|
||||
API_PASSWORD = 'test1234'
|
||||
SERVER_PORT = get_test_instance_port()
|
||||
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
|
||||
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
|
||||
HA_HEADERS = {
|
||||
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
|
||||
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
|
||||
}
|
||||
|
||||
# Don't add 127.0.0.1/::1 as trusted, as it may interfere with other test cases
|
||||
TRUSTED_NETWORKS = ['192.0.2.0/24', '2001:DB8:ABCD::/48', '100.64.0.1',
|
||||
'FD01:DB8::1']
|
||||
|
|
@ -28,142 +21,131 @@ TRUSTED_ADDRESSES = ['100.64.0.1', '192.0.2.100', 'FD01:DB8::1',
|
|||
'2001:DB8:ABCD::1']
|
||||
UNTRUSTED_ADDRESSES = ['198.51.100.1', '2001:DB8:FA1::1', '127.0.0.1', '::1']
|
||||
|
||||
hass = None
|
||||
|
||||
|
||||
def _url(path=''):
|
||||
"""Helper method to generate URLs."""
|
||||
return HTTP_BASE_URL + path
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def setUpModule():
|
||||
"""Initialize a Home Assistant server."""
|
||||
global hass
|
||||
|
||||
hass = get_test_home_assistant()
|
||||
|
||||
setup.setup_component(
|
||||
hass, http.DOMAIN, {
|
||||
http.DOMAIN: {
|
||||
http.CONF_API_PASSWORD: API_PASSWORD,
|
||||
http.CONF_SERVER_PORT: SERVER_PORT,
|
||||
}
|
||||
@pytest.fixture
|
||||
def mock_api_client(hass, test_client):
|
||||
"""Start the Hass HTTP component."""
|
||||
hass.loop.run_until_complete(async_setup_component(hass, 'api', {
|
||||
'http': {
|
||||
http.CONF_API_PASSWORD: API_PASSWORD,
|
||||
}
|
||||
)
|
||||
}))
|
||||
return hass.loop.run_until_complete(test_client(hass.http.app))
|
||||
|
||||
setup.setup_component(hass, 'api')
|
||||
|
||||
@pytest.fixture
|
||||
def mock_trusted_networks(hass, mock_api_client):
|
||||
"""Mock trusted networks."""
|
||||
hass.http.app[KEY_TRUSTED_NETWORKS] = [
|
||||
ip_network(trusted_network)
|
||||
for trusted_network in TRUSTED_NETWORKS]
|
||||
|
||||
hass.start()
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_access_denied_without_password(mock_api_client):
|
||||
"""Test access without password."""
|
||||
resp = yield from mock_api_client.get(const.URL_API)
|
||||
assert resp.status == 401
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def tearDownModule():
|
||||
"""Stop the Home Assistant server."""
|
||||
hass.stop()
|
||||
@asyncio.coroutine
|
||||
def test_access_denied_with_wrong_password_in_header(mock_api_client):
|
||||
"""Test access with wrong password."""
|
||||
resp = yield from mock_api_client.get(const.URL_API, headers={
|
||||
const.HTTP_HEADER_HA_AUTH: 'wrongpassword'
|
||||
})
|
||||
assert resp.status == 401
|
||||
|
||||
|
||||
class TestHttp:
|
||||
"""Test HTTP component."""
|
||||
@asyncio.coroutine
|
||||
def test_access_denied_with_x_forwarded_for(hass, mock_api_client,
|
||||
mock_trusted_networks):
|
||||
"""Test access denied through the X-Forwarded-For http header."""
|
||||
hass.http.use_x_forwarded_for = True
|
||||
for remote_addr in UNTRUSTED_ADDRESSES:
|
||||
resp = yield from mock_api_client.get(const.URL_API, headers={
|
||||
HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
|
||||
|
||||
def test_access_denied_without_password(self):
|
||||
"""Test access without password."""
|
||||
req = requests.get(_url(const.URL_API))
|
||||
assert resp.status == 401, \
|
||||
"{} shouldn't be trusted".format(remote_addr)
|
||||
|
||||
assert req.status_code == 401
|
||||
|
||||
def test_access_denied_with_wrong_password_in_header(self):
|
||||
"""Test access with wrong password."""
|
||||
req = requests.get(
|
||||
_url(const.URL_API),
|
||||
headers={const.HTTP_HEADER_HA_AUTH: 'wrongpassword'})
|
||||
@asyncio.coroutine
|
||||
def test_access_denied_with_untrusted_ip(mock_api_client,
|
||||
mock_trusted_networks):
|
||||
"""Test access with an untrusted ip address."""
|
||||
for remote_addr in UNTRUSTED_ADDRESSES:
|
||||
with patch('homeassistant.components.http.'
|
||||
'util.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API, params={'api_password': ''})
|
||||
|
||||
assert req.status_code == 401
|
||||
|
||||
def test_access_denied_with_x_forwarded_for(self, caplog):
|
||||
"""Test access denied through the X-Forwarded-For http header."""
|
||||
hass.http.use_x_forwarded_for = True
|
||||
for remote_addr in UNTRUSTED_ADDRESSES:
|
||||
req = requests.get(_url(const.URL_API), headers={
|
||||
HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
|
||||
|
||||
assert req.status_code == 401, \
|
||||
assert resp.status == 401, \
|
||||
"{} shouldn't be trusted".format(remote_addr)
|
||||
|
||||
def test_access_denied_with_untrusted_ip(self, caplog):
|
||||
"""Test access with an untrusted ip address."""
|
||||
for remote_addr in UNTRUSTED_ADDRESSES:
|
||||
with patch('homeassistant.components.http.'
|
||||
'util.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
req = requests.get(
|
||||
_url(const.URL_API), params={'api_password': ''})
|
||||
|
||||
assert req.status_code == 401, \
|
||||
"{} shouldn't be trusted".format(remote_addr)
|
||||
@asyncio.coroutine
|
||||
def test_access_with_password_in_header(mock_api_client, caplog):
|
||||
"""Test access with password in URL."""
|
||||
# Hide logging from requests package that we use to test logging
|
||||
req = yield from mock_api_client.get(
|
||||
const.URL_API, headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
|
||||
|
||||
def test_access_with_password_in_header(self, caplog):
|
||||
"""Test access with password in URL."""
|
||||
# Hide logging from requests package that we use to test logging
|
||||
caplog.set_level(
|
||||
logging.WARNING, logger='requests.packages.urllib3.connectionpool')
|
||||
assert req.status == 200
|
||||
|
||||
req = requests.get(
|
||||
_url(const.URL_API),
|
||||
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
|
||||
logs = caplog.text
|
||||
|
||||
assert req.status_code == 200
|
||||
assert const.URL_API in logs
|
||||
assert API_PASSWORD not in logs
|
||||
|
||||
logs = caplog.text
|
||||
|
||||
assert const.URL_API in logs
|
||||
assert API_PASSWORD not in logs
|
||||
@asyncio.coroutine
|
||||
def test_access_denied_with_wrong_password_in_url(mock_api_client):
|
||||
"""Test access with wrong password."""
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API, params={'api_password': 'wrongpassword'})
|
||||
|
||||
def test_access_denied_with_wrong_password_in_url(self):
|
||||
"""Test access with wrong password."""
|
||||
req = requests.get(
|
||||
_url(const.URL_API), params={'api_password': 'wrongpassword'})
|
||||
assert resp.status == 401
|
||||
|
||||
assert req.status_code == 401
|
||||
|
||||
def test_access_with_password_in_url(self, caplog):
|
||||
"""Test access with password in URL."""
|
||||
# Hide logging from requests package that we use to test logging
|
||||
caplog.set_level(
|
||||
logging.WARNING, logger='requests.packages.urllib3.connectionpool')
|
||||
@asyncio.coroutine
|
||||
def test_access_with_password_in_url(mock_api_client, caplog):
|
||||
"""Test access with password in URL."""
|
||||
req = yield from mock_api_client.get(
|
||||
const.URL_API, params={'api_password': API_PASSWORD})
|
||||
|
||||
req = requests.get(
|
||||
_url(const.URL_API), params={'api_password': API_PASSWORD})
|
||||
assert req.status == 200
|
||||
|
||||
assert req.status_code == 200
|
||||
logs = caplog.text
|
||||
|
||||
logs = caplog.text
|
||||
assert const.URL_API in logs
|
||||
assert API_PASSWORD not in logs
|
||||
|
||||
assert const.URL_API in logs
|
||||
assert API_PASSWORD not in logs
|
||||
|
||||
def test_access_granted_with_x_forwarded_for(self, caplog):
|
||||
"""Test access denied through the X-Forwarded-For http header."""
|
||||
hass.http.app[KEY_USE_X_FORWARDED_FOR] = True
|
||||
for remote_addr in TRUSTED_ADDRESSES:
|
||||
req = requests.get(_url(const.URL_API), headers={
|
||||
HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
|
||||
@asyncio.coroutine
|
||||
def test_access_granted_with_x_forwarded_for(hass, mock_api_client, caplog,
|
||||
mock_trusted_networks):
|
||||
"""Test access denied through the X-Forwarded-For http header."""
|
||||
hass.http.app[KEY_USE_X_FORWARDED_FOR] = True
|
||||
for remote_addr in TRUSTED_ADDRESSES:
|
||||
resp = yield from mock_api_client.get(const.URL_API, headers={
|
||||
HTTP_HEADER_X_FORWARDED_FOR: remote_addr})
|
||||
|
||||
assert req.status_code == 200, \
|
||||
"{} should be trusted".format(remote_addr)
|
||||
assert resp.status == 200, \
|
||||
"{} should be trusted".format(remote_addr)
|
||||
|
||||
def test_access_granted_with_trusted_ip(self, caplog):
|
||||
"""Test access with trusted addresses."""
|
||||
for remote_addr in TRUSTED_ADDRESSES:
|
||||
with patch('homeassistant.components.http.'
|
||||
'auth.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
req = requests.get(
|
||||
_url(const.URL_API), params={'api_password': ''})
|
||||
|
||||
assert req.status_code == 200, \
|
||||
'{} should be trusted'.format(remote_addr)
|
||||
@asyncio.coroutine
|
||||
def test_access_granted_with_trusted_ip(mock_api_client, caplog,
|
||||
mock_trusted_networks):
|
||||
"""Test access with trusted addresses."""
|
||||
for remote_addr in TRUSTED_ADDRESSES:
|
||||
with patch('homeassistant.components.http.'
|
||||
'auth.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API, params={'api_password': ''})
|
||||
|
||||
assert resp.status == 200, \
|
||||
'{} should be trusted'.format(remote_addr)
|
||||
|
|
|
|||
|
|
@ -1,117 +1,91 @@
|
|||
"""The tests for the Home Assistant HTTP component."""
|
||||
# pylint: disable=protected-access
|
||||
import asyncio
|
||||
from ipaddress import ip_address
|
||||
from unittest.mock import patch, mock_open
|
||||
|
||||
import requests
|
||||
import pytest
|
||||
|
||||
from homeassistant import setup, const
|
||||
from homeassistant import const
|
||||
from homeassistant.setup import async_setup_component
|
||||
import homeassistant.components.http as http
|
||||
from homeassistant.components.http.const import (
|
||||
KEY_BANS_ENABLED, KEY_LOGIN_THRESHOLD, KEY_BANNED_IPS)
|
||||
from homeassistant.components.http.ban import IpBan, IP_BANS_FILE
|
||||
|
||||
from tests.common import get_test_instance_port, get_test_home_assistant
|
||||
|
||||
API_PASSWORD = 'test1234'
|
||||
SERVER_PORT = get_test_instance_port()
|
||||
HTTP_BASE = '127.0.0.1:{}'.format(SERVER_PORT)
|
||||
HTTP_BASE_URL = 'http://{}'.format(HTTP_BASE)
|
||||
HA_HEADERS = {
|
||||
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
|
||||
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
|
||||
}
|
||||
BANNED_IPS = ['200.201.202.203', '100.64.0.2']
|
||||
|
||||
hass = None
|
||||
|
||||
|
||||
def _url(path=''):
|
||||
"""Helper method to generate URLs."""
|
||||
return HTTP_BASE_URL + path
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def setUpModule():
|
||||
"""Initialize a Home Assistant server."""
|
||||
global hass
|
||||
|
||||
hass = get_test_home_assistant()
|
||||
|
||||
setup.setup_component(
|
||||
hass, http.DOMAIN, {
|
||||
http.DOMAIN: {
|
||||
http.CONF_API_PASSWORD: API_PASSWORD,
|
||||
http.CONF_SERVER_PORT: SERVER_PORT,
|
||||
}
|
||||
@pytest.fixture
|
||||
def mock_api_client(hass, test_client):
|
||||
"""Start the Hass HTTP component."""
|
||||
hass.loop.run_until_complete(async_setup_component(hass, 'api', {
|
||||
'http': {
|
||||
http.CONF_API_PASSWORD: API_PASSWORD,
|
||||
}
|
||||
)
|
||||
|
||||
setup.setup_component(hass, 'api')
|
||||
|
||||
}))
|
||||
hass.http.app[KEY_BANNED_IPS] = [IpBan(banned_ip) for banned_ip
|
||||
in BANNED_IPS]
|
||||
hass.start()
|
||||
return hass.loop.run_until_complete(test_client(hass.http.app))
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def tearDownModule():
|
||||
"""Stop the Home Assistant server."""
|
||||
hass.stop()
|
||||
@asyncio.coroutine
|
||||
def test_access_from_banned_ip(hass, mock_api_client):
|
||||
"""Test accessing to server from banned IP. Both trusted and not."""
|
||||
hass.http.app[KEY_BANS_ENABLED] = True
|
||||
for remote_addr in BANNED_IPS:
|
||||
with patch('homeassistant.components.http.'
|
||||
'ban.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API)
|
||||
assert resp.status == 403
|
||||
|
||||
|
||||
class TestHttp:
|
||||
"""Test HTTP component."""
|
||||
@asyncio.coroutine
|
||||
def test_access_from_banned_ip_when_ban_is_off(hass, mock_api_client):
|
||||
"""Test accessing to server from banned IP when feature is off."""
|
||||
hass.http.app[KEY_BANS_ENABLED] = False
|
||||
for remote_addr in BANNED_IPS:
|
||||
with patch('homeassistant.components.http.'
|
||||
'ban.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API,
|
||||
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
|
||||
assert resp.status == 200
|
||||
|
||||
def test_access_from_banned_ip(self):
|
||||
"""Test accessing to server from banned IP. Both trusted and not."""
|
||||
hass.http.app[KEY_BANS_ENABLED] = True
|
||||
for remote_addr in BANNED_IPS:
|
||||
with patch('homeassistant.components.http.'
|
||||
'ban.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
req = requests.get(
|
||||
_url(const.URL_API))
|
||||
assert req.status_code == 403
|
||||
|
||||
def test_access_from_banned_ip_when_ban_is_off(self):
|
||||
"""Test accessing to server from banned IP when feature is off."""
|
||||
hass.http.app[KEY_BANS_ENABLED] = False
|
||||
for remote_addr in BANNED_IPS:
|
||||
with patch('homeassistant.components.http.'
|
||||
'ban.get_real_ip',
|
||||
return_value=ip_address(remote_addr)):
|
||||
req = requests.get(
|
||||
_url(const.URL_API),
|
||||
headers={const.HTTP_HEADER_HA_AUTH: API_PASSWORD})
|
||||
assert req.status_code == 200
|
||||
@asyncio.coroutine
|
||||
def test_ip_bans_file_creation(hass, mock_api_client):
|
||||
"""Testing if banned IP file created."""
|
||||
hass.http.app[KEY_BANS_ENABLED] = True
|
||||
hass.http.app[KEY_LOGIN_THRESHOLD] = 1
|
||||
|
||||
def test_ip_bans_file_creation(self):
|
||||
"""Testing if banned IP file created."""
|
||||
hass.http.app[KEY_BANS_ENABLED] = True
|
||||
hass.http.app[KEY_LOGIN_THRESHOLD] = 1
|
||||
m = mock_open()
|
||||
|
||||
m = mock_open()
|
||||
@asyncio.coroutine
|
||||
def call_server():
|
||||
with patch('homeassistant.components.http.'
|
||||
'ban.get_real_ip',
|
||||
return_value=ip_address("200.201.202.204")):
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API,
|
||||
headers={const.HTTP_HEADER_HA_AUTH: 'Wrong password'})
|
||||
return resp
|
||||
|
||||
def call_server():
|
||||
with patch('homeassistant.components.http.'
|
||||
'ban.get_real_ip',
|
||||
return_value=ip_address("200.201.202.204")):
|
||||
return requests.get(
|
||||
_url(const.URL_API),
|
||||
headers={const.HTTP_HEADER_HA_AUTH: 'Wrong password'})
|
||||
with patch('homeassistant.components.http.ban.open', m, create=True):
|
||||
resp = yield from call_server()
|
||||
assert resp.status == 401
|
||||
assert len(hass.http.app[KEY_BANNED_IPS]) == len(BANNED_IPS)
|
||||
assert m.call_count == 0
|
||||
|
||||
with patch('homeassistant.components.http.ban.open', m, create=True):
|
||||
req = call_server()
|
||||
assert req.status_code == 401
|
||||
assert len(hass.http.app[KEY_BANNED_IPS]) == len(BANNED_IPS)
|
||||
assert m.call_count == 0
|
||||
resp = yield from call_server()
|
||||
assert resp.status == 401
|
||||
assert len(hass.http.app[KEY_BANNED_IPS]) == len(BANNED_IPS) + 1
|
||||
m.assert_called_once_with(hass.config.path(IP_BANS_FILE), 'a')
|
||||
|
||||
req = call_server()
|
||||
assert req.status_code == 401
|
||||
assert len(hass.http.app[KEY_BANNED_IPS]) == len(BANNED_IPS) + 1
|
||||
m.assert_called_once_with(hass.config.path(IP_BANS_FILE), 'a')
|
||||
|
||||
req = call_server()
|
||||
assert req.status_code == 403
|
||||
assert m.call_count == 1
|
||||
resp = yield from call_server()
|
||||
assert resp.status == 403
|
||||
assert m.call_count == 1
|
||||
|
|
|
|||
|
|
@ -1,394 +1,380 @@
|
|||
"""The tests for the Home Assistant API component."""
|
||||
# pylint: disable=protected-access
|
||||
from contextlib import closing
|
||||
import asyncio
|
||||
import json
|
||||
import unittest
|
||||
|
||||
import requests
|
||||
import pytest
|
||||
|
||||
from homeassistant import setup, const
|
||||
from homeassistant import const
|
||||
import homeassistant.core as ha
|
||||
import homeassistant.components.http as http
|
||||
|
||||
from tests.common import get_test_instance_port, get_test_home_assistant
|
||||
|
||||
API_PASSWORD = "test1234"
|
||||
SERVER_PORT = get_test_instance_port()
|
||||
HTTP_BASE_URL = "http://127.0.0.1:{}".format(SERVER_PORT)
|
||||
HA_HEADERS = {
|
||||
const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
|
||||
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
|
||||
}
|
||||
|
||||
hass = None
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
|
||||
def _url(path=""):
|
||||
"""Helper method to generate URLs."""
|
||||
return HTTP_BASE_URL + path
|
||||
@pytest.fixture
|
||||
def mock_api_client(hass, test_client):
|
||||
"""Start the Hass HTTP component."""
|
||||
hass.loop.run_until_complete(async_setup_component(hass, 'api', {}))
|
||||
return hass.loop.run_until_complete(test_client(hass.http.app))
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_list_state_entities(hass, mock_api_client):
|
||||
"""Test if the debug interface allows us to list state entities."""
|
||||
hass.states.async_set('test.entity', 'hello')
|
||||
resp = yield from mock_api_client.get(const.URL_API_STATES)
|
||||
assert resp.status == 200
|
||||
json = yield from resp.json()
|
||||
|
||||
remote_data = [ha.State.from_dict(item) for item in json]
|
||||
assert remote_data == hass.states.async_all()
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_get_state(hass, mock_api_client):
|
||||
"""Test if the debug interface allows us to get a state."""
|
||||
hass.states.async_set('hello.world', 'nice', {
|
||||
'attr': 1,
|
||||
})
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API_STATES_ENTITY.format("hello.world"))
|
||||
assert resp.status == 200
|
||||
json = yield from resp.json()
|
||||
|
||||
data = ha.State.from_dict(json)
|
||||
|
||||
state = hass.states.get("hello.world")
|
||||
|
||||
assert data.state == state.state
|
||||
assert data.last_changed == state.last_changed
|
||||
assert data.attributes == state.attributes
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_get_non_existing_state(hass, mock_api_client):
|
||||
"""Test if the debug interface allows us to get a state."""
|
||||
resp = yield from mock_api_client.get(
|
||||
const.URL_API_STATES_ENTITY.format("does_not_exist"))
|
||||
assert resp.status == 404
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_state_change(hass, mock_api_client):
|
||||
"""Test if we can change the state of an entity that exists."""
|
||||
hass.states.async_set("test.test", "not_to_be_set")
|
||||
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_STATES_ENTITY.format("test.test"),
|
||||
json={"state": "debug_state_change2"})
|
||||
|
||||
assert hass.states.get("test.test").state == "debug_state_change2"
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def setUpModule():
|
||||
"""Initialize a Home Assistant server."""
|
||||
global hass
|
||||
@asyncio.coroutine
|
||||
def test_api_state_change_of_non_existing_entity(hass, mock_api_client):
|
||||
"""Test if changing a state of a non existing entity is possible."""
|
||||
new_state = "debug_state_change"
|
||||
|
||||
hass = get_test_home_assistant()
|
||||
resp = yield from mock_api_client.post(
|
||||
const.URL_API_STATES_ENTITY.format("test_entity.that_does_not_exist"),
|
||||
json={'state': new_state})
|
||||
|
||||
hass.bus.listen('test_event', lambda _: _)
|
||||
hass.states.set('test.test', 'a_state')
|
||||
assert resp.status == 201
|
||||
|
||||
setup.setup_component(
|
||||
hass, http.DOMAIN,
|
||||
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD,
|
||||
http.CONF_SERVER_PORT: SERVER_PORT}})
|
||||
|
||||
setup.setup_component(hass, 'api')
|
||||
|
||||
hass.start()
|
||||
assert hass.states.get("test_entity.that_does_not_exist").state == \
|
||||
new_state
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def tearDownModule():
|
||||
"""Stop the Home Assistant server."""
|
||||
hass.stop()
|
||||
@asyncio.coroutine
|
||||
def test_api_state_change_with_bad_data(hass, mock_api_client):
|
||||
"""Test if API sends appropriate error if we omit state."""
|
||||
resp = yield from mock_api_client.post(
|
||||
const.URL_API_STATES_ENTITY.format("test_entity.that_does_not_exist"),
|
||||
json={})
|
||||
|
||||
assert resp.status == 400
|
||||
|
||||
|
||||
class TestAPI(unittest.TestCase):
|
||||
"""Test the API."""
|
||||
# pylint: disable=invalid-name
|
||||
@asyncio.coroutine
|
||||
def test_api_state_change_push(hass, mock_api_client):
|
||||
"""Test if we can push a change the state of an entity."""
|
||||
hass.states.async_set("test.test", "not_to_be_set")
|
||||
|
||||
def tearDown(self):
|
||||
"""Stop everything that was started."""
|
||||
hass.block_till_done()
|
||||
events = []
|
||||
|
||||
def test_api_list_state_entities(self):
|
||||
"""Test if the debug interface allows us to list state entities."""
|
||||
req = requests.get(_url(const.URL_API_STATES),
|
||||
headers=HA_HEADERS)
|
||||
@ha.callback
|
||||
def event_listener(event):
|
||||
"""Track events."""
|
||||
events.append(event)
|
||||
|
||||
remote_data = [ha.State.from_dict(item) for item in req.json()]
|
||||
hass.bus.async_listen(const.EVENT_STATE_CHANGED, event_listener)
|
||||
|
||||
self.assertEqual(hass.states.all(), remote_data)
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_STATES_ENTITY.format("test.test"),
|
||||
json={"state": "not_to_be_set"})
|
||||
yield from hass.async_block_till_done()
|
||||
assert len(events) == 0
|
||||
|
||||
def test_api_get_state(self):
|
||||
"""Test if the debug interface allows us to get a state."""
|
||||
req = requests.get(
|
||||
_url(const.URL_API_STATES_ENTITY.format("test.test")),
|
||||
headers=HA_HEADERS)
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_STATES_ENTITY.format("test.test"),
|
||||
json={"state": "not_to_be_set", "force_update": True})
|
||||
yield from hass.async_block_till_done()
|
||||
assert len(events) == 1
|
||||
|
||||
data = ha.State.from_dict(req.json())
|
||||
|
||||
state = hass.states.get("test.test")
|
||||
# pylint: disable=invalid-name
|
||||
@asyncio.coroutine
|
||||
def test_api_fire_event_with_no_data(hass, mock_api_client):
|
||||
"""Test if the API allows us to fire an event."""
|
||||
test_value = []
|
||||
|
||||
self.assertEqual(state.state, data.state)
|
||||
self.assertEqual(state.last_changed, data.last_changed)
|
||||
self.assertEqual(state.attributes, data.attributes)
|
||||
@ha.callback
|
||||
def listener(event):
|
||||
"""Helper method that will verify our event got called."""
|
||||
test_value.append(1)
|
||||
|
||||
def test_api_get_non_existing_state(self):
|
||||
"""Test if the debug interface allows us to get a state."""
|
||||
req = requests.get(
|
||||
_url(const.URL_API_STATES_ENTITY.format("does_not_exist")),
|
||||
headers=HA_HEADERS)
|
||||
hass.bus.async_listen_once("test.event_no_data", listener)
|
||||
|
||||
self.assertEqual(404, req.status_code)
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_EVENTS_EVENT.format("test.event_no_data"))
|
||||
yield from hass.async_block_till_done()
|
||||
|
||||
def test_api_state_change(self):
|
||||
"""Test if we can change the state of an entity that exists."""
|
||||
hass.states.set("test.test", "not_to_be_set")
|
||||
assert len(test_value) == 1
|
||||
|
||||
requests.post(_url(const.URL_API_STATES_ENTITY.format("test.test")),
|
||||
data=json.dumps({"state": "debug_state_change2"}),
|
||||
headers=HA_HEADERS)
|
||||
|
||||
self.assertEqual("debug_state_change2",
|
||||
hass.states.get("test.test").state)
|
||||
# pylint: disable=invalid-name
|
||||
@asyncio.coroutine
|
||||
def test_api_fire_event_with_data(hass, mock_api_client):
|
||||
"""Test if the API allows us to fire an event."""
|
||||
test_value = []
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def test_api_state_change_of_non_existing_entity(self):
|
||||
"""Test if changing a state of a non existing entity is possible."""
|
||||
new_state = "debug_state_change"
|
||||
@ha.callback
|
||||
def listener(event):
|
||||
"""Helper method that will verify that our event got called.
|
||||
|
||||
req = requests.post(
|
||||
_url(const.URL_API_STATES_ENTITY.format(
|
||||
"test_entity.that_does_not_exist")),
|
||||
data=json.dumps({'state': new_state}),
|
||||
headers=HA_HEADERS)
|
||||
|
||||
cur_state = (hass.states.
|
||||
get("test_entity.that_does_not_exist").state)
|
||||
|
||||
self.assertEqual(201, req.status_code)
|
||||
self.assertEqual(cur_state, new_state)
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def test_api_state_change_with_bad_data(self):
|
||||
"""Test if API sends appropriate error if we omit state."""
|
||||
req = requests.post(
|
||||
_url(const.URL_API_STATES_ENTITY.format(
|
||||
"test_entity.that_does_not_exist")),
|
||||
data=json.dumps({}),
|
||||
headers=HA_HEADERS)
|
||||
|
||||
self.assertEqual(400, req.status_code)
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def test_api_state_change_push(self):
|
||||
"""Test if we can push a change the state of an entity."""
|
||||
hass.states.set("test.test", "not_to_be_set")
|
||||
|
||||
events = []
|
||||
hass.bus.listen(const.EVENT_STATE_CHANGED,
|
||||
lambda ev: events.append(ev))
|
||||
|
||||
requests.post(_url(const.URL_API_STATES_ENTITY.format("test.test")),
|
||||
data=json.dumps({"state": "not_to_be_set"}),
|
||||
headers=HA_HEADERS)
|
||||
hass.block_till_done()
|
||||
self.assertEqual(0, len(events))
|
||||
|
||||
requests.post(_url(const.URL_API_STATES_ENTITY.format("test.test")),
|
||||
data=json.dumps({"state": "not_to_be_set",
|
||||
"force_update": True}),
|
||||
headers=HA_HEADERS)
|
||||
hass.block_till_done()
|
||||
self.assertEqual(1, len(events))
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def test_api_fire_event_with_no_data(self):
|
||||
"""Test if the API allows us to fire an event."""
|
||||
test_value = []
|
||||
|
||||
def listener(event):
|
||||
"""Helper method that will verify our event got called."""
|
||||
Also test if our data came through.
|
||||
"""
|
||||
if "test" in event.data:
|
||||
test_value.append(1)
|
||||
|
||||
hass.bus.listen_once("test.event_no_data", listener)
|
||||
hass.bus.async_listen_once("test_event_with_data", listener)
|
||||
|
||||
requests.post(
|
||||
_url(const.URL_API_EVENTS_EVENT.format("test.event_no_data")),
|
||||
headers=HA_HEADERS)
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_EVENTS_EVENT.format("test_event_with_data"),
|
||||
json={"test": 1})
|
||||
|
||||
hass.block_till_done()
|
||||
yield from hass.async_block_till_done()
|
||||
|
||||
self.assertEqual(1, len(test_value))
|
||||
assert len(test_value) == 1
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def test_api_fire_event_with_data(self):
|
||||
"""Test if the API allows us to fire an event."""
|
||||
test_value = []
|
||||
|
||||
def listener(event):
|
||||
"""Helper method that will verify that our event got called.
|
||||
# pylint: disable=invalid-name
|
||||
@asyncio.coroutine
|
||||
def test_api_fire_event_with_invalid_json(hass, mock_api_client):
|
||||
"""Test if the API allows us to fire an event."""
|
||||
test_value = []
|
||||
|
||||
Also test if our data came through.
|
||||
"""
|
||||
if "test" in event.data:
|
||||
test_value.append(1)
|
||||
@ha.callback
|
||||
def listener(event):
|
||||
"""Helper method that will verify our event got called."""
|
||||
test_value.append(1)
|
||||
|
||||
hass.bus.listen_once("test_event_with_data", listener)
|
||||
hass.bus.async_listen_once("test_event_bad_data", listener)
|
||||
|
||||
requests.post(
|
||||
_url(const.URL_API_EVENTS_EVENT.format("test_event_with_data")),
|
||||
data=json.dumps({"test": 1}),
|
||||
headers=HA_HEADERS)
|
||||
resp = yield from mock_api_client.post(
|
||||
const.URL_API_EVENTS_EVENT.format("test_event_bad_data"),
|
||||
data=json.dumps('not an object'))
|
||||
|
||||
hass.block_till_done()
|
||||
yield from hass.async_block_till_done()
|
||||
|
||||
self.assertEqual(1, len(test_value))
|
||||
assert resp.status == 400
|
||||
assert len(test_value) == 0
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def test_api_fire_event_with_invalid_json(self):
|
||||
"""Test if the API allows us to fire an event."""
|
||||
test_value = []
|
||||
# Try now with valid but unusable JSON
|
||||
resp = yield from mock_api_client.post(
|
||||
const.URL_API_EVENTS_EVENT.format("test_event_bad_data"),
|
||||
data=json.dumps([1, 2, 3]))
|
||||
|
||||
def listener(event):
|
||||
"""Helper method that will verify our event got called."""
|
||||
yield from hass.async_block_till_done()
|
||||
|
||||
assert resp.status == 400
|
||||
assert len(test_value) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_get_config(hass, mock_api_client):
|
||||
"""Test the return of the configuration."""
|
||||
resp = yield from mock_api_client.get(const.URL_API_CONFIG)
|
||||
result = yield from resp.json()
|
||||
if 'components' in result:
|
||||
result['components'] = set(result['components'])
|
||||
|
||||
assert hass.config.as_dict() == result
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_get_components(hass, mock_api_client):
|
||||
"""Test the return of the components."""
|
||||
resp = yield from mock_api_client.get(const.URL_API_COMPONENTS)
|
||||
result = yield from resp.json()
|
||||
assert set(result) == hass.config.components
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_get_event_listeners(hass, mock_api_client):
|
||||
"""Test if we can get the list of events being listened for."""
|
||||
resp = yield from mock_api_client.get(const.URL_API_EVENTS)
|
||||
data = yield from resp.json()
|
||||
|
||||
local = hass.bus.async_listeners()
|
||||
|
||||
for event in data:
|
||||
assert local.pop(event["event"]) == event["listener_count"]
|
||||
|
||||
assert len(local) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_get_services(hass, mock_api_client):
|
||||
"""Test if we can get a dict describing current services."""
|
||||
resp = yield from mock_api_client.get(const.URL_API_SERVICES)
|
||||
data = yield from resp.json()
|
||||
local_services = hass.services.async_services()
|
||||
|
||||
for serv_domain in data:
|
||||
local = local_services.pop(serv_domain["domain"])
|
||||
|
||||
assert serv_domain["services"] == local
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_call_service_no_data(hass, mock_api_client):
|
||||
"""Test if the API allows us to call a service."""
|
||||
test_value = []
|
||||
|
||||
@ha.callback
|
||||
def listener(service_call):
|
||||
"""Helper method that will verify that our service got called."""
|
||||
test_value.append(1)
|
||||
|
||||
hass.services.async_register("test_domain", "test_service", listener)
|
||||
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_SERVICES_SERVICE.format(
|
||||
"test_domain", "test_service"))
|
||||
yield from hass.async_block_till_done()
|
||||
assert len(test_value) == 1
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_api_call_service_with_data(hass, mock_api_client):
|
||||
"""Test if the API allows us to call a service."""
|
||||
test_value = []
|
||||
|
||||
@ha.callback
|
||||
def listener(service_call):
|
||||
"""Helper method that will verify that our service got called.
|
||||
|
||||
Also test if our data came through.
|
||||
"""
|
||||
if "test" in service_call.data:
|
||||
test_value.append(1)
|
||||
|
||||
hass.bus.listen_once("test_event_bad_data", listener)
|
||||
hass.services.async_register("test_domain", "test_service", listener)
|
||||
|
||||
req = requests.post(
|
||||
_url(const.URL_API_EVENTS_EVENT.format("test_event_bad_data")),
|
||||
data=json.dumps('not an object'),
|
||||
headers=HA_HEADERS)
|
||||
yield from mock_api_client.post(
|
||||
const.URL_API_SERVICES_SERVICE.format("test_domain", "test_service"),
|
||||
json={"test": 1})
|
||||
|
||||
hass.block_till_done()
|
||||
yield from hass.async_block_till_done()
|
||||
assert len(test_value) == 1
|
||||
|
||||
self.assertEqual(400, req.status_code)
|
||||
self.assertEqual(0, len(test_value))
|
||||
|
||||
# Try now with valid but unusable JSON
|
||||
req = requests.post(
|
||||
_url(const.URL_API_EVENTS_EVENT.format("test_event_bad_data")),
|
||||
data=json.dumps([1, 2, 3]),
|
||||
headers=HA_HEADERS)
|
||||
@asyncio.coroutine
|
||||
def test_api_template(hass, mock_api_client):
|
||||
"""Test the template API."""
|
||||
hass.states.async_set('sensor.temperature', 10)
|
||||
|
||||
hass.block_till_done()
|
||||
resp = yield from mock_api_client.post(
|
||||
const.URL_API_TEMPLATE,
|
||||
json={"template": '{{ states.sensor.temperature.state }}'})
|
||||
|
||||
self.assertEqual(400, req.status_code)
|
||||
self.assertEqual(0, len(test_value))
|
||||
body = yield from resp.text()
|
||||
|
||||
def test_api_get_config(self):
|
||||
"""Test the return of the configuration."""
|
||||
req = requests.get(_url(const.URL_API_CONFIG),
|
||||
headers=HA_HEADERS)
|
||||
result = req.json()
|
||||
if 'components' in result:
|
||||
result['components'] = set(result['components'])
|
||||
assert body == '10'
|
||||
|
||||
self.assertEqual(hass.config.as_dict(), result)
|
||||
|
||||
def test_api_get_components(self):
|
||||
"""Test the return of the components."""
|
||||
req = requests.get(_url(const.URL_API_COMPONENTS),
|
||||
headers=HA_HEADERS)
|
||||
self.assertEqual(hass.config.components, set(req.json()))
|
||||
@asyncio.coroutine
|
||||
def test_api_template_error(hass, mock_api_client):
|
||||
"""Test the template API."""
|
||||
hass.states.async_set('sensor.temperature', 10)
|
||||
|
||||
def test_api_get_event_listeners(self):
|
||||
"""Test if we can get the list of events being listened for."""
|
||||
req = requests.get(_url(const.URL_API_EVENTS),
|
||||
headers=HA_HEADERS)
|
||||
resp = yield from mock_api_client.post(
|
||||
const.URL_API_TEMPLATE,
|
||||
json={"template": '{{ states.sensor.temperature.state'})
|
||||
|
||||
local = hass.bus.listeners
|
||||
assert resp.status == 400
|
||||
|
||||
for event in req.json():
|
||||
self.assertEqual(event["listener_count"],
|
||||
local.pop(event["event"]))
|
||||
|
||||
self.assertEqual(0, len(local))
|
||||
@asyncio.coroutine
|
||||
def test_stream(hass, mock_api_client):
|
||||
"""Test the stream."""
|
||||
listen_count = _listen_count(hass)
|
||||
|
||||
def test_api_get_services(self):
|
||||
"""Test if we can get a dict describing current services."""
|
||||
req = requests.get(_url(const.URL_API_SERVICES),
|
||||
headers=HA_HEADERS)
|
||||
resp = yield from mock_api_client.get(const.URL_API_STREAM)
|
||||
assert resp.status == 200
|
||||
assert listen_count + 1 == _listen_count(hass)
|
||||
|
||||
local_services = hass.services.services
|
||||
hass.bus.async_fire('test_event')
|
||||
|
||||
for serv_domain in req.json():
|
||||
local = local_services.pop(serv_domain["domain"])
|
||||
data = yield from _stream_next_event(resp.content)
|
||||
|
||||
self.assertEqual(local, serv_domain["services"])
|
||||
assert data['event_type'] == 'test_event'
|
||||
|
||||
def test_api_call_service_no_data(self):
|
||||
"""Test if the API allows us to call a service."""
|
||||
test_value = []
|
||||
|
||||
@ha.callback
|
||||
def listener(service_call):
|
||||
"""Helper method that will verify that our service got called."""
|
||||
test_value.append(1)
|
||||
@asyncio.coroutine
|
||||
def test_stream_with_restricted(hass, mock_api_client):
|
||||
"""Test the stream with restrictions."""
|
||||
listen_count = _listen_count(hass)
|
||||
|
||||
hass.services.register("test_domain", "test_service", listener)
|
||||
resp = yield from mock_api_client.get(
|
||||
'{}?restrict=test_event1,test_event3'.format(const.URL_API_STREAM))
|
||||
assert resp.status == 200
|
||||
assert listen_count + 1 == _listen_count(hass)
|
||||
|
||||
requests.post(
|
||||
_url(const.URL_API_SERVICES_SERVICE.format(
|
||||
"test_domain", "test_service")),
|
||||
headers=HA_HEADERS)
|
||||
hass.bus.async_fire('test_event1')
|
||||
data = yield from _stream_next_event(resp.content)
|
||||
assert data['event_type'] == 'test_event1'
|
||||
|
||||
hass.block_till_done()
|
||||
hass.bus.async_fire('test_event2')
|
||||
hass.bus.async_fire('test_event3')
|
||||
data = yield from _stream_next_event(resp.content)
|
||||
assert data['event_type'] == 'test_event3'
|
||||
|
||||
self.assertEqual(1, len(test_value))
|
||||
|
||||
def test_api_call_service_with_data(self):
|
||||
"""Test if the API allows us to call a service."""
|
||||
test_value = []
|
||||
@asyncio.coroutine
|
||||
def _stream_next_event(stream):
|
||||
"""Read the stream for next event while ignoring ping."""
|
||||
while True:
|
||||
last_new_line = False
|
||||
data = b''
|
||||
|
||||
@ha.callback
|
||||
def listener(service_call):
|
||||
"""Helper method that will verify that our service got called.
|
||||
|
||||
Also test if our data came through.
|
||||
"""
|
||||
if "test" in service_call.data:
|
||||
test_value.append(1)
|
||||
|
||||
hass.services.register("test_domain", "test_service", listener)
|
||||
|
||||
requests.post(
|
||||
_url(const.URL_API_SERVICES_SERVICE.format(
|
||||
"test_domain", "test_service")),
|
||||
data=json.dumps({"test": 1}),
|
||||
headers=HA_HEADERS)
|
||||
|
||||
hass.block_till_done()
|
||||
|
||||
self.assertEqual(1, len(test_value))
|
||||
|
||||
def test_api_template(self):
|
||||
"""Test the template API."""
|
||||
hass.states.set('sensor.temperature', 10)
|
||||
|
||||
req = requests.post(
|
||||
_url(const.URL_API_TEMPLATE),
|
||||
json={"template": '{{ states.sensor.temperature.state }}'},
|
||||
headers=HA_HEADERS)
|
||||
|
||||
self.assertEqual('10', req.text)
|
||||
|
||||
def test_api_template_error(self):
|
||||
"""Test the template API."""
|
||||
hass.states.set('sensor.temperature', 10)
|
||||
|
||||
req = requests.post(
|
||||
_url(const.URL_API_TEMPLATE),
|
||||
data=json.dumps({"template":
|
||||
'{{ states.sensor.temperature.state'}),
|
||||
headers=HA_HEADERS)
|
||||
|
||||
self.assertEqual(400, req.status_code)
|
||||
|
||||
def test_stream(self):
|
||||
"""Test the stream."""
|
||||
listen_count = self._listen_count()
|
||||
with closing(requests.get(_url(const.URL_API_STREAM), timeout=3,
|
||||
stream=True, headers=HA_HEADERS)) as req:
|
||||
stream = req.iter_content(1)
|
||||
self.assertEqual(listen_count + 1, self._listen_count())
|
||||
|
||||
hass.bus.fire('test_event')
|
||||
|
||||
data = self._stream_next_event(stream)
|
||||
|
||||
self.assertEqual('test_event', data['event_type'])
|
||||
|
||||
def test_stream_with_restricted(self):
|
||||
"""Test the stream with restrictions."""
|
||||
listen_count = self._listen_count()
|
||||
url = _url('{}?restrict=test_event1,test_event3'.format(
|
||||
const.URL_API_STREAM))
|
||||
with closing(requests.get(url, stream=True, timeout=3,
|
||||
headers=HA_HEADERS)) as req:
|
||||
stream = req.iter_content(1)
|
||||
self.assertEqual(listen_count + 1, self._listen_count())
|
||||
|
||||
hass.bus.fire('test_event1')
|
||||
data = self._stream_next_event(stream)
|
||||
self.assertEqual('test_event1', data['event_type'])
|
||||
|
||||
hass.bus.fire('test_event2')
|
||||
hass.bus.fire('test_event3')
|
||||
|
||||
data = self._stream_next_event(stream)
|
||||
self.assertEqual('test_event3', data['event_type'])
|
||||
|
||||
def _stream_next_event(self, stream):
|
||||
"""Read the stream for next event while ignoring ping."""
|
||||
while True:
|
||||
data = b''
|
||||
last_new_line = False
|
||||
for dat in stream:
|
||||
if dat == b'\n' and last_new_line:
|
||||
break
|
||||
data += dat
|
||||
last_new_line = dat == b'\n'
|
||||
|
||||
conv = data.decode('utf-8').strip()[6:]
|
||||
|
||||
if conv != 'ping':
|
||||
dat = yield from stream.read(1)
|
||||
if dat == b'\n' and last_new_line:
|
||||
break
|
||||
data += dat
|
||||
last_new_line = dat == b'\n'
|
||||
|
||||
return json.loads(conv)
|
||||
conv = data.decode('utf-8').strip()[6:]
|
||||
|
||||
def _listen_count(self):
|
||||
"""Return number of event listeners."""
|
||||
return sum(hass.bus.listeners.values())
|
||||
if conv != 'ping':
|
||||
break
|
||||
return json.loads(conv)
|
||||
|
||||
|
||||
def _listen_count(hass):
|
||||
"""Return number of event listeners."""
|
||||
return sum(hass.bus.async_listeners().values())
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
"""The tests device sun light trigger component."""
|
||||
# pylint: disable=protected-access
|
||||
from datetime import datetime
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
|
|
@ -12,32 +11,7 @@ from homeassistant.components import (
|
|||
device_tracker, light, device_sun_light_trigger)
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from tests.common import (
|
||||
get_test_config_dir, get_test_home_assistant, fire_time_changed)
|
||||
|
||||
|
||||
KNOWN_DEV_YAML_PATH = os.path.join(get_test_config_dir(),
|
||||
device_tracker.YAML_DEVICES)
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def setUpModule():
|
||||
"""Write a device tracker known devices file to be used."""
|
||||
device_tracker.update_config(
|
||||
KNOWN_DEV_YAML_PATH, 'device_1', device_tracker.Device(
|
||||
None, None, True, 'device_1', 'DEV1',
|
||||
picture='http://example.com/dev1.jpg'))
|
||||
|
||||
device_tracker.update_config(
|
||||
KNOWN_DEV_YAML_PATH, 'device_2', device_tracker.Device(
|
||||
None, None, True, 'device_2', 'DEV2',
|
||||
picture='http://example.com/dev2.jpg'))
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
def tearDownModule():
|
||||
"""Remove device tracker known devices file."""
|
||||
os.remove(KNOWN_DEV_YAML_PATH)
|
||||
from tests.common import get_test_home_assistant, fire_time_changed
|
||||
|
||||
|
||||
class TestDeviceSunLightTrigger(unittest.TestCase):
|
||||
|
|
@ -55,9 +29,28 @@ class TestDeviceSunLightTrigger(unittest.TestCase):
|
|||
|
||||
loader.get_component('light.test').init()
|
||||
|
||||
self.assertTrue(setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {CONF_PLATFORM: 'test'}
|
||||
}))
|
||||
with patch(
|
||||
'homeassistant.components.device_tracker.load_yaml_config_file',
|
||||
return_value={
|
||||
'device_1': {
|
||||
'hide_if_away': False,
|
||||
'mac': 'DEV1',
|
||||
'name': 'Unnamed Device',
|
||||
'picture': 'http://example.com/dev1.jpg',
|
||||
'track': True,
|
||||
'vendor': None
|
||||
},
|
||||
'device_2': {
|
||||
'hide_if_away': False,
|
||||
'mac': 'DEV2',
|
||||
'name': 'Unnamed Device',
|
||||
'picture': 'http://example.com/dev2.jpg',
|
||||
'track': True,
|
||||
'vendor': None}
|
||||
}):
|
||||
self.assertTrue(setup_component(self.hass, device_tracker.DOMAIN, {
|
||||
device_tracker.DOMAIN: {CONF_PLATFORM: 'test'}
|
||||
}))
|
||||
|
||||
self.assertTrue(setup_component(self.hass, light.DOMAIN, {
|
||||
light.DOMAIN: {CONF_PLATFORM: 'test'}
|
||||
|
|
|
|||
|
|
@ -8,10 +8,10 @@ from homeassistant.setup import async_setup_component
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_http_client(loop, hass, test_client):
|
||||
def mock_http_client(hass, test_client):
|
||||
"""Start the Hass HTTP component."""
|
||||
loop.run_until_complete(async_setup_component(hass, 'frontend', {}))
|
||||
return loop.run_until_complete(test_client(hass.http.app))
|
||||
hass.loop.run_until_complete(async_setup_component(hass, 'frontend', {}))
|
||||
return hass.loop.run_until_complete(test_client(hass.http.app))
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue