github/ha-core
1
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-09-09 12:51:22 +02:00
Code Releases Activity

Ability to mock long poll requests + refactor qwikswitch unit… (#33804)

Browse Source 
* added the ability to mock a "long poll" get request by setting up the
waiting request and feeding responses to it
with this, refactored the qwikswitch test so it doesn't use global variables
and is more understandable and maintainable

* added import asyncio from merge

* added assert that first call with long_poll has empty content

* passing json instead of the binary string

* use json instead of text in mock requests

* refactored to use a proxy

* return the proxy also for the http methods other than get

* refactored so any side_effect can be used and created the long_poll side effect

* simplified by using kwargs
needed to move the json->text->content logic from mocker to mockrequest

* no need to explicitly define method and url
This commit is contained in:
Ziv 2020-04-11 00:57:39 +03:00 committed by GitHub
parent 8e6e8dfbe0
commit 302e631984
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 118 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
tests/components/qwikswitch/test_init.py
View File

@ -1,92 +1,81 @@
"""Test qwikswitch sensors."""
import asyncio
import logging
from aiohttp.client_exceptions import ClientError
import pytest
from homeassistant.bootstrap import async_setup_component
from homeassistant.components.qwikswitch import DOMAIN as QWIKSWITCH
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.setup import async_setup_component
from tests.test_util.aiohttp import mock_aiohttp_client
from tests.test_util.aiohttp import MockLongPollSideEffect
_LOGGER = logging.getLogger(__name__)
class AiohttpClientMockResponseList(list):
"""Return multiple values for aiohttp Mocker.
aoihttp mocker uses decode to fetch the next value.
"""
def decode(self, _):
"""Return next item from list."""
try:
res = list.pop(self, 0)
_LOGGER.debug("MockResponseList popped %s: %s", res, self)
if isinstance(res, Exception):
raise res
return res
except IndexError:
raise AssertionError("MockResponseList empty")
async def wait_till_empty(self, hass):
"""Wait until empty."""
while self:
await hass.async_block_till_done()
await hass.async_block_till_done()
DEVICES = [
{
"id": "@000001",
"name": "Switch 1",
"type": "rel",
"val": "OFF",
"time": "1522777506",
"rssi": "51%",
},
{
"id": "@000002",
"name": "Light 2",
"type": "rel",
"val": "ON",
"time": "1522777507",
"rssi": "45%",
},
{
"id": "@000003",
"name": "Dim 3",
"type": "dim",
"val": "280c00",
"time": "1522777544",
"rssi": "62%",
},
]
LISTEN = AiohttpClientMockResponseList()
@pytest.fixture
def aioclient_mock():
"""HTTP client listen and devices."""
devices = """[
{"id":"@000001","name":"Switch 1","type":"rel","val":"OFF",
"time":"1522777506","rssi":"51%"},
{"id":"@000002","name":"Light 2","type":"rel","val":"ON",
"time":"1522777507","rssi":"45%"},
{"id":"@000003","name":"Dim 3","type":"dim","val":"280c00",
"time":"1522777544","rssi":"62%"}]"""
with mock_aiohttp_client() as mock_session:
mock_session.get("http://127.0.0.1:2020/&listen", content=LISTEN)
mock_session.get("http://127.0.0.1:2020/&device", text=devices)
yield mock_session
async def test_binary_sensor_device(hass, aioclient_mock): # noqa: F811
async def test_binary_sensor_device(hass, aioclient_mock):
"""Test a binary sensor device."""
config = {
"qwikswitch": {
"sensors": {"name": "s1", "id": "@a00001", "channel": 1, "type": "imod"}
}
}
aioclient_mock.get("http://127.0.0.1:2020/&device", json=DEVICES)
listen_mock = MockLongPollSideEffect()
aioclient_mock.get("http://127.0.0.1:2020/&listen", side_effect=listen_mock)
await async_setup_component(hass, QWIKSWITCH, config)
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
state_obj = hass.states.get("binary_sensor.s1")
assert state_obj.state == "off"
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
LISTEN.append('{"id":"@a00001","cmd":"","data":"4e0e1601","rssi":"61%"}')
LISTEN.append(ClientError()) # Will cause a sleep
listen_mock.queue_response(
json={"id": "@a00001", "cmd": "", "data": "4e0e1601", "rssi": "61%"}
)
await asyncio.sleep(0.01)
await hass.async_block_till_done()
state_obj = hass.states.get("binary_sensor.s1")
assert state_obj.state == "on"
LISTEN.append('{"id":"@a00001","cmd":"","data":"4e0e1701","rssi":"61%"}')
hass.data[QWIKSWITCH]._sleep_task.cancel()
await LISTEN.wait_till_empty(hass)
listen_mock.queue_response(
json={"id": "@a00001", "cmd": "", "data": "4e0e1701", "rssi": "61%"},
)
await asyncio.sleep(0.01)
await hass.async_block_till_done()
state_obj = hass.states.get("binary_sensor.s1")
assert state_obj.state == "off"
listen_mock.stop()
async def test_sensor_device(hass, aioclient_mock): # noqa: F811
async def test_sensor_device(hass, aioclient_mock):
"""Test a sensor device."""
config = {
"qwikswitch": {
@ -98,18 +87,22 @@ async def test_sensor_device(hass, aioclient_mock): # noqa: F811
}
}
}
aioclient_mock.get("http://127.0.0.1:2020/&device", json=DEVICES)
listen_mock = MockLongPollSideEffect()
aioclient_mock.get("http://127.0.0.1:2020/&listen", side_effect=listen_mock)
await async_setup_component(hass, QWIKSWITCH, config)
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
state_obj = hass.states.get("sensor.ss1")
assert state_obj.state == "None"
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
LISTEN.append(
'{"id":"@a00001","name":"ss1","type":"rel",' '"val":"4733800001a00000"}'
listen_mock.queue_response(
json={"id": "@a00001", "name": "ss1", "type": "rel", "val": "4733800001a00000"},
)
await asyncio.sleep(0.01)
await hass.async_block_till_done()
state_obj = hass.states.get("sensor.ss1")
assert state_obj.state == "416"
listen_mock.stop()

74
tests/test_util/aiohttp.py
View File

@ -1,4 +1,5 @@
"""Aiohttp test utils."""
import asyncio
from contextlib import contextmanager
import json as _json
import re
@ -6,7 +7,7 @@ from unittest import mock
from urllib.parse import parse_qs
from aiohttp import ClientSession
from aiohttp.client_exceptions import ClientResponseError
from aiohttp.client_exceptions import ClientError, ClientResponseError
from aiohttp.streams import StreamReader
from yarl import URL
@ -48,15 +49,9 @@ class AiohttpClientMocker:
headers={},
exc=None,
cookies=None,
side_effect=None,
):
"""Mock a request."""
if json is not None:
text = _json.dumps(json)
if text is not None:
content = text.encode("utf-8")
if content is None:
content = b""
if not isinstance(url, RETYPE):
url = URL(url)
if params:
@ -64,7 +59,16 @@ class AiohttpClientMocker:
self._mocks.append(
AiohttpClientMockResponse(
method, url, status, content, cookies, exc, headers
method=method,
url=url,
status=status,
response=content,
json=json,
text=text,
cookies=cookies,
exc=exc,
headers=headers,
side_effect=side_effect,
)
)
@ -134,7 +138,8 @@ class AiohttpClientMocker:
for response in self._mocks:
if response.match_request(method, url, params):
self.mock_calls.append((method, url, data, headers))
if response.side_effect:
response = await response.side_effect(method, url, data)
if response.exc:
raise response.exc
return response
@ -148,15 +153,32 @@ class AiohttpClientMockResponse:
"""Mock Aiohttp client response."""
def __init__(
self, method, url, status, response, cookies=None, exc=None, headers=None
self,
method,
url,
status=200,
response=None,
json=None,
text=None,
cookies=None,
exc=None,
headers=None,
side_effect=None,
):
"""Initialize a fake response."""
if json is not None:
text = _json.dumps(json)
if text is not None:
response = text.encode("utf-8")
if response is None:
response = b""
self.method = method
self._url = url
self.status = status
self.response = response
self.exc = exc
self.side_effect = side_effect
self._headers = headers or {}
self._cookies = {}
@ -270,3 +292,31 @@ def mock_aiohttp_client():
side_effect=create_session,
):
yield mocker
class MockLongPollSideEffect:
"""Imitate a long_poll request. Once created, actual responses are queued and if queue is empty, will await until done."""
def __init__(self):
"""Initialize the queue."""
self.semaphore = asyncio.Semaphore(0)
self.response_list = []
self.stopping = False
async def __call__(self, method, url, data):
"""Fetch the next response from the queue or wait until the queue has items."""
if self.stopping:
raise ClientError()
await self.semaphore.acquire()
kwargs = self.response_list.pop(0)
return AiohttpClientMockResponse(method=method, url=url, **kwargs)
def queue_response(self, **kwargs):
"""Add a response to the long_poll queue."""
self.response_list.append(kwargs)
self.semaphore.release()
def stop(self):
"""Stop the current request and future ones. Avoids exception if there is someone waiting when exiting test."""
self.stopping = True
self.queue_response(exc=ClientError())