tests

README.md

@@ -1,4 +1,4 @@
-[![codecov](https://codecov.io/gh/lbryio/txupnp/branch/master/graph/badge.svg)](https://codecov.io/gh/lbryio/txupnp)
+[![codecov](https://codecov.io/gh/lbryio/aioupnp/branch/master/graph/badge.svg)](https://codecov.io/gh/lbryio/aioupnp)
 
 # UPnP for asyncio
 
@@ -32,4 +32,3 @@ This project is MIT licensed. For the full license, see [LICENSE](LICENSE).
 ## Contact
 
 The primary contact for this project is [@jackrobison](mailto:jack@lbry.io)
-

aioupnp/serialization/test_scpd.py

@@ -0,0 +1,11 @@
+import unittest
+from aioupnp.serialization.scpd import serialize_scpd_get
+
+
+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'
+
+    def test_serialize_get(self):
+        self.assertEqual(serialize_scpd_get(self.path, self.lan_address), self.expected_result)

aioupnp/serialization/test_soap.py

@@ -0,0 +1,22 @@
+import unittest
+from aioupnp.serialization.soap import serialize_soap_post
+
+
+class TestSOAPSerialization(unittest.TestCase):
+    method, param_names, gateway_address, kwargs = "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'
+
+
+    def test_serialize_get(self):
+        self.assertEqual(serialize_soap_post(
+            self.method, self.param_names, self.st, self.gateway_address, self.path, **self.kwargs
+        ), self.expected_result)

aioupnp/serialization/test_ssdp.py

@@ -67,6 +67,14 @@ class TestParseMSearchResponseCaseInsensitive(TestParseMSearchResponse):
         'USN: uuid:00000000-0000-0000-0000-000000000000::urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1'
     ]).encode()
 
+    def test_get_case_insensitive(self):
+        packet = SSDPDatagram.decode(self.datagram)
+        self.assertEqual('max-age=1800', packet['Cache_Control'])
+
+    def test_key_error(self):
+        packet = SSDPDatagram.decode(self.datagram)
+        self.assertRaises(KeyError, lambda : packet['Cache Control'])
+
 
 class TestFailToParseMSearchResponseNoST(unittest.TestCase):
     datagram = "\r\n".join([

setup.py

@@ -21,7 +21,7 @@ setup(
     long_description=long_description,
     url="https://github.com/lbryio/aioupnp",
     license=__license__,
-    packages=find_packages(exclude=['tests']),
+    packages=find_packages(),
     entry_points={'console_scripts': console_scripts},
     install_requires=[
         'netifaces',

tests/__init__.py

@@ -1,7 +0,0 @@
-import logging
-
-log = logging.getLogger("aioupnp")
-handler = logging.StreamHandler()
-handler.setFormatter(logging.Formatter('%(asctime)-15s-%(filename)s:%(lineno)s->%(message)s'))
-log.addHandler(handler)
-log.setLevel(logging.INFO)

tests/test_devices.py

@@ -1,117 +0,0 @@
-# from twisted.internet import reactor, defer
-# from twisted.trial import unittest
-# from aioupnp.constants import SSDP_PORT, SSDP_IP_ADDRESS
-# from aioupnp.upnp import UPnP
-# from aioupnp.mocks import MockReactor, MockSSDPServiceGatewayProtocol, get_device_test_case
-#
-#
-# class TestDevice(unittest.TestCase):
-#     manufacturer, model = "Cisco", "CGA4131COM"
-#
-#     device = get_device_test_case(manufacturer, model)
-#     router_address = device.device_dict['router_address']
-#     client_address = device.device_dict['client_address']
-#     expected_devices = device.device_dict['expected_devices']
-#     packets_rx = device.device_dict['ssdp']['received']
-#     packets_tx = device.device_dict['ssdp']['sent']
-#     expected_available_commands = device.device_dict['commands']['available']
-#     scdp_packets = device.device_dict['scpd']
-#
-#     def setUp(self):
-#         fake_reactor = MockReactor(self.client_address, self.scdp_packets)
-#         reactor.listenMulticast = fake_reactor.listenMulticast
-#         self.reactor = reactor
-#         server_protocol = MockSSDPServiceGatewayProtocol(
-#             self.client_address, self.router_address, self.packets_rx, self.packets_tx
-#         )
-#         self.server_port = self.reactor.listenMulticast(SSDP_PORT, server_protocol, interface=self.router_address)
-#         self.server_port.transport.joinGroup(SSDP_IP_ADDRESS, interface=self.router_address)
-#
-#         self.upnp = UPnP(
-#             self.reactor, debug_ssdp=True, router_ip=self.router_address,
-#             lan_ip=self.client_address, iface_name='mock'
-#         )
-#
-#     def tearDown(self):
-#         self.upnp.sspd_factory.disconnect()
-#         self.server_port.stopListening()
-#
-#
-# class TestSSDP(TestDevice):
-#     @defer.inlineCallbacks
-#     def test_discover_device(self):
-#         result = yield self.upnp.m_search(self.router_address, timeout=1)
-#         self.assertEqual(len(self.expected_devices), len(result))
-#         self.assertEqual(len(result), 1)
-#         self.assertDictEqual(self.expected_devices[0], result[0])
-#
-#
-# class TestSCPD(TestDevice):
-#     @defer.inlineCallbacks
-#     def setUp(self):
-#         fake_reactor = MockReactor(self.client_address, self.scdp_packets)
-#         reactor.listenMulticast = fake_reactor.listenMulticast
-#         reactor.connectTCP = fake_reactor.connectTCP
-#         self.reactor = reactor
-#         server_protocol = MockSSDPServiceGatewayProtocol(
-#             self.client_address, self.router_address, self.packets_rx, self.packets_tx
-#         )
-#         self.server_port = self.reactor.listenMulticast(SSDP_PORT, server_protocol, interface=self.router_address)
-#         self.server_port.transport.joinGroup(SSDP_IP_ADDRESS, interface=self.router_address)
-#
-#         self.upnp = UPnP(
-#             self.reactor, debug_ssdp=True, router_ip=self.router_address,
-#             lan_ip=self.client_address, iface_name='mock'
-#         )
-#         yield self.upnp.discover()
-#
-#     def test_parse_available_commands(self):
-#         self.assertDictEqual(self.expected_available_commands, self.upnp.gateway.debug_commands()['available'])
-#
-#     def test_parse_gateway(self):
-#         self.assertDictEqual(self.device.device_dict['gateway_dict'], self.upnp.gateway.as_dict())
-#
-#     @defer.inlineCallbacks
-#     def test_commands(self):
-#         method, args, expected = self.device.device_dict['soap'][0]
-#         command1 = getattr(self.upnp, method)
-#         result = yield command1(*tuple(args))
-#         self.assertEqual(result, expected)
-#
-#         method, args, expected = self.device.device_dict['soap'][1]
-#         command2 = getattr(self.upnp, method)
-#         result = yield command2(*tuple(args))
-#         result = [[i for i in r] for r in result]
-#         self.assertListEqual(result, expected)
-#
-#         method, args, expected = self.device.device_dict['soap'][2]
-#         command3 = getattr(self.upnp, method)
-#         result = yield command3(*tuple(args))
-#         self.assertEqual(result, expected)
-#
-#         method, args, expected = self.device.device_dict['soap'][3]
-#         command4 = getattr(self.upnp, method)
-#         result = yield command4(*tuple(args))
-#         result = [r for r in result]
-#         self.assertEqual(result, expected)
-#
-#         method, args, expected = self.device.device_dict['soap'][4]
-#         command5 = getattr(self.upnp, method)
-#         result = yield command5(*tuple(args))
-#         self.assertEqual(result, expected)
-#
-#
-# class TestDDWRTSSDP(TestSSDP):
-#     manufacturer, model = "DD-WRT", "router"
-#
-#
-# class TestDDWRTSCPD(TestSCPD):
-#     manufacturer, model = "DD-WRT", "router"
-#
-#
-# class TestMiniUPnPMiniUPnPd(TestSSDP):
-#     manufacturer, model = "MiniUPnP", "MiniUPnPd"
-#
-#
-# class TestMiniUPnPMiniUPnPdSCPD(TestSCPD):
-#     manufacturer, model = "MiniUPnP", "MiniUPnPd"