tests

aioupnp/interfaces.py

@@ -2,6 +2,7 @@ import socket
 from collections import OrderedDict
 import typing
 import netifaces
+from aioupnp.fault import UPnPError
 
 
 def get_netifaces():
@@ -50,4 +51,4 @@ def get_gateway_and_lan_addresses(interface_name: str) -> typing.Tuple[str, str]
     for iface_name, (gateway, lan) in get_interfaces().items():
         if interface_name == iface_name:
             return gateway, lan
-    return '', ''
+    raise UPnPError(f'failed to get lan and gateway addresses for {interface_name}')

aioupnp/upnp.py

@@ -47,10 +47,7 @@ class UPnP:
     async def discover(cls, lan_address: str = '', gateway_address: str = '', timeout: int = 30,
                        igd_args: Optional[Dict[str, Union[str, int]]] = None, interface_name: str = 'default',
                        loop: Optional[asyncio.AbstractEventLoop] = None) -> 'UPnP':
-        try:
+        lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name)
-            lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name)
-        except Exception as err:
-            raise UPnPError("failed to get lan and gateway addresses: %s" % str(err))
         gateway = await Gateway.discover_gateway(
             lan_address, gateway_address, timeout, igd_args, loop
         )
@@ -135,8 +132,14 @@ class UPnP:
 
     async def get_next_mapping(self, port: int, protocol: str, description: str,
                                internal_port: Optional[int] = None) -> int:
-        if protocol not in ["UDP", "TCP"]:
+        """
-            raise UPnPError("unsupported protocol: {}".format(protocol))
+        :param port:          (int) external port to redirect from
+        :param protocol:      (str) 'UDP' | 'TCP'
+        :param description:   (str) mapping description
+        :param internal_port: (int) internal port to redirect to
+
+        :return: (int) <mapped port>
+        """
         _internal_port = int(internal_port or port)
         requested_port = int(_internal_port)
         port = int(port)

tests/test_interfaces.py

@@ -1,6 +1,7 @@
-import unittest
 from unittest import mock
+from aioupnp.fault import UPnPError
 from aioupnp.upnp import UPnP
+from tests import AsyncioTestCase
 
 
 class mock_netifaces:
@@ -46,7 +47,7 @@ class mock_netifaces:
         }[interface]
 
 
-class TestParseInterfaces(unittest.TestCase):
+class TestParseInterfaces(AsyncioTestCase):
     def test_parse_interfaces(self):
         with mock.patch('aioupnp.interfaces.get_netifaces') as patch:
             patch.return_value = mock_netifaces
@@ -54,3 +55,16 @@ class TestParseInterfaces(unittest.TestCase):
             lan, gateway = UPnP.get_lan_and_gateway(interface_name='test0')
             self.assertEqual(gateway, '192.168.1.1')
             self.assertEqual(lan, '192.168.1.2')
+
+    async def test_netifaces_fail(self):
+        checked = []
+        with mock.patch('aioupnp.interfaces.get_netifaces') as patch:
+            patch.return_value = mock_netifaces
+            try:
+                await UPnP.discover(interface_name='test1')
+            except UPnPError as err:
+                self.assertEqual(str(err), 'failed to get lan and gateway addresses for test1')
+                checked.append(True)
+            else:
+                self.assertTrue(False)
+        self.assertTrue(len(checked) == 1)

tests/test_upnp.py

@@ -1,10 +1,22 @@
 from tests import AsyncioTestCase, mock_tcp_and_udp
 from collections import OrderedDict
 from aioupnp.upnp import UPnP
+from aioupnp.fault import UPnPError
 from aioupnp.gateway import Gateway
 from aioupnp.serialization.ssdp import SSDPDatagram
 
 
+class TestGetAnnotations(AsyncioTestCase):
+    def test_get_annotations(self):
+        expected = {
+            'NewRemoteHost': str, 'NewExternalPort': int, 'NewProtocol': str, 'NewInternalPort': int,
+            'NewInternalClient': str, 'NewEnabled': int, 'NewPortMappingDescription': str,
+            'NewLeaseDuration': str, 'return': None
+        }
+
+        self.assertDictEqual(expected, UPnP.get_annotations('AddPortMapping'))
+
+
 class UPnPCommandTestCase(AsyncioTestCase):
     gateway_address = "11.2.3.4"
     client_address = "11.2.3.5"
