diff --git a/aioupnp/serialization/ssdp.py b/aioupnp/serialization/ssdp.py
index 04a6fac..a715d4e 100644
--- a/aioupnp/serialization/ssdp.py
+++ b/aioupnp/serialization/ssdp.py
@@ -9,11 +9,11 @@ from aioupnp.constants import line_separator
log = logging.getLogger(__name__)
-_template = "^(?i)(%s):[ ]*(.*)$"
+_template = "(?i)^(%s):[ ]*(.*)$"
ssdp_datagram_patterns = {
- 'host': (re.compile("^(?i)(host):(.*)$"), str),
+ 'host': (re.compile("(?i)^(host):(.*)$"), str),
'st': (re.compile(_template % 'st'), str),
'man': (re.compile(_template % 'man'), str),
'mx': (re.compile(_template % 'mx'), int),
@@ -97,7 +97,7 @@ class SSDPDatagram(object):
self.ext = None
for k, v in kwargs.items():
normalized = k.lower().replace("-", "_")
- if not normalized.startswith("_") and hasattr(self, normalized) and getattr(self,normalized) is None:
+ if not normalized.startswith("_") and hasattr(self, normalized) and getattr(self, normalized) is None:
setattr(self, normalized, v)
self._case_mappings: dict = {k.lower(): k for k in kwargs.keys()}
for k in self._required_fields[self._packet_type]:
diff --git a/aioupnp/serialization/test_scpd.py b/aioupnp/serialization/test_scpd.py
index 2ce4288..efbae77 100644
--- a/aioupnp/serialization/test_scpd.py
+++ b/aioupnp/serialization/test_scpd.py
@@ -1,11 +1,100 @@
import unittest
-from aioupnp.serialization.scpd import serialize_scpd_get
+from aioupnp.serialization.scpd import serialize_scpd_get, deserialize_scpd_get_response
class TestSCPDSerialization(unittest.TestCase):
- path, lan_address = '/InternetGatewayDevice.xml', '10.0.0.1'
- expected_result = b'GET /InternetGatewayDevice.xml HTTP/1.1\r\n' \
- b'Accept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n'
+ path, lan_address = '/IGDdevicedesc_brlan0.xml', '10.1.10.1'
+ get_request = b'GET /IGDdevicedesc_brlan0.xml HTTP/1.1\r\n' \
+ b'Accept-Encoding: gzip\r\nHost: 10.1.10.1\r\nConnection: Close\r\n\r\n'
+
+ response = b"HTTP/1.1 200 OK\r\n" \
+ b"CONTENT-LENGTH: 2972\r\n" \
+ b"CONTENT-TYPE: text/xml\r\n" \
+ b"DATE: Thu, 18 Oct 2018 01:20:23 GMT\r\n" \
+ b"LAST-MODIFIED: Fri, 28 Sep 2018 18:35:48 GMT\r\n" \
+ b"SERVER: Linux/3.14.28-Prod_17.2, UPnP/1.0, Portable SDK for UPnP devices/1.6.22\r\n" \
+ b"X-User-Agent: redsonic\r\n" \
+ b"CONNECTION: close\r\n" \
+ b"\r\n" \
+ b"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp-org:device-1-0\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevice:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:ebf5a0a0-1dd1-11b2-a92f-603d266f9915</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n <device>\n <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n <friendlyName>WANConnectionDevice:1</friendlyName>\n <manufacturer>Cisco</manufacturer>\n <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n <modelDescription>CGA4131COM</modelDescription>\n <modelName>CGA4131COM</modelName>\n <modelNumber>CGA4131COM</modelNumber>\n <modelURL>http://www.cisco.com</modelURL>\n <serialNumber></serialNumber>\n <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n <UPC>CGA4131COM</UPC>\n <serviceList>\n <service>\n <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n <controlURL>/upnp/control/WANIPConnection0</controlURL>\n <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n </service>\n </serviceList>\n </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n"
+
+ expected_parsed = {
+ 'specVersion': {'major': '1', 'minor': '0'},
+ 'device': {
+ 'deviceType': 'urn:schemas-upnp-org:device:InternetGatewayDevice:1',
+ 'friendlyName': 'CGA4131COM',
+ 'manufacturer': 'Cisco',
+ 'manufacturerURL': 'http://www.cisco.com/',
+ 'modelDescription': 'CGA4131COM',
+ 'modelName': 'CGA4131COM',
+ 'modelNumber': 'CGA4131COM',
+ 'modelURL': 'http://www.cisco.com',
+ 'UDN': 'uuid:11111111-2222-3333-4444-555555555556',
+ 'UPC': 'CGA4131COM',
+ 'serviceList': {
+ 'service': {
+ 'serviceType': 'urn:schemas-upnp-org:service:Layer3Forwarding:1',
+ 'serviceId': 'urn:upnp-org:serviceId:L3Forwarding1',
+ 'SCPDURL': '/Layer3ForwardingSCPD.xml',
+ 'controlURL': '/upnp/control/Layer3Forwarding',
+ 'eventSubURL': '/upnp/event/Layer3Forwarding'
+ }
+ },
+ 'deviceList': {
+ 'device': {
+ 'deviceType': 'urn:schemas-upnp-org:device:WANDevice:1',
+ 'friendlyName': 'WANDevice:1',
+ 'manufacturer': 'Cisco',
+ 'manufacturerURL': 'http://www.cisco.com/',
+ 'modelDescription': 'CGA4131COM',
+ 'modelName': 'CGA4131COM',
+ 'modelNumber': 'CGA4131COM',
+ 'modelURL': 'http://www.cisco.com',
+ 'UDN': 'uuid:ebf5a0a0-1dd1-11b2-a92f-603d266f9915',
+ 'UPC': 'CGA4131COM',
+ 'serviceList': {
+ 'service': {
+ 'serviceType': 'urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
+ 'serviceId': 'urn:upnp-org:serviceId:WANCommonIFC1',
+ 'SCPDURL': '/WANCommonInterfaceConfigSCPD.xml',
+ 'controlURL': '/upnp/control/WANCommonInterfaceConfig0',
+ 'eventSubURL': '/upnp/event/WANCommonInterfaceConfig0'
+ }
+ },
+ 'deviceList': {
+ 'device': {
+ 'deviceType': 'urn:schemas-upnp-org:device:WANConnectionDevice:1',
+ 'friendlyName': 'WANConnectionDevice:1',
+ 'manufacturer': 'Cisco',
+ 'manufacturerURL': 'http://www.cisco.com/',
+ 'modelDescription': 'CGA4131COM',
+ 'modelName': 'CGA4131COM',
+ 'modelNumber': 'CGA4131COM',
+ 'modelURL': 'http://www.cisco.com',
+ 'UDN': 'uuid:11111111-2222-3333-4444-555555555555',
+ 'UPC': 'CGA4131COM',
+ 'serviceList': {
+ 'service': {
+ 'serviceType': 'urn:schemas-upnp-org:service:WANIPConnection:1',
+ 'serviceId': 'urn:upnp-org:serviceId:WANIPConn1',
+ 'SCPDURL': '/WANIPConnectionServiceSCPD.xml',
+ 'controlURL': '/upnp/control/WANIPConnection0',
+ 'eventSubURL': '/upnp/event/WANIPConnection0'
+ }
+ }
+ }
+ }
+ }
+ },
+ 'presentationURL': 'http://10.1.10.1/'
+ }
+ }
def test_serialize_get(self):
- self.assertEqual(serialize_scpd_get(self.path, self.lan_address), self.expected_result)
+ self.assertEqual(serialize_scpd_get(self.path, self.lan_address), self.get_request)
+
+ def test_deserialize_get_response(self):
+ self.assertDictEqual(deserialize_scpd_get_response(self.response), self.expected_parsed)
+
+ def test_deserialize_blank(self):
+ self.assertDictEqual(deserialize_scpd_get_response(b''), {})
diff --git a/aioupnp/serialization/test_soap.py b/aioupnp/serialization/test_soap.py
index 5d8d261..487cf84 100644
--- a/aioupnp/serialization/test_soap.py
+++ b/aioupnp/serialization/test_soap.py
@@ -1,5 +1,5 @@
import unittest
-from aioupnp.serialization.soap import serialize_soap_post
+from aioupnp.serialization.soap import serialize_soap_post, deserialize_soap_post_response
class TestSOAPSerialization(unittest.TestCase):
@@ -7,17 +7,33 @@ class TestSOAPSerialization(unittest.TestCase):
kwargs: dict = {}
method, gateway_address = "GetExternalIPAddress", b'10.0.0.1'
st, lan_address, path = b'urn:schemas-upnp-org:service:WANIPConnection:1', '10.0.0.1', b'/soap.cgi?service=WANIPConn1'
- expected_result = b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\n' \
- b'Host: 10.0.0.1\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\n' \
- b'Content-Length: 285\r\nContent-Type: text/xml\r\n' \
- b'SOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\n' \
- b'Connection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n' \
- b'<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"' \
- b' s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">' \
- b'<s:Body><u:GetExternalIPAddress xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1">' \
- b'</u:GetExternalIPAddress></s:Body></s:Envelope>\r\n'
+ post_bytes = b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\n' \
+ b'Host: 10.0.0.1\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\n' \
+ b'Content-Length: 285\r\nContent-Type: text/xml\r\n' \
+ b'SOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\n' \
+ b'Connection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n' \
+ b'<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"' \
+ b' s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">' \
+ b'<s:Body><u:GetExternalIPAddress xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1">' \
+ b'</u:GetExternalIPAddress></s:Body></s:Envelope>\r\n'
- def test_serialize_get(self):
+ post_response = b"HTTP/1.1 200 OK\r\n" \
+ b"CONTENT-LENGTH: 340\r\n" \
+ b"CONTENT-TYPE: text/xml; charset=\"utf-8\"\r\n" \
+ b"DATE: Thu, 18 Oct 2018 01:20:23 GMT\r\n" \
+ b"EXT:\r\n" \
+ b"SERVER: Linux/3.14.28-Prod_17.2, UPnP/1.0, Portable SDK for UPnP devices/1.6.22\r\n" \
+ b"X-User-Agent: redsonic\r\n" \
+ b"\r\n" \
+ b"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\"><s:Body>\n<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\">\r\n<NewExternalIPAddress>11.22.33.44</NewExternalIPAddress>\r\n</u:GetExternalIPAddressResponse>\r\n</s:Body> </s:Envelope>"
+
+ def test_serialize_post(self):
self.assertEqual(serialize_soap_post(
self.method, self.param_names, self.st, self.gateway_address, self.path, **self.kwargs
- ), self.expected_result)
+ ), self.post_bytes)
+
+ def test_deserialize_post_response(self):
+ self.assertDictEqual(
+ deserialize_soap_post_response(self.post_response, self.method, service_id=self.st.decode()),
+ {'NewExternalIPAddress': '11.22.33.44'}
+ )
diff --git a/aioupnp/serialization/test_ssdp.py b/aioupnp/serialization/test_ssdp.py
index fa8094a..7a6ee01 100644
--- a/aioupnp/serialization/test_ssdp.py
+++ b/aioupnp/serialization/test_ssdp.py
@@ -2,7 +2,62 @@ import unittest
from collections import OrderedDict
from aioupnp.serialization.ssdp import SSDPDatagram
from aioupnp.fault import UPnPError
-from aioupnp.constants import UPNP_ORG_IGD, SSDP_DISCOVER
+from aioupnp.constants import UPNP_ORG_IGD
+
+
+class TestSSDPDatagram(unittest.TestCase):
+ def test_fail_to_init(self):
+ datagram_args = OrderedDict([
+ ('Host', "{}:{}".format('239.255.255.250', 1900)),
+ ('Man', '"ssdp:discover"'),
+ ('ST', 'ssdp:all'),
+ ('MX', 5),
+ ])
+
+ with self.assertRaises(UPnPError):
+ SSDPDatagram("?", datagram_args)
+
+ def test_fail_to_decode_missing_required(self):
+ packet = \
+ b'M-SEARCH * HTTP/1.1\r\n' \
+ b'Host: 239.255.255.250:1900\r\n' \
+ b'ST: ssdp:all\r\n' \
+ b'MX: 5\r\n' \
+ b'\r\n'
+
+ with self.assertRaises(UPnPError):
+ SSDPDatagram.decode(packet)
+
+ def test_cli_args(self):
+ datagram_args = OrderedDict([
+ ('Host', "{}:{}".format('239.255.255.250', 1900)),
+ ('Man', '"ssdp:discover"'),
+ ('ST', 'ssdp:all'),
+ ('MX', 5),
+ ])
+ packet = SSDPDatagram("M-SEARCH", datagram_args)
+ self.assertEqual(
+ packet.get_cli_igd_kwargs(),
+ '--Host=239.255.255.250:1900 --Man="ssdp:discover" --ST=ssdp:all --MX=5'
+ )
+
+ def test_as_dict(self):
+ datagram_args = OrderedDict([
+ ('Host', "{}:{}".format('239.255.255.250', 1900)),
+ ('Man', '"ssdp:discover"'),
+ ('ST', 'ssdp:all'),
+ ('MX', 5),
+ ])
+ packet = SSDPDatagram("M-SEARCH", datagram_args)
+ self.assertDictEqual(
+ packet.as_dict(),
+ {
+ "Host": "239.255.255.250:1900",
+ "Man": "\"ssdp:discover\"",
+ "ST": "ssdp:all",
+ "MX": 5
+ }
+ )
class TestMSearchDatagramSerialization(unittest.TestCase):