jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Convert rest sensors to async using httpx (#41973)

Browse Source
This commit is contained in:
J. Nick Koston 2020-10-16 19:21:13 -05:00 committed by GitHub
parent 39adf14079
commit ad6ce5fa83
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 1094 additions and 970 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -721,7 +721,6 @@ omit =
homeassistant/components/repetier/__init__.py
homeassistant/components/repetier/sensor.py
homeassistant/components/remote_rpi_gpio/*
homeassistant/components/rest/binary_sensor.py
homeassistant/components/rest/notify.py
homeassistant/components/rest/switch.py
homeassistant/components/ring/camera.py

14
homeassistant/components/pvoutput/sensor.py
View File

@ -43,7 +43,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the PVOutput sensor."""
name = config.get(CONF_NAME)
api_key = config.get(CONF_API_KEY)
@ -54,13 +54,13 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
headers = {"X-Pvoutput-Apikey": api_key, "X-Pvoutput-SystemId": system_id}
rest = RestData(method, _ENDPOINT, auth, headers, payload, verify_ssl)
rest.update()
await rest.async_update()
if rest.data is None:
_LOGGER.error("Unable to fetch data from PVOutput")
return False
add_entities([PvoutputSensor(rest, name)], True)
async_add_entities([PvoutputSensor(rest, name)], True)
class PvoutputSensor(Entity):
@ -112,11 +112,15 @@ class PvoutputSensor(Entity):
ATTR_VOLTAGE: self.pvcoutput.voltage,
}
def update(self):
async def async_update(self):
"""Get the latest data from the PVOutput API and updates the state."""
try:
self.rest.update()
await self.rest.async_update()
self.pvcoutput = self.status._make(self.rest.data.split(","))
except TypeError:
self.pvcoutput = None
_LOGGER.error("Unable to fetch data from PVOutput. %s", self.rest.data)
async def async_will_remove_from_hass(self):
"""Shutdown the session."""
await self.rest.async_remove()

27
homeassistant/components/rest/binary_sensor.py
View File

@ -1,7 +1,7 @@
"""Support for RESTful binary sensors."""
import logging
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import httpx
import voluptuous as vol
from homeassistant.components.binary_sensor import (
@ -29,7 +29,7 @@ from homeassistant.const import (
)
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.reload import async_setup_reload_service
from . import DOMAIN, PLATFORMS
from .sensor import RestData
@ -68,10 +68,10 @@ PLATFORM_SCHEMA = vol.All(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the REST binary sensor."""
setup_reload_service(hass, DOMAIN, PLATFORMS)
await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
name = config.get(CONF_NAME)
resource = config.get(CONF_RESOURCE)
@ -96,18 +96,18 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
if username and password:
if config.get(CONF_AUTHENTICATION) == HTTP_DIGEST_AUTHENTICATION:
auth = HTTPDigestAuth(username, password)
auth = httpx.DigestAuth(username, password)
else:
auth = HTTPBasicAuth(username, password)
auth = (username, password)
else:
auth = None
rest = RestData(method, resource, auth, headers, payload, verify_ssl, timeout)
rest.update()
await rest.async_update()
if rest.data is None:
raise PlatformNotReady
add_entities(
async_add_entities(
[
RestBinarySensor(
hass,
@ -118,7 +118,8 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
force_update,
resource_template,
)
]
],
True,
)
@ -186,9 +187,13 @@ class RestBinarySensor(BinarySensorEntity):
"""Force update."""
return self._force_update
def update(self):
async def async_will_remove_from_hass(self):
"""Shutdown the session."""
await self.rest.async_remove()
async def async_update(self):
"""Get the latest data from REST API and updates the state."""
if self._resource_template is not None:
self.rest.set_url(self._resource_template.render())
self.rest.update()
await self.rest.async_update()

2
homeassistant/components/rest/manifest.json
View File

@ -2,6 +2,6 @@
"domain": "rest",
"name": "RESTful",
"documentation": "https://www.home-assistant.io/integrations/rest",
"requirements": ["jsonpath==0.82", "xmltodict==0.12.0"],
"requirements": ["jsonpath==0.82", "xmltodict==0.12.0", "httpx==0.16.1"],
"codeowners": []
}

62
homeassistant/components/rest/sensor.py
View File

@ -3,10 +3,8 @@ import json
import logging
from xml.parsers.expat import ExpatError
import httpx
from jsonpath import jsonpath
import requests
from requests import Session
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import voluptuous as vol
import xmltodict
@ -33,7 +31,7 @@ from homeassistant.const import (
from homeassistant.exceptions import PlatformNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.reload import async_setup_reload_service
from . import DOMAIN, PLATFORMS
@ -79,9 +77,9 @@ PLATFORM_SCHEMA = vol.All(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the RESTful sensor."""
setup_reload_service(hass, DOMAIN, PLATFORMS)
await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
name = config.get(CONF_NAME)
resource = config.get(CONF_RESOURCE)
@ -109,19 +107,20 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
if username and password:
if config.get(CONF_AUTHENTICATION) == HTTP_DIGEST_AUTHENTICATION:
auth = HTTPDigestAuth(username, password)
auth = httpx.DigestAuth(username, password)
else:
auth = HTTPBasicAuth(username, password)
auth = (username, password)
else:
auth = None
rest = RestData(method, resource, auth, headers, payload, verify_ssl, timeout)
rest.update()
await rest.async_update()
if rest.data is None:
raise PlatformNotReady
# Must update the sensor now (including fetching the rest resource) to
# ensure it's updating its state.
add_entities(
async_add_entities(
[
RestSensor(
hass,
@ -200,12 +199,13 @@ class RestSensor(Entity):
"""Force update."""
return self._force_update
def update(self):
async def async_update(self):
"""Get the latest data from REST API and update the state."""
if self._resource_template is not None:
self.rest.set_url(self._resource_template.render())
self.rest.update()
await self.rest.async_update()
value = self.rest.data
_LOGGER.debug("Data fetched from resource: %s", value)
if self.rest.headers is not None:
@ -250,13 +250,21 @@ class RestSensor(Entity):
except ValueError:
_LOGGER.warning("REST result could not be parsed as JSON")
_LOGGER.debug("Erroneous JSON: %s", value)
else:
_LOGGER.warning("Empty reply found when expecting JSON data")
if value is not None and self._value_template is not None:
value = self._value_template.render_with_possible_json_value(value, None)
value = self._value_template.async_render_with_possible_json_value(
value, None
)
self._state = value
async def async_will_remove_from_hass(self):
"""Shutdown the session."""
await self.rest.async_remove()
@property
def device_state_attributes(self):
"""Return the state attributes."""
@ -267,7 +275,14 @@ class RestData:
"""Class for handling the data retrieval."""
def __init__(
self, method, resource, auth, headers, data, verify_ssl, timeout=DEFAULT_TIMEOUT
self,
method,
resource,
auth,
headers,
data,
verify_ssl,
timeout=DEFAULT_TIMEOUT,
):
"""Initialize the data object."""
self._method = method
@ -275,36 +290,39 @@ class RestData:
self._auth = auth
self._headers = headers
self._request_data = data
self._verify_ssl = verify_ssl
self._timeout = timeout
self._http_session = Session()
self._verify_ssl = verify_ssl
self._async_client = None
self.data = None
self.headers = None
def __del__(self):
async def async_remove(self):
"""Destroy the http session on destroy."""
self._http_session.close()
if self._async_client:
await self._async_client.aclose()
def set_url(self, url):
"""Set url."""
self._resource = url
def update(self):
async def async_update(self):
"""Get the latest data from REST service with provided method."""
if not self._async_client:
self._async_client = httpx.AsyncClient(verify=self._verify_ssl)
_LOGGER.debug("Updating from %s", self._resource)
try:
response = self._http_session.request(
response = await self._async_client.request(
self._method,
self._resource,
headers=self._headers,
auth=self._auth,
data=self._request_data,
timeout=self._timeout,
verify=self._verify_ssl,
)
self.data = response.text
self.headers = response.headers
except requests.exceptions.RequestException as ex:
except httpx.RequestError as ex:
_LOGGER.error("Error fetching data: %s failed with %s", self._resource, ex)
self.data = None
self.headers = None

16
homeassistant/components/scrape/sensor.py
View File

@ -53,7 +53,7 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Web scrape sensor."""
name = config.get(CONF_NAME)
resource = config.get(CONF_RESOURCE)
@ -79,12 +79,12 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
else:
auth = None
rest = RestData(method, resource, auth, headers, payload, verify_ssl)
rest.update()
await rest.async_update()
if rest.data is None:
raise PlatformNotReady
add_entities(
async_add_entities(
[ScrapeSensor(rest, name, select, attr, index, value_template, unit)], True
)
@ -118,9 +118,9 @@ class ScrapeSensor(Entity):
"""Return the state of the device."""
return self._state
def update(self):
async def async_update(self):
"""Get the latest data from the source and updates the state."""
self.rest.update()
await self.rest.async_update()
if self.rest.data is None:
_LOGGER.error("Unable to retrieve data for %s", self.name)
return
@ -143,8 +143,12 @@ class ScrapeSensor(Entity):
return
if self._value_template is not None:
self._state = self._value_template.render_with_possible_json_value(
self._state = self._value_template.async_render_with_possible_json_value(
value, None
)
else:
self._state = value
async def async_will_remove_from_hass(self):
"""Shutdown the session."""
await self.rest.async_remove()

3
requirements_all.txt
View File

@ -777,6 +777,9 @@ horimote==0.4.1
# homeassistant.components.remember_the_milk
httplib2==0.10.3
# homeassistant.components.rest
httpx==0.16.1
# homeassistant.components.huawei_lte
huawei-lte-api==1.4.12

1
requirements_test.txt
View File

@ -24,5 +24,6 @@ pytest-xdist==2.1.0
pytest==6.0.2
requests_mock==1.8.0
responses==0.12.0
respx==0.14.0
stdlib-list==0.7.0
tqdm==4.49.0

3
requirements_test_all.txt
View File

@ -400,6 +400,9 @@ homematicip==0.11.0
# homeassistant.components.remember_the_milk
httplib2==0.10.3
# homeassistant.components.rest
httpx==0.16.1
# homeassistant.components.huawei_lte
huawei-lte-api==1.4.12

618
tests/components/rest/test_binary_sensor.py
View File

@ -1,267 +1,381 @@
"""The tests for the REST binary sensor platform."""
import unittest
import pytest
from pytest import raises
import requests
from requests.exceptions import Timeout
import requests_mock
import asyncio
from os import path
import httpx
import respx
from homeassistant import config as hass_config
import homeassistant.components.binary_sensor as binary_sensor
import homeassistant.components.rest.binary_sensor as rest
from homeassistant.const import CONTENT_TYPE_JSON, STATE_OFF, STATE_ON
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import template
from homeassistant.setup import setup_component
from homeassistant.const import (
ATTR_ENTITY_ID,
CONTENT_TYPE_JSON,
SERVICE_RELOAD,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
)
from homeassistant.setup import async_setup_component
from tests.async_mock import Mock, patch
from tests.common import assert_setup_component, get_test_home_assistant
class TestRestBinarySensorSetup(unittest.TestCase):
"""Tests for setting up the REST binary sensor platform."""
DEVICES = []
def add_devices(self, devices, update_before_add=False):
"""Mock add devices."""
for device in devices:
self.DEVICES.append(device)
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
# Reset for this test.
self.DEVICES = []
self.addCleanup(self.hass.stop)
def test_setup_missing_config(self):
"""Test setup with configuration missing required entries."""
with assert_setup_component(0):
assert setup_component(
self.hass, binary_sensor.DOMAIN, {"binary_sensor": {"platform": "rest"}}
)
def test_setup_missing_schema(self):
"""Test setup with resource missing schema."""
with pytest.raises(PlatformNotReady):
rest.setup_platform(
self.hass,
{"platform": "rest", "resource": "localhost", "method": "GET"},
None,
)
@patch("requests.Session.send", side_effect=requests.exceptions.ConnectionError())
def test_setup_failed_connect(self, mock_req):
"""Test setup when connection error occurs."""
with raises(PlatformNotReady):
rest.setup_platform(
self.hass,
{"platform": "rest", "resource": "http://localhost", "method": "GET"},
self.add_devices,
None,
)
assert len(self.DEVICES) == 0
@patch("requests.Session.send", side_effect=Timeout())
def test_setup_timeout(self, mock_req):
"""Test setup when connection timeout occurs."""
with raises(PlatformNotReady):
rest.setup_platform(
self.hass,
{"platform": "rest", "resource": "http://localhost", "method": "GET"},
self.add_devices,
None,
)
assert len(self.DEVICES) == 0
@requests_mock.Mocker()
def test_setup_minimum(self, mock_req):
"""Test setup with minimum configuration."""
mock_req.get("http://localhost", status_code=200)
with assert_setup_component(1, "binary_sensor"):
assert setup_component(
self.hass,
"binary_sensor",
{"binary_sensor": {"platform": "rest", "resource": "http://localhost"}},
)
self.hass.block_till_done()
assert 1 == mock_req.call_count
@requests_mock.Mocker()
def test_setup_minimum_resource_template(self, mock_req):
"""Test setup with minimum configuration (resource_template)."""
mock_req.get("http://localhost", status_code=200)
with assert_setup_component(1, "binary_sensor"):
assert setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource_template": "http://localhost",
}
},
)
self.hass.block_till_done()
assert mock_req.call_count == 1
@requests_mock.Mocker()
def test_setup_duplicate_resource(self, mock_req):
"""Test setup with duplicate resources."""
mock_req.get("http://localhost", status_code=200)
with assert_setup_component(0, "binary_sensor"):
assert setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"resource_template": "http://localhost",
}
},
)
self.hass.block_till_done()
@requests_mock.Mocker()
def test_setup_get(self, mock_req):
"""Test setup with valid configuration."""
mock_req.get("http://localhost", status_code=200)
with assert_setup_component(1, "binary_sensor"):
assert setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
"value_template": "{{ value_json.key }}",
"name": "foo",
"verify_ssl": "true",
"authentication": "basic",
"username": "my username",
"password": "my password",
"headers": {"Accept": CONTENT_TYPE_JSON},
}
},
)
self.hass.block_till_done()
assert 1 == mock_req.call_count
@requests_mock.Mocker()
def test_setup_post(self, mock_req):
"""Test setup with valid configuration."""
mock_req.post("http://localhost", status_code=200)
with assert_setup_component(1, "binary_sensor"):
assert setup_component(
self.hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "POST",
"value_template": "{{ value_json.key }}",
"payload": '{ "device": "toaster"}',
"name": "foo",
"verify_ssl": "true",
"authentication": "basic",
"username": "my username",
"password": "my password",
"headers": {"Accept": CONTENT_TYPE_JSON},
}
},
)
self.hass.block_till_done()
assert 1 == mock_req.call_count
async def test_setup_missing_basic_config(hass):
"""Test setup with configuration missing required entries."""
assert await async_setup_component(
hass, binary_sensor.DOMAIN, {"binary_sensor": {"platform": "rest"}}
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
class TestRestBinarySensor(unittest.TestCase):
"""Tests for REST binary sensor platform."""
async def test_setup_missing_config(hass):
"""Test setup with configuration missing required entries."""
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
"binary_sensor": {
"platform": "rest",
"resource": "localhost",
"method": "GET",
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.rest = Mock("RestData")
self.rest.update = Mock(
"RestData.update", side_effect=self.update_side_effect('{ "key": false }')
@respx.mock
async def test_setup_failed_connect(hass):
"""Test setup when connection error occurs."""
respx.get(
"http://localhost", content=httpx.RequestError(message="any", request=Mock())
)
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
@respx.mock
async def test_setup_timeout(hass):
"""Test setup when connection timeout occurs."""
respx.get("http://localhost", content=asyncio.TimeoutError())
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
"binary_sensor": {
"platform": "rest",
"resource": "localhost",
"method": "GET",
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
@respx.mock
async def test_setup_minimum(hass):
"""Test setup with minimum configuration."""
respx.get("http://localhost", status_code=200)
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
@respx.mock
async def test_setup_minimum_resource_template(hass):
"""Test setup with minimum configuration (resource_template)."""
respx.get("http://localhost", status_code=200)
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
"binary_sensor": {
"platform": "rest",
"resource_template": "http://localhost",
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
@respx.mock
async def test_setup_duplicate_resource_template(hass):
"""Test setup with duplicate resources."""
respx.get("http://localhost", status_code=200)
assert await async_setup_component(
hass,
binary_sensor.DOMAIN,
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"resource_template": "http://localhost",
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 0
@respx.mock
async def test_setup_get(hass):
"""Test setup with valid configuration."""
respx.get("http://localhost", status_code=200, content="{}")
assert await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
"value_template": "{{ value_json.key }}",
"name": "foo",
"verify_ssl": "true",
"timeout": 30,
"authentication": "basic",
"username": "my username",
"password": "my password",
"headers": {"Accept": CONTENT_TYPE_JSON},
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
@respx.mock
async def test_setup_get_digest_auth(hass):
"""Test setup with valid configuration."""
respx.get("http://localhost", status_code=200, content="{}")
assert await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
"value_template": "{{ value_json.key }}",
"name": "foo",
"verify_ssl": "true",
"timeout": 30,
"authentication": "digest",
"username": "my username",
"password": "my password",
"headers": {"Accept": CONTENT_TYPE_JSON},
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
@respx.mock
async def test_setup_post(hass):
"""Test setup with valid configuration."""
respx.post("http://localhost", status_code=200, content="{}")
assert await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "POST",
"value_template": "{{ value_json.key }}",
"payload": '{ "device": "toaster"}',
"name": "foo",
"verify_ssl": "true",
"timeout": 30,
"authentication": "basic",
"username": "my username",
"password": "my password",
"headers": {"Accept": CONTENT_TYPE_JSON},
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
@respx.mock
async def test_setup_get_off(hass):
"""Test setup with valid off configuration."""
respx.get(
"http://localhost",
status_code=200,
headers={"content-type": "text/json"},
content='{"dog": false}',
)
assert await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
"value_template": "{{ value_json.dog }}",
"name": "foo",
"verify_ssl": "true",
"timeout": 30,
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
state = hass.states.get("binary_sensor.foo")
assert state.state == STATE_OFF
@respx.mock
async def test_setup_get_on(hass):
"""Test setup with valid on configuration."""
respx.get(
"http://localhost",
status_code=200,
headers={"content-type": "text/json"},
content='{"dog": true}',
)
assert await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
"value_template": "{{ value_json.dog }}",
"name": "foo",
"verify_ssl": "true",
"timeout": 30,
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
state = hass.states.get("binary_sensor.foo")
assert state.state == STATE_ON
@respx.mock
async def test_setup_with_exception(hass):
"""Test setup with exception."""
respx.get("http://localhost", status_code=200, content="{}")
assert await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"resource": "http://localhost",
"method": "GET",
"value_template": "{{ value_json.dog }}",
"name": "foo",
"verify_ssl": "true",
"timeout": 30,
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
state = hass.states.get("binary_sensor.foo")
assert state.state == STATE_OFF
await async_setup_component(hass, "homeassistant", {})
await hass.async_block_till_done()
respx.clear()
respx.get(
"http://localhost", content=httpx.RequestError(message="any", request=Mock())
)
await hass.services.async_call(
"homeassistant",
"update_entity",
{ATTR_ENTITY_ID: ["binary_sensor.foo"]},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.foo")
assert state.state == STATE_UNAVAILABLE
@respx.mock
async def test_reload(hass):
"""Verify we can reload reset sensors."""
respx.get("http://localhost", status_code=200)
await async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "rest",
"method": "GET",
"name": "mockrest",
"resource": "http://localhost",
}
},
)
await hass.async_block_till_done()
await hass.async_start()
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
assert hass.states.get("binary_sensor.mockrest")
yaml_path = path.join(
_get_fixtures_base_path(),
"fixtures",
"rest/configuration.yaml",
)
with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
"rest",
SERVICE_RELOAD,
{},
blocking=True,
)
self.name = "foo"
self.device_class = "light"
self.value_template = template.Template("{{ value_json.key }}", self.hass)
self.force_update = False
self.resource_template = None
await hass.async_block_till_done()
self.binary_sensor = rest.RestBinarySensor(
self.hass,
self.rest,
self.name,
self.device_class,
self.value_template,
self.force_update,
self.resource_template,
)
self.addCleanup(self.hass.stop)
assert hass.states.get("binary_sensor.mockreset") is None
assert hass.states.get("binary_sensor.rollout")
def update_side_effect(self, data):
"""Side effect function for mocking RestData.update()."""
self.rest.data = data
def test_name(self):
"""Test the name."""
assert self.name == self.binary_sensor.name
def test_device_class(self):
"""Test the device class."""
assert self.device_class == self.binary_sensor.device_class
def test_initial_state(self):
"""Test the initial state."""
self.binary_sensor.update()
assert STATE_OFF == self.binary_sensor.state
def test_update_when_value_is_none(self):
"""Test state gets updated to unknown when sensor returns no data."""
self.rest.update = Mock(
"RestData.update", side_effect=self.update_side_effect(None)
)
self.binary_sensor.update()
assert not self.binary_sensor.available
def test_update_when_value_changed(self):
"""Test state gets updated when sensor returns a new status."""
self.rest.update = Mock(
"rest.RestData.update",
side_effect=self.update_side_effect('{ "key": true }'),
)
self.binary_sensor.update()
assert STATE_ON == self.binary_sensor.state
assert self.binary_sensor.available
def test_update_when_failed_request(self):
"""Test state gets updated when sensor returns a new status."""
self.rest.update = Mock(
"rest.RestData.update", side_effect=self.update_side_effect(None)
)
self.binary_sensor.update()
assert not self.binary_sensor.available
def test_update_with_no_template(self):
"""Test update when there is no value template."""
self.rest.update = Mock(
"rest.RestData.update", side_effect=self.update_side_effect("true")
)
self.binary_sensor = rest.RestBinarySensor(
self.hass,
self.rest,
self.name,
self.device_class,
None,
self.force_update,
self.resource_template,
)
self.binary_sensor.update()
assert STATE_ON == self.binary_sensor.state
assert self.binary_sensor.available
def _get_fixtures_base_path():
return path.dirname(path.dirname(path.dirname(__file__)))

1311
tests/components/rest/test_sensor.py
View File

File diff suppressed because it is too large Load Diff

6
tests/fixtures/rest/configuration.yaml vendored
View File

@ -4,6 +4,12 @@ sensor:
method: GET
name: rollout
binary_sensor:
- platform: rest
resource: "http://localhost"
method: GET
name: rollout
notify:
- name: rest_reloaded
platform: rest