@@ -88,3 +100,26 @@ class TestGetNextPortMapping(UPnPCommandTestCase):
             self.assertEqual(4567, ext_port)
             result = await upnp.delete_port_mapping(ext_port, "UDP")
             self.assertEqual(None, result)
+
+
+class TestGetSpecificPortMapping(UPnPCommandTestCase):
+    client_address = '11.2.3.4'
+
+    def setUp(self) -> None:
+        self.replies.update({
+            b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 11.2.3.4\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 399\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetSpecificPortMappingEntry"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:GetSpecificPortMappingEntry xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"><NewRemoteHost></NewRemoteHost><NewExternalPort>1000</NewExternalPort><NewProtocol>UDP</NewProtocol></u:GetSpecificPortMappingEntry></s:Body></s:Envelope>\r\n': b'HTTP/1.1 500 Internal Server Error\r\nServer: WebServer\r\nDate: Wed, 22 May 2019 06:48:57 GMT\r\nConnection: close\r\nCONTENT-TYPE: text/xml; charset="utf-8"\r\nCONTENT-LENGTH: 474 \r\nEXT:\r\n\r\n<?xml version="1.0"?>\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">\n\t<s:Body>\n\t\t<s:Fault>\n\t\t\t<faultcode>s:Client</faultcode>\n\t\t\t<faultstring>UPnPError</faultstring>\n\t\t\t<detail>\n\t\t\t\t<UPnPError xmlns="urn:schemas-upnp-org:control-1-0">\n\t\t\t\t\t<errorCode>714</errorCode>\n\t\t\t\t\t<errorDescription>NoSuchEntryInArray</errorDescription>\n\t\t\t\t</UPnPError>\n\t\t\t</detail>\n\t\t</s:Fault>\n\t</s:Body>\n</s:Envelope>\n',
+            b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 11.2.3.4\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 399\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetSpecificPortMappingEntry"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n<?xml version="1.0"?>\r\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body><u:GetSpecificPortMappingEntry xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1"><NewRemoteHost></NewRemoteHost><NewExternalPort>9308</NewExternalPort><NewProtocol>UDP</NewProtocol></u:GetSpecificPortMappingEntry></s:Body></s:Envelope>\r\n': b'HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Wed, 22 May 2019 06:50:07 GMT\r\nConnection: close\r\nCONTENT-TYPE: text/xml; charset="utf-8"\r\nCONTENT-LENGTH: 562 \r\nEXT:\r\n\r\n<?xml version="1.0"?>\n<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">\n\t<s:Body>\n\t\t<u:GetSpecificPortMappingEntryResponse xmlns:u="urn:schemas-upnp-org:service:WANIPConnection:1">\n\n<NewInternalPort>9308</NewInternalPort>\n<NewInternalClient>11.2.3.55</NewInternalClient>\n<NewEnabled>1</NewEnabled>\n<NewPortMappingDescription>11.2.3.55:9308 to 9308 (UDP)</NewPortMappingDescription>\n<NewLeaseDuration>0</NewLeaseDuration>\n</u:GetSpecificPortMappingEntryResponse>\n\t</s:Body>\n</s:Envelope>\n'
+        })
+
+    async def test_get_specific_port_mapping(self):
+        with mock_tcp_and_udp(self.loop, tcp_replies=self.replies):
+            gateway = Gateway(self.reply, self.m_search_args, self.client_address, self.gateway_address)
+            await gateway.discover_commands(self.loop)
+            upnp = UPnP(self.client_address, self.gateway_address, gateway)
+            try:
+                await upnp.get_specific_port_mapping(1000, 'UDP')
+            except UPnPError:
+                result = await upnp.get_specific_port_mapping(9308, 'UDP')
+                self.assertEqual((9308, '11.2.3.55', True, '11.2.3.55:9308 to 9308 (UDP)', 0), result)
+            else:
+                self.assertTrue(False)