Migrate emulated_hue tests from unittest to pytest (#55794)
* Migrate emulated_hue tests from unittest to pytest * Remove unused variablespull/55839/head
Erik Montnemery
GitHub
parent
753285eae7
commit
05abf1405d
|
|
@ -2,23 +2,19 @@
|
|||
import json
|
||||
import unittest
|
||||
|
||||
from aiohttp.hdrs import CONTENT_TYPE
|
||||
from aiohttp import web
|
||||
import defusedxml.ElementTree as ET
|
||||
import requests
|
||||
import pytest
|
||||
|
||||
from homeassistant import const, setup
|
||||
from homeassistant import setup
|
||||
from homeassistant.components import emulated_hue
|
||||
from homeassistant.components.emulated_hue import upnp
|
||||
from homeassistant.const import CONTENT_TYPE_JSON, HTTP_OK
|
||||
|
||||
from tests.common import get_test_home_assistant, get_test_instance_port
|
||||
from tests.common import get_test_instance_port
|
||||
|
||||
HTTP_SERVER_PORT = get_test_instance_port()
|
||||
BRIDGE_SERVER_PORT = get_test_instance_port()
|
||||
|
||||
BRIDGE_URL_BASE = f"http://127.0.0.1:{BRIDGE_SERVER_PORT}" + "{}"
|
||||
JSON_HEADERS = {CONTENT_TYPE: const.CONTENT_TYPE_JSON}
|
||||
|
||||
|
||||
class MockTransport:
|
||||
"""Mock asyncio transport."""
|
||||
|
|
@ -32,49 +28,48 @@ class MockTransport:
|
|||
self.sends.append((response, addr))
|
||||
|
||||
|
||||
class TestEmulatedHue(unittest.TestCase):
|
||||
"""Test the emulated Hue component."""
|
||||
@pytest.fixture
|
||||
def hue_client(aiohttp_client):
|
||||
"""Return a hue API client."""
|
||||
app = web.Application()
|
||||
with unittest.mock.patch(
|
||||
"homeassistant.components.emulated_hue.web.Application", return_value=app
|
||||
):
|
||||
|
||||
hass = None
|
||||
async def client():
|
||||
"""Return an authenticated client."""
|
||||
return await aiohttp_client(app)
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Set up the class."""
|
||||
cls.hass = hass = get_test_home_assistant()
|
||||
yield client
|
||||
|
||||
setup.setup_component(
|
||||
hass,
|
||||
emulated_hue.DOMAIN,
|
||||
{emulated_hue.DOMAIN: {emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT}},
|
||||
)
|
||||
|
||||
cls.hass.start()
|
||||
async def setup_hue(hass):
|
||||
"""Set up the emulated_hue integration."""
|
||||
assert await setup.async_setup_component(
|
||||
hass,
|
||||
emulated_hue.DOMAIN,
|
||||
{emulated_hue.DOMAIN: {emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT}},
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
"""Stop the class."""
|
||||
cls.hass.stop()
|
||||
|
||||
def test_upnp_discovery_basic(self):
|
||||
"""Tests the UPnP basic discovery response."""
|
||||
upnp_responder_protocol = upnp.UPNPResponderProtocol(
|
||||
None, None, "192.0.2.42", 8080
|
||||
)
|
||||
mock_transport = MockTransport()
|
||||
upnp_responder_protocol.transport = mock_transport
|
||||
def test_upnp_discovery_basic():
|
||||
"""Tests the UPnP basic discovery response."""
|
||||
upnp_responder_protocol = upnp.UPNPResponderProtocol(None, None, "192.0.2.42", 8080)
|
||||
mock_transport = MockTransport()
|
||||
upnp_responder_protocol.transport = mock_transport
|
||||
|
||||
"""Original request emitted by the Hue Bridge v1 app."""
|
||||
request = """M-SEARCH * HTTP/1.1
|
||||
"""Original request emitted by the Hue Bridge v1 app."""
|
||||
request = """M-SEARCH * HTTP/1.1
|
||||
HOST:239.255.255.250:1900
|
||||
ST:ssdp:all
|
||||
Man:"ssdp:discover"
|
||||
MX:3
|
||||
|
||||
"""
|
||||
encoded_request = request.replace("\n", "\r\n").encode("utf-8")
|
||||
encoded_request = request.replace("\n", "\r\n").encode("utf-8")
|
||||
|
||||
upnp_responder_protocol.datagram_received(encoded_request, 1234)
|
||||
expected_response = """HTTP/1.1 200 OK
|
||||
upnp_responder_protocol.datagram_received(encoded_request, 1234)
|
||||
expected_response = """HTTP/1.1 200 OK
|
||||
CACHE-CONTROL: max-age=60
|
||||
EXT:
|
||||
LOCATION: http://192.0.2.42:8080/description.xml
|
||||
|
|
@ -84,30 +79,29 @@ ST: urn:schemas-upnp-org:device:basic:1
|
|||
USN: uuid:2f402f80-da50-11e1-9b23-001788255acc
|
||||
|
||||
"""
|
||||
expected_send = expected_response.replace("\n", "\r\n").encode("utf-8")
|
||||
expected_send = expected_response.replace("\n", "\r\n").encode("utf-8")
|
||||
|
||||
assert mock_transport.sends == [(expected_send, 1234)]
|
||||
assert mock_transport.sends == [(expected_send, 1234)]
|
||||
|
||||
def test_upnp_discovery_rootdevice(self):
|
||||
"""Tests the UPnP rootdevice discovery response."""
|
||||
upnp_responder_protocol = upnp.UPNPResponderProtocol(
|
||||
None, None, "192.0.2.42", 8080
|
||||
)
|
||||
mock_transport = MockTransport()
|
||||
upnp_responder_protocol.transport = mock_transport
|
||||
|
||||
"""Original request emitted by Busch-Jaeger free@home SysAP."""
|
||||
request = """M-SEARCH * HTTP/1.1
|
||||
def test_upnp_discovery_rootdevice():
|
||||
"""Tests the UPnP rootdevice discovery response."""
|
||||
upnp_responder_protocol = upnp.UPNPResponderProtocol(None, None, "192.0.2.42", 8080)
|
||||
mock_transport = MockTransport()
|
||||
upnp_responder_protocol.transport = mock_transport
|
||||
|
||||
"""Original request emitted by Busch-Jaeger free@home SysAP."""
|
||||
request = """M-SEARCH * HTTP/1.1
|
||||
HOST: 239.255.255.250:1900
|
||||
MAN: "ssdp:discover"
|
||||
MX: 40
|
||||
ST: upnp:rootdevice
|
||||
|
||||
"""
|
||||
encoded_request = request.replace("\n", "\r\n").encode("utf-8")
|
||||
encoded_request = request.replace("\n", "\r\n").encode("utf-8")
|
||||
|
||||
upnp_responder_protocol.datagram_received(encoded_request, 1234)
|
||||
expected_response = """HTTP/1.1 200 OK
|
||||
upnp_responder_protocol.datagram_received(encoded_request, 1234)
|
||||
expected_response = """HTTP/1.1 200 OK
|
||||
CACHE-CONTROL: max-age=60
|
||||
EXT:
|
||||
LOCATION: http://192.0.2.42:8080/description.xml
|
||||
|
|
@ -117,95 +111,99 @@ ST: upnp:rootdevice
|
|||
USN: uuid:2f402f80-da50-11e1-9b23-001788255acc::upnp:rootdevice
|
||||
|
||||
"""
|
||||
expected_send = expected_response.replace("\n", "\r\n").encode("utf-8")
|
||||
expected_send = expected_response.replace("\n", "\r\n").encode("utf-8")
|
||||
|
||||
assert mock_transport.sends == [(expected_send, 1234)]
|
||||
assert mock_transport.sends == [(expected_send, 1234)]
|
||||
|
||||
def test_upnp_no_response(self):
|
||||
"""Tests the UPnP does not response on an invalid request."""
|
||||
upnp_responder_protocol = upnp.UPNPResponderProtocol(
|
||||
None, None, "192.0.2.42", 8080
|
||||
)
|
||||
mock_transport = MockTransport()
|
||||
upnp_responder_protocol.transport = mock_transport
|
||||
|
||||
"""Original request emitted by the Hue Bridge v1 app."""
|
||||
request = """INVALID * HTTP/1.1
|
||||
def test_upnp_no_response():
|
||||
"""Tests the UPnP does not response on an invalid request."""
|
||||
upnp_responder_protocol = upnp.UPNPResponderProtocol(None, None, "192.0.2.42", 8080)
|
||||
mock_transport = MockTransport()
|
||||
upnp_responder_protocol.transport = mock_transport
|
||||
|
||||
"""Original request emitted by the Hue Bridge v1 app."""
|
||||
request = """INVALID * HTTP/1.1
|
||||
HOST:239.255.255.250:1900
|
||||
ST:ssdp:all
|
||||
Man:"ssdp:discover"
|
||||
MX:3
|
||||
|
||||
"""
|
||||
encoded_request = request.replace("\n", "\r\n").encode("utf-8")
|
||||
encoded_request = request.replace("\n", "\r\n").encode("utf-8")
|
||||
|
||||
upnp_responder_protocol.datagram_received(encoded_request, 1234)
|
||||
upnp_responder_protocol.datagram_received(encoded_request, 1234)
|
||||
|
||||
assert mock_transport.sends == []
|
||||
assert mock_transport.sends == []
|
||||
|
||||
def test_description_xml(self):
|
||||
"""Test the description."""
|
||||
result = requests.get(BRIDGE_URL_BASE.format("/description.xml"), timeout=5)
|
||||
|
||||
assert result.status_code == HTTP_OK
|
||||
assert "text/xml" in result.headers["content-type"]
|
||||
async def test_description_xml(hass, hue_client):
|
||||
"""Test the description."""
|
||||
await setup_hue(hass)
|
||||
client = await hue_client()
|
||||
result = await client.get("/description.xml", timeout=5)
|
||||
|
||||
# Make sure the XML is parsable
|
||||
try:
|
||||
root = ET.fromstring(result.text)
|
||||
ns = {"s": "urn:schemas-upnp-org:device-1-0"}
|
||||
assert root.find("./s:device/s:serialNumber", ns).text == "001788FFFE23BFC2"
|
||||
except: # noqa: E722 pylint: disable=bare-except
|
||||
self.fail("description.xml is not valid XML!")
|
||||
assert result.status == HTTP_OK
|
||||
assert "text/xml" in result.headers["content-type"]
|
||||
|
||||
def test_create_username(self):
|
||||
"""Test the creation of an username."""
|
||||
request_json = {"devicetype": "my_device"}
|
||||
try:
|
||||
root = ET.fromstring(await result.text())
|
||||
ns = {"s": "urn:schemas-upnp-org:device-1-0"}
|
||||
assert root.find("./s:device/s:serialNumber", ns).text == "001788FFFE23BFC2"
|
||||
except: # noqa: E722 pylint: disable=bare-except
|
||||
pytest.fail("description.xml is not valid XML!")
|
||||
|
||||
result = requests.post(
|
||||
BRIDGE_URL_BASE.format("/api"), data=json.dumps(request_json), timeout=5
|
||||
)
|
||||
|
||||
assert result.status_code == HTTP_OK
|
||||
assert CONTENT_TYPE_JSON in result.headers["content-type"]
|
||||
async def test_create_username(hass, hue_client):
|
||||
"""Test the creation of an username."""
|
||||
await setup_hue(hass)
|
||||
client = await hue_client()
|
||||
request_json = {"devicetype": "my_device"}
|
||||
|
||||
resp_json = result.json()
|
||||
success_json = resp_json[0]
|
||||
result = await client.post("/api", data=json.dumps(request_json), timeout=5)
|
||||
|
||||
assert "success" in success_json
|
||||
assert "username" in success_json["success"]
|
||||
assert result.status == HTTP_OK
|
||||
assert CONTENT_TYPE_JSON in result.headers["content-type"]
|
||||
|
||||
def test_unauthorized_view(self):
|
||||
"""Test unauthorized view."""
|
||||
request_json = {"devicetype": "my_device"}
|
||||
resp_json = await result.json()
|
||||
success_json = resp_json[0]
|
||||
|
||||
result = requests.get(
|
||||
BRIDGE_URL_BASE.format("/api/unauthorized"),
|
||||
data=json.dumps(request_json),
|
||||
timeout=5,
|
||||
)
|
||||
assert "success" in success_json
|
||||
assert "username" in success_json["success"]
|
||||
|
||||
assert result.status_code == HTTP_OK
|
||||
assert CONTENT_TYPE_JSON in result.headers["content-type"]
|
||||
|
||||
resp_json = result.json()
|
||||
assert len(resp_json) == 1
|
||||
success_json = resp_json[0]
|
||||
assert len(success_json) == 1
|
||||
async def test_unauthorized_view(hass, hue_client):
|
||||
"""Test unauthorized view."""
|
||||
await setup_hue(hass)
|
||||
client = await hue_client()
|
||||
request_json = {"devicetype": "my_device"}
|
||||
|
||||
assert "error" in success_json
|
||||
error_json = success_json["error"]
|
||||
assert len(error_json) == 3
|
||||
assert "/" in error_json["address"]
|
||||
assert "unauthorized user" in error_json["description"]
|
||||
assert "1" in error_json["type"]
|
||||
result = await client.get(
|
||||
"/api/unauthorized", data=json.dumps(request_json), timeout=5
|
||||
)
|
||||
|
||||
def test_valid_username_request(self):
|
||||
"""Test request with a valid username."""
|
||||
request_json = {"invalid_key": "my_device"}
|
||||
assert result.status == HTTP_OK
|
||||
assert CONTENT_TYPE_JSON in result.headers["content-type"]
|
||||
|
||||
result = requests.post(
|
||||
BRIDGE_URL_BASE.format("/api"), data=json.dumps(request_json), timeout=5
|
||||
)
|
||||
resp_json = await result.json()
|
||||
assert len(resp_json) == 1
|
||||
success_json = resp_json[0]
|
||||
assert len(success_json) == 1
|
||||
|
||||
assert result.status_code == 400
|
||||
assert "error" in success_json
|
||||
error_json = success_json["error"]
|
||||
assert len(error_json) == 3
|
||||
assert "/" in error_json["address"]
|
||||
assert "unauthorized user" in error_json["description"]
|
||||
assert "1" in error_json["type"]
|
||||
|
||||
|
||||
async def test_valid_username_request(hass, hue_client):
|
||||
"""Test request with a valid username."""
|
||||
await setup_hue(hass)
|
||||
client = await hue_client()
|
||||
request_json = {"invalid_key": "my_device"}
|
||||
|
||||
result = await client.post("/api", data=json.dumps(request_json), timeout=5)
|
||||
|
||||
assert result.status == 400
|
||||
|
|
|
|||
Loading…
Reference in New Issue