Fix emulated_hue compatibility with older devices (#36090)

* Fix emulated_hue compatibility with older devices

* Fix test ugliness

* Fix pylint errors
pull/36155/head
Thomas Hollstegge 2020-05-25 21:55:23 +02:00 committed by Franck Nijhof
parent 7e90d4dd7b
commit 0f7ea290ca
No known key found for this signature in database
GPG Key ID: D62583BA8AB11CA3
3 changed files with 100 additions and 19 deletions

View File

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

View File

@ -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>
<UDN>uuid:2f402f80-da50-11e1-9b23-001788255acc</UDN>
<serialNumber>{HUE_SERIAL_NUMBER}</serialNumber>
<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
# Note that the double newline at the end of
# 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")
self.advertise_ip = advertise_ip
self.advertise_port = advertise_port
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."""

View File

@ -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)