Fix emulated_hue compatibility with older devices (#36090)

* Fix emulated_hue compatibility with older devices

* Fix test ugliness

* Fix pylint errors

homeassistant/components/emulated_hue/const.py

@@ -0,0 +1,4 @@
+"""Constants for emulated_hue."""
+
+HUE_SERIAL_NUMBER = "001788FFFE23BFC2"
+HUE_UUID = "2f402f80-da50-11e1-9b23-001788255acc"

homeassistant/components/emulated_hue/upnp.py

@@ -9,6 +9,8 @@ from aiohttp import web
 from homeassistant import core
 from homeassistant.components.http import HomeAssistantView
 
+from .const import HUE_SERIAL_NUMBER, HUE_UUID
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -42,8 +44,8 @@ class DescriptionXmlView(HomeAssistantView):
 <modelName>Philips hue bridge 2015</modelName>
 <modelNumber>BSB002</modelNumber>
 <modelURL>http://www.meethue.com</modelURL>
-<serialNumber>001788FFFE23BFC2</serialNumber>
+<serialNumber>{HUE_SERIAL_NUMBER}</serialNumber>
-<UDN>uuid:2f402f80-da50-11e1-9b23-001788255acc</UDN>
+<UDN>uuid:{HUE_UUID}</UDN>
 </device>
 </root>
 """
@@ -70,21 +72,8 @@ class UPNPResponderThread(threading.Thread):
         self.host_ip_addr = host_ip_addr
         self.listen_port = listen_port
         self.upnp_bind_multicast = upnp_bind_multicast
-
+        self.advertise_ip = advertise_ip
-        # Note that the double newline at the end of
+        self.advertise_port = advertise_port
-        # this string is required per the SSDP spec
-        resp_template = f"""HTTP/1.1 200 OK
-CACHE-CONTROL: max-age=60
-EXT:
-LOCATION: http://{advertise_ip}:{advertise_port}/description.xml
-SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
-hue-bridgeid: 001788FFFE23BFC2
-ST: upnp:rootdevice
-USN: uuid:2f402f80-da50-11e1-9b23-00178829d301::upnp:rootdevice
-
-"""
-
-        self.upnp_response = resp_template.replace("\n", "\r\n").encode("utf-8")
 
     def run(self):
         """Run the server."""
@@ -136,10 +125,13 @@ USN: uuid:2f402f80-da50-11e1-9b23-00178829d301::upnp:rootdevice
                 continue
 
             if "M-SEARCH" in data.decode("utf-8", errors="ignore"):
+                _LOGGER.debug("UPNP Responder M-SEARCH method received: %s", data)
                 # SSDP M-SEARCH method received, respond to it with our info
-                resp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                response = self._handle_request(data)
 
-                resp_socket.sendto(self.upnp_response, addr)
+                resp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                resp_socket.sendto(response, addr)
+                _LOGGER.debug("UPNP Responder responding with: %s", response)
                 resp_socket.close()
 
     def stop(self):
@@ -148,6 +140,31 @@ USN: uuid:2f402f80-da50-11e1-9b23-00178829d301::upnp:rootdevice
         self._interrupted = True
         self.join()
 
+    def _handle_request(self, data):
+        if "upnp:rootdevice" in data.decode("utf-8", errors="ignore"):
+            return self._prepare_response(
+                "upnp:rootdevice", f"uuid:{HUE_UUID}::upnp:rootdevice"
+            )
+
+        return self._prepare_response(
+            "urn:schemas-upnp-org:device:basic:1", f"uuid:{HUE_UUID}"
+        )
+
+    def _prepare_response(self, search_target, unique_service_name):
+        # Note that the double newline at the end of
+        # this string is required per the SSDP spec
+        response = f"""HTTP/1.1 200 OK
+CACHE-CONTROL: max-age=60
+EXT:
+LOCATION: http://{self.advertise_ip}:{self.advertise_port}/description.xml
+SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
+hue-bridgeid: {HUE_SERIAL_NUMBER}
+ST: {search_target}
+USN: {unique_service_name}
+
+"""
+        return response.replace("\n", "\r\n").encode("utf-8")
+
 
 def clean_socket_close(sock):
     """Close a socket connection and logs its closure."""

tests/components/emulated_hue/test_upnp.py

@@ -48,6 +48,66 @@ class TestEmulatedHue(unittest.TestCase):
         """Stop the class."""
         cls.hass.stop()
 
+    def test_upnp_discovery_basic(self):
+        """Tests the UPnP basic discovery response."""
+        with patch("threading.Thread.__init__"):
+            upnp_responder_thread = emulated_hue.UPNPResponderThread(
+                "0.0.0.0", 80, True, "192.0.2.42", 8080
+            )
+
+            """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")
+
+            response = upnp_responder_thread._handle_request(encoded_request)
+            expected_response = """HTTP/1.1 200 OK
+CACHE-CONTROL: max-age=60
+EXT:
+LOCATION: http://192.0.2.42:8080/description.xml
+SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
+hue-bridgeid: 001788FFFE23BFC2
+ST: urn:schemas-upnp-org:device:basic:1
+USN: uuid:2f402f80-da50-11e1-9b23-001788255acc
+
+"""
+            assert expected_response.replace("\n", "\r\n").encode("utf-8") == response
+
+    def test_upnp_discovery_rootdevice(self):
+        """Tests the UPnP rootdevice discovery response."""
+        with patch("threading.Thread.__init__"):
+            upnp_responder_thread = emulated_hue.UPNPResponderThread(
+                "0.0.0.0", 80, True, "192.0.2.42", 8080
+            )
+
+            """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")
+
+            response = upnp_responder_thread._handle_request(encoded_request)
+            expected_response = """HTTP/1.1 200 OK
+CACHE-CONTROL: max-age=60
+EXT:
+LOCATION: http://192.0.2.42:8080/description.xml
+SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/1.16.0
+hue-bridgeid: 001788FFFE23BFC2
+ST: upnp:rootdevice
+USN: uuid:2f402f80-da50-11e1-9b23-001788255acc::upnp:rootdevice
+
+"""
+            assert expected_response.replace("\n", "\r\n").encode("utf-8") == response
+
     def test_description_xml(self):
         """Test the description."""
         result = requests.get(BRIDGE_URL_BASE.format("/description.xml"), timeout=5)