From c5ef4d4b186b9853ac85df9ddae9ed1466b5281d Mon Sep 17 00:00:00 2001 From: Jack Robison Date: Mon, 8 Oct 2018 19:43:29 -0400 Subject: [PATCH] fix regex --- aioupnp/serialization/ssdp.py | 22 +++++++++++++--------- aioupnp/serialization/test_ssdp.py | 19 ++++++++++++++++++- 2 files changed, 31 insertions(+), 10 deletions(-) diff --git a/aioupnp/serialization/ssdp.py b/aioupnp/serialization/ssdp.py index a65d689..6ffd39f 100644 --- a/aioupnp/serialization/ssdp.py +++ b/aioupnp/serialization/ssdp.py @@ -7,17 +7,21 @@ from aioupnp.constants import line_separator log = logging.getLogger(__name__) + +_template = "^(?i)(%s):[ ]*(.*)$" + + ssdp_datagram_patterns = { 'host': (re.compile("^(?i)(host):(.*)$"), str), - 'st': (re.compile("^(?i)(st):(.*)$"), str), - 'man': (re.compile("^(?i)(man):|(\"(.*)\")$"), str), - 'mx': (re.compile("^(?i)(mx):(.*)$"), int), - 'nt': (re.compile("^(?i)(nt):(.*)$"), str), - 'nts': (re.compile("^(?i)(nts):(.*)$"), str), - 'usn': (re.compile("^(?i)(usn):(.*)$"), str), - 'location': (re.compile("^(?i)(location):(.*)$"), str), - 'cache_control': (re.compile("^(?i)(cache[-|_]control):(.*)$"), str), - 'server': (re.compile("^(?i)(server):(.*)$"), str), + 'st': (re.compile(_template % 'st'), str), + 'man': (re.compile(_template % 'man'), str), + 'mx': (re.compile(_template % 'mx'), int), + 'nt': (re.compile(_template % 'nt'), str), + 'nts': (re.compile(_template % 'nts'), str), + 'usn': (re.compile(_template % 'usn'), str), + 'location': (re.compile(_template % 'location'), str), + 'cache_control': (re.compile(_template % 'cache[-|_]control'), str), + 'server': (re.compile(_template % 'server'), str), } vendor_pattern = re.compile("^([\w|\d]*)\.([\w|\d]*\.com):([ \"|\w|\d\:]*)$") diff --git a/aioupnp/serialization/test_ssdp.py b/aioupnp/serialization/test_ssdp.py index 324feed..ed35ba8 100644 --- a/aioupnp/serialization/test_ssdp.py +++ b/aioupnp/serialization/test_ssdp.py @@ -4,7 +4,7 @@ from aioupnp.fault import UPnPError from aioupnp.constants import UPNP_ORG_IGD -class TestParseMSearchRequest(unittest.TestCase): +class TestParseMSearchRequestWithQuotes(unittest.TestCase): datagram = b'M-SEARCH * HTTP/1.1\r\n' \ b'HOST: 239.255.255.250:1900\r\n' \ b'ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1\r\n' \ @@ -12,6 +12,23 @@ class TestParseMSearchRequest(unittest.TestCase): b'MX: 1\r\n' \ b'\r\n' + def test_parse_m_search(self): + packet = SSDPDatagram.decode(self.datagram) + self.assertTrue(packet._packet_type, packet._M_SEARCH) + self.assertEqual(packet.host, '239.255.255.250:1900') + self.assertEqual(packet.st, 'urn:schemas-upnp-org:device:InternetGatewayDevice:1') + self.assertEqual(packet.man, '"ssdp:discover"') + self.assertEqual(packet.mx, 1) + + +class TestParseMSearchRequestWithoutQuotes(unittest.TestCase): + datagram = b'M-SEARCH * HTTP/1.1\r\n' \ + b'HOST: 239.255.255.250:1900\r\n' \ + b'ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1\r\n' \ + b'MAN: ssdp:discover\r\n' \ + b'MX: 1\r\n' \ + b'\r\n' + def test_parse_m_search(self): packet = SSDPDatagram.decode(self.datagram) self.assertTrue(packet._packet_type, packet._M_SEARCH)