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

ssdp datagram unit tests

Browse source
This commit is contained in:
Jack Robison 2018-09-25 14:53:34 -04:00
parent ad05cbb457
commit d24e5c1a7b
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
2 changed files with 107 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

101
tests/test_ssdp_datagram.py Normal file
View file

@ -0,0 +1,101 @@
from twisted.trial import unittest
from txupnp.ssdp_datagram import SSDPDatagram
from txupnp.fault import UPnPError
class TestParseMSearchRequest(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_response(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)
def test_serialize_m_search(self):
packet = SSDPDatagram.decode(self.datagram)
self.assertEqual(packet.encode().encode(), self.datagram)
class TestParseMSearchResponse(unittest.TestCase):
datagram = "\r\n".join([
'HTTP/1.1 200 OK',
'CACHE_CONTROL: max-age=1800',
'LOCATION: http://10.0.0.1:49152/InternetGatewayDevice.xml',
'SERVER: Linux, UPnP/1.0, DIR-890L Ver 1.20',
'ST: urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
]).encode()
def test_parse_m_search_response(self):
packet = SSDPDatagram.decode(self.datagram)
self.assertTrue(packet._packet_type, packet._OK)
self.assertEqual(packet.cache_control, 'max-age=1800')
self.assertEqual(packet.location, 'http://10.0.0.1:49152/InternetGatewayDevice.xml')
self.assertEqual(packet.server, 'Linux, UPnP/1.0, DIR-890L Ver 1.20')
self.assertEqual(packet.st, 'urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1')
self.assertEqual(packet.usn, 'uuid:00000000-0000-0000-0000-000000000000::urn:'
'schemas-upnp-org:service:WANCommonInterfaceConfig:1'
)
class TestParseMSearchResponseDashCacheControl(TestParseMSearchResponse):
datagram = "\r\n".join([
'HTTP/1.1 200 OK',
'CACHE-CONTROL: max-age=1800',
'LOCATION: http://10.0.0.1:49152/InternetGatewayDevice.xml',
'SERVER: Linux, UPnP/1.0, DIR-890L Ver 1.20',
'ST: urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
]).encode()
class TestParseMSearchResponseCaseInsensitive(TestParseMSearchResponse):
datagram = "\r\n".join([
'HTTP/1.1 200 OK',
'cache-CONTROL: max-age=1800',
'LOCATION: http://10.0.0.1:49152/InternetGatewayDevice.xml',
'Server: Linux, UPnP/1.0, DIR-890L Ver 1.20',
'st: urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
]).encode()
class TestFailToParseMSearchResponseNoST(unittest.TestCase):
datagram = "\r\n".join([
'HTTP/1.1 200 OK',
'CACHE_CONTROL: max-age=1800',
'LOCATION: http://10.0.0.1:49152/InternetGatewayDevice.xml',
'SERVER: Linux, UPnP/1.0, DIR-890L Ver 1.20',
'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
]).encode()
def test_fail_to_parse_m_search_response(self):
self.assertRaises(UPnPError, SSDPDatagram.decode, self.datagram)
class TestFailToParseMSearchResponseNoOK(TestFailToParseMSearchResponseNoST):
datagram = "\r\n".join([
'cache-CONTROL: max-age=1800',
'LOCATION: http://10.0.0.1:49152/InternetGatewayDevice.xml',
'Server: Linux, UPnP/1.0, DIR-890L Ver 1.20',
'st: urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
]).encode()
class TestFailToParseMSearchResponseNoLocation(TestFailToParseMSearchResponseNoST):
datagram = "\r\n".join([
'HTTP/1.1 200 OK',
'cache-CONTROL: max-age=1800',
'Server: Linux, UPnP/1.0, DIR-890L Ver 1.20',
'st: urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
]).encode()

8
txupnp/ssdp_datagram.py
View file

@ -15,7 +15,7 @@ _ssdp_datagram_patterns = {
'nts': (re.compile("^(?i)(nts):(.*)$"), str), 'nts': (re.compile("^(?i)(nts):(.*)$"), str),
'usn': (re.compile("^(?i)(usn):(.*)$"), str), 'usn': (re.compile("^(?i)(usn):(.*)$"), str),
'location': (re.compile("^(?i)(location):(.*)$"), str), 'location': (re.compile("^(?i)(location):(.*)$"), str),
'cache_control': (re.compile("^(?i)(cache-control):(.*)$"), str), 'cache_control': (re.compile("^(?i)(cache[-|_]control):(.*)$"), str),
'server': (re.compile("^(?i)(server):(.*)$"), str), 'server': (re.compile("^(?i)(server):(.*)$"), str),
} }
@ -124,8 +124,12 @@ class SSDPDatagram(object):
return self._lines_to_content_dict(self.encode().split(line_separator)) return self._lines_to_content_dict(self.encode().split(line_separator))
@classmethod @classmethod
def decode(cls, datagram): def decode(cls, datagram: bytes):
packet = cls._from_string(datagram.decode()) packet = cls._from_string(datagram.decode())
if packet is None:
raise UPnPError(
"failed to decode datagram: {}".format(binascii.hexlify(datagram))
)
for attr_name in packet._required_fields[packet._packet_type]: for attr_name in packet._required_fields[packet._packet_type]:
attr = getattr(packet, attr_name) attr = getattr(packet, attr_name)
if attr is None: if attr is None: