From 05abf1405d3be9b686dde979f825d2a3cbdd36ad Mon Sep 17 00:00:00 2001 From: Erik Montnemery Date: Mon, 6 Sep 2021 13:24:00 +0200 Subject: [PATCH] Migrate emulated_hue tests from unittest to pytest (#55794) * Migrate emulated_hue tests from unittest to pytest * Remove unused variables --- tests/components/emulated_hue/test_upnp.py | 230 ++++++++++----------- 1 file changed, 114 insertions(+), 116 deletions(-) diff --git a/tests/components/emulated_hue/test_upnp.py b/tests/components/emulated_hue/test_upnp.py index e68688399e0..8ea65380359 100644 --- a/tests/components/emulated_hue/test_upnp.py +++ b/tests/components/emulated_hue/test_upnp.py @@ -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