jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-18 02:37:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate rest switch to httpx (#90768)

Browse Source
This commit is contained in:
epenet 2023-05-05 14:43:39 +02:00 committed by GitHub
parent 5843c1fa3b
commit 802e907a35
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 120 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
homeassistant/components/rest/switch.py
View File

@ -6,8 +6,8 @@ from http import HTTPStatus
import logging import logging
from typing import Any from typing import Any
import aiohttp
import async_timeout import async_timeout
import httpx
import voluptuous as vol import voluptuous as vol
from homeassistant.components.switch import ( from homeassistant.components.switch import (
@ -30,8 +30,8 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import config_validation as cv, template from homeassistant.helpers import config_validation as cv, template
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.template_entity import ( from homeassistant.helpers.template_entity import (
TEMPLATE_ENTITY_BASE_SCHEMA, TEMPLATE_ENTITY_BASE_SCHEMA,
TemplateEntity, TemplateEntity,
@ -89,8 +89,8 @@ async def async_setup_platform(
switch = RestSwitch(hass, config, unique_id) switch = RestSwitch(hass, config, unique_id)
req = await switch.get_device_state(hass) req = await switch.get_device_state(hass)
if req.status >= HTTPStatus.BAD_REQUEST: if req.status_code >= HTTPStatus.BAD_REQUEST:
_LOGGER.error("Got non-ok response from resource: %s", req.status) _LOGGER.error("Got non-ok response from resource: %s", req.status_code)
else: else:
async_add_entities([switch]) async_add_entities([switch])
except (TypeError, ValueError): except (TypeError, ValueError):
@ -98,7 +98,7 @@ async def async_setup_platform(
"Missing resource or schema in configuration. " "Missing resource or schema in configuration. "
"Add http:// or https:// to your URL" "Add http:// or https:// to your URL"
) )
except (asyncio.TimeoutError, aiohttp.ClientError) as exc: except (asyncio.TimeoutError, httpx.RequestError) as exc:
raise PlatformNotReady(f"No route to resource/endpoint: {resource}") from exc raise PlatformNotReady(f"No route to resource/endpoint: {resource}") from exc
@ -120,11 +120,11 @@ class RestSwitch(TemplateEntity, SwitchEntity):
unique_id=unique_id, unique_id=unique_id,
) )
auth: aiohttp.BasicAuth | None = None auth: httpx.BasicAuth | None = None
username: str | None = None username: str | None = None
if username := config.get(CONF_USERNAME): if username := config.get(CONF_USERNAME):
password: str = config[CONF_PASSWORD] password: str = config[CONF_PASSWORD]
auth = aiohttp.BasicAuth(username, password=password) auth = httpx.BasicAuth(username, password=password)
self._resource: str = config[CONF_RESOURCE] self._resource: str = config[CONF_RESOURCE]
self._state_resource: str = config.get(CONF_STATE_RESOURCE) or self._resource self._state_resource: str = config.get(CONF_STATE_RESOURCE) or self._resource
@ -155,13 +155,13 @@ class RestSwitch(TemplateEntity, SwitchEntity):
try: try:
req = await self.set_device_state(body_on_t) req = await self.set_device_state(body_on_t)
if req.status == HTTPStatus.OK: if req.status_code == HTTPStatus.OK:
self._attr_is_on = True self._attr_is_on = True
else: else:
_LOGGER.error( _LOGGER.error(
"Can't turn on %s. Is resource/endpoint offline?", self._resource "Can't turn on %s. Is resource/endpoint offline?", self._resource
) )
except (asyncio.TimeoutError, aiohttp.ClientError): except (asyncio.TimeoutError, httpx.RequestError):
_LOGGER.error("Error while switching on %s", self._resource) _LOGGER.error("Error while switching on %s", self._resource)
async def async_turn_off(self, **kwargs: Any) -> None: async def async_turn_off(self, **kwargs: Any) -> None:
@ -170,24 +170,24 @@ class RestSwitch(TemplateEntity, SwitchEntity):
try: try:
req = await self.set_device_state(body_off_t) req = await self.set_device_state(body_off_t)
if req.status == HTTPStatus.OK: if req.status_code == HTTPStatus.OK:
self._attr_is_on = False self._attr_is_on = False
else: else:
_LOGGER.error( _LOGGER.error(
"Can't turn off %s. Is resource/endpoint offline?", self._resource "Can't turn off %s. Is resource/endpoint offline?", self._resource
) )
except (asyncio.TimeoutError, aiohttp.ClientError): except (asyncio.TimeoutError, httpx.RequestError):
_LOGGER.error("Error while switching off %s", self._resource) _LOGGER.error("Error while switching off %s", self._resource)
async def set_device_state(self, body: Any) -> aiohttp.ClientResponse: async def set_device_state(self, body: Any) -> httpx.Response:
"""Send a state update to the device.""" """Send a state update to the device."""
websession = async_get_clientsession(self.hass, self._verify_ssl) websession = get_async_client(self.hass, self._verify_ssl)
rendered_headers = template.render_complex(self._headers, parse_result=False) rendered_headers = template.render_complex(self._headers, parse_result=False)
rendered_params = template.render_complex(self._params) rendered_params = template.render_complex(self._params)
async with async_timeout.timeout(self._timeout): async with async_timeout.timeout(self._timeout):
req: aiohttp.ClientResponse = await getattr(websession, self._method)( req: httpx.Response = await getattr(websession, self._method)(
self._resource, self._resource,
auth=self._auth, auth=self._auth,
data=bytes(body, "utf-8"), data=bytes(body, "utf-8"),
@ -202,12 +202,12 @@ class RestSwitch(TemplateEntity, SwitchEntity):
await self.get_device_state(self.hass) await self.get_device_state(self.hass)
except asyncio.TimeoutError: except asyncio.TimeoutError:
_LOGGER.exception("Timed out while fetching data") _LOGGER.exception("Timed out while fetching data")
except aiohttp.ClientError as err: except httpx.RequestError as err:
_LOGGER.exception("Error while fetching data: %s", err) _LOGGER.exception("Error while fetching data: %s", err)
async def get_device_state(self, hass: HomeAssistant) -> aiohttp.ClientResponse: async def get_device_state(self, hass: HomeAssistant) -> httpx.Response:
"""Get the latest data from REST API and update the state.""" """Get the latest data from REST API and update the state."""
websession = async_get_clientsession(hass, self._verify_ssl) websession = get_async_client(hass, self._verify_ssl)
rendered_headers = template.render_complex(self._headers, parse_result=False) rendered_headers = template.render_complex(self._headers, parse_result=False)
rendered_params = template.render_complex(self._params) rendered_params = template.render_complex(self._params)
@ -219,7 +219,7 @@ class RestSwitch(TemplateEntity, SwitchEntity):
headers=rendered_headers, headers=rendered_headers,
params=rendered_params, params=rendered_params,
) )
text = await req.text() text = req.text
if self._is_on_template is not None: if self._is_on_template is not None:
text = self._is_on_template.async_render_with_possible_json_value( text = self._is_on_template.async_render_with_possible_json_value(

212
tests/components/rest/test_switch.py
View File

@ -2,8 +2,9 @@
import asyncio import asyncio
from http import HTTPStatus from http import HTTPStatus
import aiohttp import httpx
import pytest import pytest
import respx
from homeassistant.components.rest import DOMAIN from homeassistant.components.rest import DOMAIN
from homeassistant.components.rest.switch import ( from homeassistant.components.rest.switch import (
@ -45,7 +46,6 @@ from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow from homeassistant.util.dt import utcnow
from tests.common import assert_setup_component, async_fire_time_changed from tests.common import assert_setup_component, async_fire_time_changed
from tests.test_util.aiohttp import AiohttpClientMocker
NAME = "foo" NAME = "foo"
DEVICE_CLASS = SwitchDeviceClass.SWITCH DEVICE_CLASS = SwitchDeviceClass.SWITCH
@ -75,13 +75,13 @@ async def test_setup_missing_schema(
assert "Invalid config for [switch.rest]: invalid url" in caplog.text assert "Invalid config for [switch.rest]: invalid url" in caplog.text
@respx.mock
async def test_setup_failed_connect( async def test_setup_failed_connect(
hass: HomeAssistant, hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test setup when connection error occurs.""" """Test setup when connection error occurs."""
aioclient_mock.get(RESOURCE, exc=aiohttp.ClientError) respx.get(RESOURCE).mock(side_effect=asyncio.TimeoutError())
config = {SWITCH_DOMAIN: {CONF_PLATFORM: DOMAIN, CONF_RESOURCE: RESOURCE}} config = {SWITCH_DOMAIN: {CONF_PLATFORM: DOMAIN, CONF_RESOURCE: RESOURCE}}
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -89,13 +89,13 @@ async def test_setup_failed_connect(
assert "No route to resource/endpoint" in caplog.text assert "No route to resource/endpoint" in caplog.text
@respx.mock
async def test_setup_timeout( async def test_setup_timeout(
hass: HomeAssistant, hass: HomeAssistant,
aioclient_mock: AiohttpClientMocker,
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test setup when connection timeout occurs.""" """Test setup when connection timeout occurs."""
aioclient_mock.get(RESOURCE, exc=asyncio.TimeoutError()) respx.get(RESOURCE).mock(side_effect=asyncio.TimeoutError())
config = {SWITCH_DOMAIN: {CONF_PLATFORM: DOMAIN, CONF_RESOURCE: RESOURCE}} config = {SWITCH_DOMAIN: {CONF_PLATFORM: DOMAIN, CONF_RESOURCE: RESOURCE}}
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -103,23 +103,21 @@ async def test_setup_timeout(
assert "No route to resource/endpoint" in caplog.text assert "No route to resource/endpoint" in caplog.text
async def test_setup_minimum( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_setup_minimum(hass: HomeAssistant) -> None:
) -> None:
"""Test setup with minimum configuration.""" """Test setup with minimum configuration."""
aioclient_mock.get(RESOURCE, status=HTTPStatus.OK) route = respx.get(RESOURCE) % HTTPStatus.OK
config = {SWITCH_DOMAIN: {CONF_PLATFORM: DOMAIN, CONF_RESOURCE: RESOURCE}} config = {SWITCH_DOMAIN: {CONF_PLATFORM: DOMAIN, CONF_RESOURCE: RESOURCE}}
with assert_setup_component(1, SWITCH_DOMAIN): with assert_setup_component(1, SWITCH_DOMAIN):
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.call_count == 1 assert route.call_count == 1
async def test_setup_query_params( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_setup_query_params(hass: HomeAssistant) -> None:
) -> None:
"""Test setup with query params.""" """Test setup with query params."""
aioclient_mock.get("http://localhost/?search=something", status=HTTPStatus.OK) route = respx.get("http://localhost/?search=something") % HTTPStatus.OK
config = { config = {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
CONF_PLATFORM: DOMAIN, CONF_PLATFORM: DOMAIN,
@ -131,12 +129,13 @@ async def test_setup_query_params(
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.call_count == 1 assert route.call_count == 1
async def test_setup(hass: HomeAssistant, aioclient_mock: AiohttpClientMocker) -> None: @respx.mock
async def test_setup(hass: HomeAssistant) -> None:
"""Test setup with valid configuration.""" """Test setup with valid configuration."""
aioclient_mock.get(RESOURCE, status=HTTPStatus.OK) route = respx.get(RESOURCE) % HTTPStatus.OK
config = { config = {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
CONF_PLATFORM: DOMAIN, CONF_PLATFORM: DOMAIN,
@ -149,16 +148,15 @@ async def test_setup(hass: HomeAssistant, aioclient_mock: AiohttpClientMocker) -
} }
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.call_count == 1 assert route.call_count == 1
assert_setup_component(1, SWITCH_DOMAIN) assert_setup_component(1, SWITCH_DOMAIN)
async def test_setup_with_state_resource( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_setup_with_state_resource(hass: HomeAssistant) -> None:
) -> None:
"""Test setup with valid configuration.""" """Test setup with valid configuration."""
aioclient_mock.get(RESOURCE, status=HTTPStatus.NOT_FOUND) respx.get(RESOURCE) % HTTPStatus.NOT_FOUND
aioclient_mock.get("http://localhost/state", status=HTTPStatus.OK) route = respx.get("http://localhost/state") % HTTPStatus.OK
config = { config = {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
CONF_PLATFORM: DOMAIN, CONF_PLATFORM: DOMAIN,
@ -172,15 +170,14 @@ async def test_setup_with_state_resource(
} }
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.call_count == 1 assert route.call_count == 1
assert_setup_component(1, SWITCH_DOMAIN) assert_setup_component(1, SWITCH_DOMAIN)
async def test_setup_with_templated_headers_params( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_setup_with_templated_headers_params(hass: HomeAssistant) -> None:
) -> None:
"""Test setup with valid configuration.""" """Test setup with valid configuration."""
aioclient_mock.get(RESOURCE, status=HTTPStatus.OK) route = respx.get(RESOURCE) % HTTPStatus.OK
config = { config = {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
CONF_PLATFORM: DOMAIN, CONF_PLATFORM: DOMAIN,
@ -198,21 +195,21 @@ async def test_setup_with_templated_headers_params(
} }
assert await async_setup_component(hass, SWITCH_DOMAIN, config) assert await async_setup_component(hass, SWITCH_DOMAIN, config)
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.call_count == 1 assert route.call_count == 1
assert aioclient_mock.mock_calls[-1][3].get("Accept") == CONTENT_TYPE_JSON last_call = route.calls[-1]
assert aioclient_mock.mock_calls[-1][3].get("User-Agent") == "Mozilla/5.0" last_request: httpx.Request = last_call.request
assert aioclient_mock.mock_calls[-1][1].query["start"] == "0" assert last_request.headers.get("Accept") == CONTENT_TYPE_JSON
assert aioclient_mock.mock_calls[-1][1].query["end"] == "5" assert last_request.headers.get("User-Agent") == "Mozilla/5.0"
assert last_request.url.params["start"] == "0"
assert last_request.url.params["end"] == "5"
assert_setup_component(1, SWITCH_DOMAIN) assert_setup_component(1, SWITCH_DOMAIN)
# Tests for REST switch platform. # Tests for REST switch platform.
async def _async_setup_test_switch( async def _async_setup_test_switch(hass: HomeAssistant) -> None:
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker respx.get(RESOURCE) % HTTPStatus.OK
) -> None:
aioclient_mock.get(RESOURCE, status=HTTPStatus.OK)
headers = {"Content-type": CONTENT_TYPE_JSON} headers = {"Content-type": CONTENT_TYPE_JSON}
config = { config = {
@ -223,51 +220,48 @@ async def _async_setup_test_switch(
CONF_STATE_RESOURCE: STATE_RESOURCE, CONF_STATE_RESOURCE: STATE_RESOURCE,
CONF_HEADERS: headers, CONF_HEADERS: headers,
} }
assert await async_setup_component(hass, SWITCH_DOMAIN, {SWITCH_DOMAIN: config}) assert await async_setup_component(hass, SWITCH_DOMAIN, {SWITCH_DOMAIN: config})
await hass.async_block_till_done() await hass.async_block_till_done()
assert_setup_component(1, SWITCH_DOMAIN) assert_setup_component(1, SWITCH_DOMAIN)
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
aioclient_mock.clear_requests() respx.reset()
async def test_name(hass: HomeAssistant, aioclient_mock: AiohttpClientMocker) -> None: @respx.mock
async def test_name(hass: HomeAssistant) -> None:
"""Test the name.""" """Test the name."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
state = hass.states.get("switch.foo") state = hass.states.get("switch.foo")
assert state.attributes[ATTR_FRIENDLY_NAME] == NAME assert state.attributes[ATTR_FRIENDLY_NAME] == NAME
async def test_device_class( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_device_class(hass: HomeAssistant) -> None:
) -> None:
"""Test the device class.""" """Test the device class."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
state = hass.states.get("switch.foo") state = hass.states.get("switch.foo")
assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS assert state.attributes[ATTR_DEVICE_CLASS] == DEVICE_CLASS
async def test_is_on_before_update( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_is_on_before_update(hass: HomeAssistant) -> None:
) -> None:
"""Test is_on in initial state.""" """Test is_on in initial state."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
state = hass.states.get("switch.foo") state = hass.states.get("switch.foo")
assert state.state == STATE_UNKNOWN assert state.state == STATE_UNKNOWN
async def test_turn_on_success( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_turn_on_success(hass: HomeAssistant) -> None:
) -> None:
"""Test turn_on.""" """Test turn_on."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.post(RESOURCE, status=HTTPStatus.OK) route = respx.post(RESOURCE) % HTTPStatus.OK
aioclient_mock.get(RESOURCE, exc=aiohttp.ClientError) respx.get(RESOURCE).mock(side_effect=httpx.RequestError)
assert await hass.services.async_call( assert await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
@ -276,17 +270,18 @@ async def test_turn_on_success(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.mock_calls[-2][2].decode() == "ON" last_call = route.calls[-1]
last_request: httpx.Request = last_call.request
assert last_request.content.decode() == "ON"
assert hass.states.get("switch.foo").state == STATE_ON assert hass.states.get("switch.foo").state == STATE_ON
async def test_turn_on_status_not_ok( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_turn_on_status_not_ok(hass: HomeAssistant) -> None:
) -> None:
"""Test turn_on when error status returned.""" """Test turn_on when error status returned."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.post(RESOURCE, status=HTTPStatus.INTERNAL_SERVER_ERROR) route = respx.post(RESOURCE) % HTTPStatus.INTERNAL_SERVER_ERROR
assert await hass.services.async_call( assert await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
@ -295,17 +290,18 @@ async def test_turn_on_status_not_ok(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.mock_calls[-1][2].decode() == "ON" last_call = route.calls[-1]
last_request: httpx.Request = last_call.request
assert last_request.content.decode() == "ON"
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
async def test_turn_on_timeout( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_turn_on_timeout(hass: HomeAssistant) -> None:
) -> None:
"""Test turn_on when timeout occurs.""" """Test turn_on when timeout occurs."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.post(RESOURCE, status=HTTPStatus.INTERNAL_SERVER_ERROR) respx.post(RESOURCE) % HTTPStatus.INTERNAL_SERVER_ERROR
assert await hass.services.async_call( assert await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_ON, SERVICE_TURN_ON,
@ -317,14 +313,13 @@ async def test_turn_on_timeout(
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
async def test_turn_off_success( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_turn_off_success(hass: HomeAssistant) -> None:
) -> None:
"""Test turn_off.""" """Test turn_off."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.post(RESOURCE, status=HTTPStatus.OK) route = respx.post(RESOURCE) % HTTPStatus.OK
aioclient_mock.get(RESOURCE, exc=aiohttp.ClientError) respx.get(RESOURCE).mock(side_effect=httpx.RequestError)
assert await hass.services.async_call( assert await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
@ -333,18 +328,19 @@ async def test_turn_off_success(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.mock_calls[-2][2].decode() == "OFF" last_call = route.calls[-1]
last_request: httpx.Request = last_call.request
assert last_request.content.decode() == "OFF"
assert hass.states.get("switch.foo").state == STATE_OFF assert hass.states.get("switch.foo").state == STATE_OFF
async def test_turn_off_status_not_ok( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_turn_off_status_not_ok(hass: HomeAssistant) -> None:
) -> None:
"""Test turn_off when error status returned.""" """Test turn_off when error status returned."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.post(RESOURCE, status=HTTPStatus.INTERNAL_SERVER_ERROR) route = respx.post(RESOURCE) % HTTPStatus.INTERNAL_SERVER_ERROR
assert await hass.services.async_call( assert await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
@ -353,18 +349,19 @@ async def test_turn_off_status_not_ok(
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert aioclient_mock.mock_calls[-1][2].decode() == "OFF" last_call = route.calls[-1]
last_request: httpx.Request = last_call.request
assert last_request.content.decode() == "OFF"
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
async def test_turn_off_timeout( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_turn_off_timeout(hass: HomeAssistant) -> None:
) -> None:
"""Test turn_off when timeout occurs.""" """Test turn_off when timeout occurs."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.post(RESOURCE, exc=asyncio.TimeoutError()) respx.post(RESOURCE).mock(side_effect=asyncio.TimeoutError())
assert await hass.services.async_call( assert await hass.services.async_call(
SWITCH_DOMAIN, SWITCH_DOMAIN,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
@ -376,64 +373,59 @@ async def test_turn_off_timeout(
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
async def test_update_when_on( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_update_when_on(hass: HomeAssistant) -> None:
) -> None:
"""Test update when switch is on.""" """Test update when switch is on."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.get(RESOURCE, text="ON") respx.get(RESOURCE).respond(text="ON")
async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get("switch.foo").state == STATE_ON assert hass.states.get("switch.foo").state == STATE_ON
async def test_update_when_off( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_update_when_off(hass: HomeAssistant) -> None:
) -> None:
"""Test update when switch is off.""" """Test update when switch is off."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.get(RESOURCE, text="OFF") respx.get(RESOURCE).respond(text="OFF")
async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get("switch.foo").state == STATE_OFF assert hass.states.get("switch.foo").state == STATE_OFF
async def test_update_when_unknown( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_update_when_unknown(hass: HomeAssistant) -> None:
) -> None:
"""Test update when unknown status returned.""" """Test update when unknown status returned."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.get(RESOURCE, text="unknown status") respx.get(RESOURCE).respond(text="unknown status")
async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
async def test_update_timeout( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_update_timeout(hass: HomeAssistant) -> None:
) -> None:
"""Test update when timeout occurs.""" """Test update when timeout occurs."""
await _async_setup_test_switch(hass, aioclient_mock) await _async_setup_test_switch(hass)
aioclient_mock.get(RESOURCE, exc=asyncio.TimeoutError()) respx.get(RESOURCE).mock(side_effect=asyncio.TimeoutError())
async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL) async_fire_time_changed(hass, utcnow() + SCAN_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
assert hass.states.get("switch.foo").state == STATE_UNKNOWN assert hass.states.get("switch.foo").state == STATE_UNKNOWN
async def test_entity_config( @respx.mock
hass: HomeAssistant, aioclient_mock: AiohttpClientMocker async def test_entity_config(hass: HomeAssistant) -> None:
) -> None:
"""Test entity configuration.""" """Test entity configuration."""
aioclient_mock.get(RESOURCE, status=HTTPStatus.OK) respx.get(RESOURCE) % HTTPStatus.OK
config = { config = {
SWITCH_DOMAIN: { SWITCH_DOMAIN: {
# REST configuration # REST configuration