LBRYCommunity/aioupnp
1
0
Fork 0
Code Issues 7 Pull requests 1 Projects Releases 1 Packages Wiki Activity

fix regex

Browse source
This commit is contained in:
Jack Robison 2018-10-08 19:43:29 -04:00
parent e98b3a240b
commit c5ef4d4b18
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 31 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
aioupnp/serialization/ssdp.py
View file

@ -7,17 +7,21 @@ from aioupnp.constants import line_separator
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
_template = "^(?i)(%s):[ ]*(.*)$"
ssdp_datagram_patterns = { ssdp_datagram_patterns = {
'host': (re.compile("^(?i)(host):(.*)$"), str), 'host': (re.compile("^(?i)(host):(.*)$"), str),
'st': (re.compile("^(?i)(st):(.*)$"), str), 'st': (re.compile(_template % 'st'), str),
'man': (re.compile("^(?i)(man):|(\"(.*)\")$"), str), 'man': (re.compile(_template % 'man'), str),
'mx': (re.compile("^(?i)(mx):(.*)$"), int), 'mx': (re.compile(_template % 'mx'), int),
'nt': (re.compile("^(?i)(nt):(.*)$"), str), 'nt': (re.compile(_template % 'nt'), str),
'nts': (re.compile("^(?i)(nts):(.*)$"), str), 'nts': (re.compile(_template % 'nts'), str),
'usn': (re.compile("^(?i)(usn):(.*)$"), str), 'usn': (re.compile(_template % 'usn'), str),
'location': (re.compile("^(?i)(location):(.*)$"), str), 'location': (re.compile(_template % 'location'), str),
'cache_control': (re.compile("^(?i)(cache[-|_]control):(.*)$"), str), 'cache_control': (re.compile(_template % 'cache[-|_]control'), str),
'server': (re.compile("^(?i)(server):(.*)$"), str), 'server': (re.compile(_template % 'server'), str),
} }
vendor_pattern = re.compile("^([\w|\d]*)\.([\w|\d]*\.com):([ \"|\w|\d\:]*)$") vendor_pattern = re.compile("^([\w|\d]*)\.([\w|\d]*\.com):([ \"|\w|\d\:]*)$")

19
aioupnp/serialization/test_ssdp.py
View file

@ -4,7 +4,7 @@ from aioupnp.fault import UPnPError
from aioupnp.constants import UPNP_ORG_IGD from aioupnp.constants import UPNP_ORG_IGD
class TestParseMSearchRequest(unittest.TestCase): class TestParseMSearchRequestWithQuotes(unittest.TestCase):
datagram = b'M-SEARCH * HTTP/1.1\r\n' \ datagram = b'M-SEARCH * HTTP/1.1\r\n' \
b'HOST: 239.255.255.250:1900\r\n' \ b'HOST: 239.255.255.250:1900\r\n' \
b'ST: urn:schemas-upnp-org:device:InternetGatewayDevice:1\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'MX: 1\r\n' \
b'\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): def test_parse_m_search(self):
packet = SSDPDatagram.decode(self.datagram) packet = SSDPDatagram.decode(self.datagram)
self.assertTrue(packet._packet_type, packet._M_SEARCH) self.assertTrue(packet._packet_type, packet._M_SEARCH)