From 836271c6e0b3175545e7b6970d929c546f8e968a Mon Sep 17 00:00:00 2001 From: Jack Robison Date: Tue, 21 May 2019 18:17:17 -0400 Subject: [PATCH] tests --- aioupnp/__init__.py | 1 + stubs/defusedxml/__init__.py | 23 ++++ stubs/netifaces.py | 9 +- tests/__init__.py | 145 +++++++++++++++++------ tests/mocks.py | 64 ---------- tests/protocols/test_multicast.py | 23 ++++ tests/protocols/test_scpd.py | 7 +- tests/protocols/test_ssdp.py | 5 +- tests/serialization/test_scpd.py | 188 ++++++++++++++++++++++++++++++ tests/serialization/test_soap.py | 37 ++++++ tests/test_cli.py | 11 +- tests/test_gateway.py | 139 +++++++++++++++++++++- tests/test_interfaces.py | 55 +++++++++ 13 files changed, 582 insertions(+), 125 deletions(-) create mode 100644 stubs/defusedxml/__init__.py delete mode 100644 tests/mocks.py create mode 100644 tests/protocols/test_multicast.py create mode 100644 tests/test_interfaces.py diff --git a/aioupnp/__init__.py b/aioupnp/__init__.py index 7fab9dc..523847a 100644 --- a/aioupnp/__init__.py +++ b/aioupnp/__init__.py @@ -4,3 +4,4 @@ __author__ = "Jack Robison" __maintainer__ = "Jack Robison" __license__ = "MIT" __email__ = "jackrobison@lbry.io" + diff --git a/stubs/defusedxml/__init__.py b/stubs/defusedxml/__init__.py new file mode 100644 index 0000000..db9fd1c --- /dev/null +++ b/stubs/defusedxml/__init__.py @@ -0,0 +1,23 @@ +import typing + + +class ElementTree: + tag: typing.Optional[str] = None + """The element's name.""" + + attrib: typing.Optional[typing.Dict[str, str]] = None + """Dictionary of the element's attributes.""" + + text: typing.Optional[str] = None + + tail: typing.Optional[str] = None + + def __len__(self) -> int: + raise NotImplementedError() + + def __iter__(self) -> typing.Iterator['ElementTree']: + raise NotImplementedError() + + @classmethod + def fromstring(cls, xml_str: str) -> 'ElementTree': + raise NotImplementedError() \ No newline at end of file diff --git a/stubs/netifaces.py b/stubs/netifaces.py index 36b7d07..075b09c 100644 --- a/stubs/netifaces.py +++ b/stubs/netifaces.py @@ -36,8 +36,9 @@ version = '0.10.7' # functions - -def gateways(*args, **kwargs) -> typing.List: # real signature unknown +def gateways(*args, **kwargs) -> typing.Dict[typing.Union[str, int], + typing.Union[typing.Dict[int, typing.Tuple[str, str]], + typing.List[typing.Tuple[str, str, bool]]]]: """ Obtain a list of the gateways on this machine. @@ -56,7 +57,7 @@ def gateways(*args, **kwargs) -> typing.List: # real signature unknown pass -def ifaddresses(*args, **kwargs) -> typing.Dict: # real signature unknown +def ifaddresses(*args, **kwargs) -> typing.Dict[int, typing.List[typing.Dict[str, str]]]: """ Obtain information about the specified network interface. @@ -67,7 +68,7 @@ def ifaddresses(*args, **kwargs) -> typing.Dict: # real signature unknown pass -def interfaces(*args, **kwargs) -> typing.List: # real signature unknown +def interfaces(*args, **kwargs) -> typing.List[str]: """ Obtain a list of the interfaces available on this machine. """ pass diff --git a/tests/__init__.py b/tests/__init__.py index dd299fc..51ce45a 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,7 +1,11 @@ import asyncio import unittest +import contextlib +import socket +from unittest import mock from unittest.case import _Outcome + try: from asyncio.runners import _cancel_all_tasks except ImportError: @@ -10,30 +14,89 @@ except ImportError: pass -class TestBase(unittest.TestCase): +@contextlib.contextmanager +def mock_tcp_and_udp(loop, udp_expected_addr=None, udp_replies=None, udp_delay_reply=0.0, sent_udp_packets=None, + tcp_replies=None, tcp_delay_reply=0.0, sent_tcp_packets=None): + sent_udp_packets = sent_udp_packets if sent_udp_packets is not None else [] + udp_replies = udp_replies or {} + + sent_tcp_packets = sent_tcp_packets if sent_tcp_packets is not None else [] + tcp_replies = tcp_replies or {} + + async def create_connection(protocol_factory, host=None, port=None): + def write(p: asyncio.Protocol): + def _write(data): + sent_tcp_packets.append(data) + if data in tcp_replies: + loop.call_later(tcp_delay_reply, p.data_received, tcp_replies[data]) + + return _write + + protocol = protocol_factory() + transport = asyncio.Transport(extra={'socket': mock.Mock(spec=socket.socket)}) + transport.close = lambda: None + transport.write = write(protocol) + protocol.connection_made(transport) + return transport, protocol + + async def create_datagram_endpoint(proto_lam, sock=None): + def sendto(p: asyncio.DatagramProtocol): + def _sendto(data, addr): + sent_udp_packets.append(data) + if (data, addr) in udp_replies: + loop.call_later(udp_delay_reply, p.datagram_received, udp_replies[(data, addr)], + (udp_expected_addr, 1900)) + + return _sendto + + protocol = proto_lam() + transport = asyncio.DatagramTransport(extra={'socket': mock_sock}) + transport.close = lambda: mock_sock.close() + mock_sock.sendto = sendto(protocol) + transport.sendto = mock_sock.sendto + protocol.connection_made(transport) + return transport, protocol + + with mock.patch('socket.socket') as mock_socket: + mock_sock = mock.Mock(spec=socket.socket) + mock_sock.setsockopt = lambda *_: None + mock_sock.bind = lambda *_: None + mock_sock.setblocking = lambda *_: None + mock_sock.getsockname = lambda: "0.0.0.0" + mock_sock.getpeername = lambda: "" + mock_sock.close = lambda: None + mock_sock.type = socket.SOCK_DGRAM + mock_sock.fileno = lambda: 7 + + mock_socket.return_value = mock_sock + loop.create_datagram_endpoint = create_datagram_endpoint + loop.create_connection = create_connection + yield + + +class AsyncioTestCase(unittest.TestCase): # Implementation inspired by discussion: # https://bugs.python.org/issue32972 - async def asyncSetUp(self): + maxDiff = None + + async def asyncSetUp(self): # pylint: disable=C0103 pass - async def asyncTearDown(self): + async def asyncTearDown(self): # pylint: disable=C0103 pass - async def doAsyncCleanups(self): - pass - - def run(self, result=None): + def run(self, result=None): # pylint: disable=R0915 orig_result = result if result is None: result = self.defaultTestResult() - startTestRun = getattr(result, 'startTestRun', None) + startTestRun = getattr(result, 'startTestRun', None) # pylint: disable=C0103 if startTestRun is not None: startTestRun() result.startTest(self) - testMethod = getattr(self, self._testMethodName) + testMethod = getattr(self, self._testMethodName) # pylint: disable=C0103 if (getattr(self.__class__, "__unittest_skip__", False) or getattr(testMethod, "__unittest_skip__", False)): # If the class or method was skipped. @@ -50,36 +113,36 @@ class TestBase(unittest.TestCase): "__unittest_expecting_failure__", False) expecting_failure = expecting_failure_class or expecting_failure_method outcome = _Outcome(result) + + self.loop = asyncio.new_event_loop() # pylint: disable=W0201 + asyncio.set_event_loop(self.loop) + self.loop.set_debug(True) + try: self._outcome = outcome - loop = asyncio.new_event_loop() - try: - asyncio.set_event_loop(loop) - loop.set_debug(True) - + with outcome.testPartExecutor(self): + self.setUp() + self.loop.run_until_complete(self.asyncSetUp()) + if outcome.success: + outcome.expecting_failure = expecting_failure + with outcome.testPartExecutor(self, isTest=True): + maybe_coroutine = testMethod() + if asyncio.iscoroutine(maybe_coroutine): + self.loop.run_until_complete(maybe_coroutine) + outcome.expecting_failure = False with outcome.testPartExecutor(self): - self.setUp() - loop.run_until_complete(self.asyncSetUp()) - if outcome.success: - outcome.expecting_failure = expecting_failure - with outcome.testPartExecutor(self, isTest=True): - possible_coroutine = testMethod() - if asyncio.iscoroutine(possible_coroutine): - loop.run_until_complete(possible_coroutine) - outcome.expecting_failure = False - with outcome.testPartExecutor(self): - loop.run_until_complete(self.asyncTearDown()) - self.tearDown() - finally: - try: - _cancel_all_tasks(loop) - loop.run_until_complete(loop.shutdown_asyncgens()) - finally: - asyncio.set_event_loop(None) - loop.close() + self.loop.run_until_complete(self.asyncTearDown()) + self.tearDown() - self.doCleanups() + self.doAsyncCleanups() + + try: + _cancel_all_tasks(self.loop) + self.loop.run_until_complete(self.loop.shutdown_asyncgens()) + finally: + asyncio.set_event_loop(None) + self.loop.close() for test, reason in outcome.skipped: self._addSkip(result, test, reason) @@ -96,9 +159,9 @@ class TestBase(unittest.TestCase): finally: result.stopTest(self) if orig_result is None: - stopTestRun = getattr(result, 'stopTestRun', None) + stopTestRun = getattr(result, 'stopTestRun', None) # pylint: disable=C0103 if stopTestRun is not None: - stopTestRun() + stopTestRun() # pylint: disable=E1102 # explicitly break reference cycles: # outcome.errors -> frame -> outcome -> outcome.errors @@ -109,5 +172,11 @@ class TestBase(unittest.TestCase): # clear the outcome, no more needed self._outcome = None - def setUp(self): - self.loop = asyncio.get_event_loop_policy().get_event_loop() + def doAsyncCleanups(self): # pylint: disable=C0103 + outcome = self._outcome or _Outcome() + while self._cleanups: + function, args, kwargs = self._cleanups.pop() + with outcome.testPartExecutor(self): + maybe_coroutine = function(*args, **kwargs) + if asyncio.iscoroutine(maybe_coroutine): + self.loop.run_until_complete(maybe_coroutine) diff --git a/tests/mocks.py b/tests/mocks.py deleted file mode 100644 index 1351e78..0000000 --- a/tests/mocks.py +++ /dev/null @@ -1,64 +0,0 @@ -import asyncio -import contextlib -import socket -import mock - - -@contextlib.contextmanager -def mock_tcp_and_udp(loop, udp_expected_addr=None, udp_replies=None, udp_delay_reply=0.0, sent_udp_packets=None, - tcp_replies=None, tcp_delay_reply=0.0, sent_tcp_packets=None): - sent_udp_packets = sent_udp_packets if sent_udp_packets is not None else [] - udp_replies = udp_replies or {} - - sent_tcp_packets = sent_tcp_packets if sent_tcp_packets is not None else [] - tcp_replies = tcp_replies or {} - - async def create_connection(protocol_factory, host=None, port=None): - def write(p: asyncio.Protocol): - def _write(data): - sent_tcp_packets.append(data) - if data in tcp_replies: - loop.call_later(tcp_delay_reply, p.data_received, tcp_replies[data]) - - return _write - - protocol = protocol_factory() - transport = asyncio.Transport(extra={'socket': mock.Mock(spec=socket.socket)}) - transport.close = lambda: None - transport.write = write(protocol) - protocol.connection_made(transport) - return transport, protocol - - async def create_datagram_endpoint(proto_lam, sock=None): - def sendto(p: asyncio.DatagramProtocol): - def _sendto(data, addr): - sent_udp_packets.append(data) - if (data, addr) in udp_replies: - loop.call_later(udp_delay_reply, p.datagram_received, udp_replies[(data, addr)], - (udp_expected_addr, 1900)) - - return _sendto - - protocol = proto_lam() - transport = asyncio.DatagramTransport(extra={'socket': mock_sock}) - transport.close = lambda: mock_sock.close() - mock_sock.sendto = sendto(protocol) - transport.sendto = mock_sock.sendto - protocol.connection_made(transport) - return transport, protocol - - with mock.patch('socket.socket') as mock_socket: - mock_sock = mock.Mock(spec=socket.socket) - mock_sock.setsockopt = lambda *_: None - mock_sock.bind = lambda *_: None - mock_sock.setblocking = lambda *_: None - mock_sock.getsockname = lambda: "0.0.0.0" - mock_sock.getpeername = lambda: "" - mock_sock.close = lambda: None - mock_sock.type = socket.SOCK_DGRAM - mock_sock.fileno = lambda: 7 - - mock_socket.return_value = mock_sock - loop.create_datagram_endpoint = create_datagram_endpoint - loop.create_connection = create_connection - yield diff --git a/tests/protocols/test_multicast.py b/tests/protocols/test_multicast.py new file mode 100644 index 0000000..065f12e --- /dev/null +++ b/tests/protocols/test_multicast.py @@ -0,0 +1,23 @@ +import unittest + +from asyncio import DatagramTransport +from aioupnp.protocols.multicast import MulticastProtocol + + +class TestMulticast(unittest.TestCase): + def test_it(self): + class none_socket: + sock = None + + def get(self, name, default=None): + return default + + protocol = MulticastProtocol('1.2.3.4', '1.2.3.4') + transport = DatagramTransport(none_socket()) + protocol.set_ttl(1) + with self.assertRaises(ValueError): + _ = protocol.get_ttl() + protocol.connection_made(transport) + protocol.set_ttl(1) + with self.assertRaises(ValueError): + _ = protocol.get_ttl() diff --git a/tests/protocols/test_scpd.py b/tests/protocols/test_scpd.py index 7c80dbb..6ff5442 100644 --- a/tests/protocols/test_scpd.py +++ b/tests/protocols/test_scpd.py @@ -1,10 +1,9 @@ from aioupnp.fault import UPnPError from aioupnp.protocols.scpd import scpd_post, scpd_get -from tests import TestBase -from tests.mocks import mock_tcp_and_udp +from tests import AsyncioTestCase, mock_tcp_and_udp -class TestSCPDGet(TestBase): +class TestSCPDGet(AsyncioTestCase): path, lan_address, port = '/IGDdevicedesc_brlan0.xml', '10.1.10.1', 49152 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' @@ -142,7 +141,7 @@ class TestSCPDGet(TestBase): self.assertTrue(str(err).startswith('too many bytes written')) -class TestSCPDPost(TestBase): +class TestSCPDPost(AsyncioTestCase): param_names: list = [] kwargs: dict = {} method, gateway_address, port = "GetExternalIPAddress", '10.0.0.1', 49152 diff --git a/tests/protocols/test_ssdp.py b/tests/protocols/test_ssdp.py index 8e780e0..a36f1a0 100644 --- a/tests/protocols/test_ssdp.py +++ b/tests/protocols/test_ssdp.py @@ -4,11 +4,10 @@ from aioupnp.protocols.m_search_patterns import packet_generator from aioupnp.serialization.ssdp import SSDPDatagram from aioupnp.constants import SSDP_IP_ADDRESS from aioupnp.protocols.ssdp import fuzzy_m_search, m_search -from tests import TestBase -from tests.mocks import mock_tcp_and_udp +from tests import AsyncioTestCase, mock_tcp_and_udp -class TestSSDP(TestBase): +class TestSSDP(AsyncioTestCase): packet_args = list(packet_generator()) byte_packets = [SSDPDatagram("M-SEARCH", p).encode().encode() for p in packet_args] diff --git a/tests/serialization/test_scpd.py b/tests/serialization/test_scpd.py index 5617fec..5e82918 100644 --- a/tests/serialization/test_scpd.py +++ b/tests/serialization/test_scpd.py @@ -1,5 +1,7 @@ import unittest +from aioupnp.fault import UPnPError from aioupnp.serialization.scpd import serialize_scpd_get, deserialize_scpd_get_response +from aioupnp.serialization.xml import xml_to_dict from aioupnp.device import Device from aioupnp.util import get_dict_val_case_insensitive @@ -20,6 +22,28 @@ class TestSCPDSerialization(unittest.TestCase): b"\r\n" \ b"\n\n\n1\n0\n\n\nurn:schemas-upnp-org:device:InternetGatewayDevice:1\nCGA4131COM\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:Layer3Forwarding:1\nurn:upnp-org:serviceId:L3Forwarding1\n/Layer3ForwardingSCPD.xml\n/upnp/control/Layer3Forwarding\n/upnp/event/Layer3Forwarding\n\n\n\n\nurn:schemas-upnp-org:device:WANDevice:1\nWANDevice:1\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:WANCommonInterfaceConfig:1\nurn:upnp-org:serviceId:WANCommonIFC1\n/WANCommonInterfaceConfigSCPD.xml\n/upnp/control/WANCommonInterfaceConfig0\n/upnp/event/WANCommonInterfaceConfig0\n\n\n\n \n urn:schemas-upnp-org:device:WANConnectionDevice:1\n WANConnectionDevice:1\n Cisco\n http://www.cisco.com/\n CGA4131COM\n CGA4131COM\n CGA4131COM\n http://www.cisco.com\n \n uuid:11111111-2222-3333-4444-555555555555\n CGA4131COM\n \n \n urn:schemas-upnp-org:service:WANIPConnection:1\n urn:upnp-org:serviceId:WANIPConn1\n /WANIPConnectionServiceSCPD.xml\n /upnp/control/WANIPConnection0\n /upnp/event/WANIPConnection0\n \n \n \n\n\n\nhttp://10.1.10.1/\n\n" + response_bad_root_device_name = 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"\n\n\n1\n0\n\n\nurn:schemas-upnp-org:device:InternetGatewayDevic3:1\nCGA4131COM\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:Layer3Forwarding:1\nurn:upnp-org:serviceId:L3Forwarding1\n/Layer3ForwardingSCPD.xml\n/upnp/control/Layer3Forwarding\n/upnp/event/Layer3Forwarding\n\n\n\n\nurn:schemas-upnp-org:device:WANDevice:1\nWANDevice:1\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:WANCommonInterfaceConfig:1\nurn:upnp-org:serviceId:WANCommonIFC1\n/WANCommonInterfaceConfigSCPD.xml\n/upnp/control/WANCommonInterfaceConfig0\n/upnp/event/WANCommonInterfaceConfig0\n\n\n\n \n urn:schemas-upnp-org:device:WANConnectionDevice:1\n WANConnectionDevice:1\n Cisco\n http://www.cisco.com/\n CGA4131COM\n CGA4131COM\n CGA4131COM\n http://www.cisco.com\n \n uuid:11111111-2222-3333-4444-555555555555\n CGA4131COM\n \n \n urn:schemas-upnp-org:service:WANIPConnection:1\n urn:upnp-org:serviceId:WANIPConn1\n /WANIPConnectionServiceSCPD.xml\n /upnp/control/WANIPConnection0\n /upnp/event/WANIPConnection0\n \n \n \n\n\n\nhttp://10.1.10.1/\n\n" + + response_bad_root_xmls = 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"\n\n\n1\n0\n\n\nurn:schemas-upnp-org:device:InternetGatewayDevic3:1\nCGA4131COM\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:Layer3Forwarding:1\nurn:upnp-org:serviceId:L3Forwarding1\n/Layer3ForwardingSCPD.xml\n/upnp/control/Layer3Forwarding\n/upnp/event/Layer3Forwarding\n\n\n\n\nurn:schemas-upnp-org:device:WANDevice:1\nWANDevice:1\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:WANCommonInterfaceConfig:1\nurn:upnp-org:serviceId:WANCommonIFC1\n/WANCommonInterfaceConfigSCPD.xml\n/upnp/control/WANCommonInterfaceConfig0\n/upnp/event/WANCommonInterfaceConfig0\n\n\n\n \n urn:schemas-upnp-org:device:WANConnectionDevice:1\n WANConnectionDevice:1\n Cisco\n http://www.cisco.com/\n CGA4131COM\n CGA4131COM\n CGA4131COM\n http://www.cisco.com\n \n uuid:11111111-2222-3333-4444-555555555555\n CGA4131COM\n \n \n urn:schemas-upnp-org:service:WANIPConnection:1\n urn:upnp-org:serviceId:WANIPConn1\n /WANIPConnectionServiceSCPD.xml\n /upnp/control/WANIPConnection0\n /upnp/event/WANIPConnection0\n \n \n \n\n\n\nhttp://10.1.10.1/\n\n" + expected_parsed = { 'specVersion': {'major': '1', 'minor': '0'}, 'device': { @@ -94,6 +118,87 @@ class TestSCPDSerialization(unittest.TestCase): def test_serialize_get(self): self.assertEqual(serialize_scpd_get(self.path, self.lan_address), self.get_request) + self.assertEqual(serialize_scpd_get(self.path, 'http://' + self.lan_address), self.get_request) + self.assertEqual(serialize_scpd_get(self.path, 'http://' + self.lan_address + ':1337'), self.get_request) + self.assertEqual(serialize_scpd_get(self.path, self.lan_address + ':1337'), self.get_request) + + def test_parse_device_response_xml(self): + self.assertDictEqual( + xml_to_dict('\n\n\t\n\t\t1\n\t\t0\n\t\n\thttp://10.0.0.1:49152\n\t\n\t\turn:schemas-upnp-org:device:InternetGatewayDevice:1\n\t\tWireless Broadband Router\n\t\tD-Link Corporation\n\t\thttp://www.dlink.com\n\t\tD-Link Router\n\t\tD-Link Router\n\t\tDIR-890L\n\t\thttp://www.dlink.com\n\t\t120\n\t\tuuid:11111111-2222-3333-4444-555555555555\n\t\t\n\t\t\t\n\t\t\t\timage/gif\n\t\t\t\t118\n\t\t\t\t119\n\t\t\t\t8\n\t\t\t\t/ligd.gif\n\t\t\t\n\t\t\n\t\t\n\t\t\t\n\t\t\t\turn:schemas-microsoft-com:service:OSInfo:1\n\t\t\t\turn:microsoft-com:serviceId:OSInfo1\n\t\t\t\t/soap.cgi?service=OSInfo1\n\t\t\t\t/gena.cgi?service=OSInfo1\n\t\t\t\t/OSInfo.xml\n\t\t\t\n\t\t\t\n\t\t\t\turn:schemas-upnp-org:service:Layer3Forwarding:1\n\t\t\t\turn:upnp-org:serviceId:L3Forwarding1\n\t\t\t\t/soap.cgi?service=L3Forwarding1\n\t\t\t\t/gena.cgi?service=L3Forwarding1\n\t\t\t\t/Layer3Forwarding.xml\n\t\t\t\n\t\t\n\t\t\n\t\t\t\n\t\t\t\turn:schemas-upnp-org:device:WANDevice:1\n\t\t\t\tWANDevice\n\t\t\t\tD-Link\n\t\t\t\thttp://www.dlink.com\n\t\t\t\tWANDevice\n\t\t\t\tDIR-890L\n\t\t\t\t1\n\t\t\t\thttp://www.dlink.com\n\t\t\t\t120\n\t\t\t\tuuid:11111111-2222-3333-4444-555555555555\n\t\t\t\t\n\t\t\t\t\t\n\t\t\t\t\t\turn:schemas-upnp-org:service:WANCommonInterfaceConfig:1\n\t\t\t\t\t\turn:upnp-org:serviceId:WANCommonIFC1\n\t\t\t\t\t\t/soap.cgi?service=WANCommonIFC1\n\t\t\t\t\t\t/gena.cgi?service=WANCommonIFC1\n\t\t\t\t\t\t/WANCommonInterfaceConfig.xml\n\t\t\t\t\t\n\t\t\t\t\n\t\t\t\t\n\t\t\t\t\t\n\t\t\t\t\t\turn:schemas-upnp-org:device:WANConnectionDevice:1\n\t\t\t\t\t\tWANConnectionDevice\n\t\t\t\t\t\tD-Link\n\t\t\t\t\t\thttp://www.dlink.com\n\t\t\t\t\t\tWanConnectionDevice\n\t\t\t\t\t\tDIR-890L\n\t\t\t\t\t\t1\n\t\t\t\t\t\thttp://www.dlink.com\n\t\t\t\t\t\t120\n\t\t\t\t\t\tuuid:11111111-2222-3333-4444-555555555555\n\t\t\t\t\t\t\n\t\t\t\t\t\t\t\n\t\t\t\t\t\t\t\turn:schemas-upnp-org:service:WANEthernetLinkConfig:1\n\t\t\t\t\t\t\t\turn:upnp-org:serviceId:WANEthLinkC1\n\t\t\t\t\t\t\t\t/soap.cgi?service=WANEthLinkC1\n\t\t\t\t\t\t\t\t/gena.cgi?service=WANEthLinkC1\n\t\t\t\t\t\t\t\t/WANEthernetLinkConfig.xml\n\t\t\t\t\t\t\t\n\t\t\t\t\t\t\t\n\t\t\t\t\t\t\t\turn:schemas-upnp-org:service:WANIPConnection:1\n\t\t\t\t\t\t\t\turn:upnp-org:serviceId:WANIPConn1\n\t\t\t\t\t\t\t\t/soap.cgi?service=WANIPConn1\n\t\t\t\t\t\t\t\t/gena.cgi?service=WANIPConn1\n\t\t\t\t\t\t\t\t/WANIPConnection.xml\n\t\t\t\t\t\t\t\n\t\t\t\t\t\t\n\t\t\t\t\t\n\t\t\t\t\n\t\t\t\n\t\t\n\t\thttp://10.0.0.1\n\t\n\n'), + {'{urn:schemas-upnp-org:device-1-0}root': { + '{urn:schemas-upnp-org:device-1-0}specVersion': {'{urn:schemas-upnp-org:device-1-0}major': '1', + '{urn:schemas-upnp-org:device-1-0}minor': '0'}, + '{urn:schemas-upnp-org:device-1-0}URLBase': 'http://10.0.0.1:49152', + '{urn:schemas-upnp-org:device-1-0}device': { + '{urn:schemas-upnp-org:device-1-0}deviceType': 'urn:schemas-upnp-org:device:InternetGatewayDevice:1', + '{urn:schemas-upnp-org:device-1-0}friendlyName': 'Wireless Broadband Router', + '{urn:schemas-upnp-org:device-1-0}manufacturer': 'D-Link Corporation', + '{urn:schemas-upnp-org:device-1-0}manufacturerURL': 'http://www.dlink.com', + '{urn:schemas-upnp-org:device-1-0}modelDescription': 'D-Link Router', + '{urn:schemas-upnp-org:device-1-0}modelName': 'D-Link Router', + '{urn:schemas-upnp-org:device-1-0}modelNumber': 'DIR-890L', + '{urn:schemas-upnp-org:device-1-0}modelURL': 'http://www.dlink.com', + '{urn:schemas-upnp-org:device-1-0}serialNumber': '120', + '{urn:schemas-upnp-org:device-1-0}UDN': 'uuid:11111111-2222-3333-4444-555555555555', + '{urn:schemas-upnp-org:device-1-0}iconList': {'{urn:schemas-upnp-org:device-1-0}icon': { + '{urn:schemas-upnp-org:device-1-0}mimetype': 'image/gif', + '{urn:schemas-upnp-org:device-1-0}width': '118', + '{urn:schemas-upnp-org:device-1-0}height': '119', '{urn:schemas-upnp-org:device-1-0}depth': '8', + '{urn:schemas-upnp-org:device-1-0}url': '/ligd.gif'}}, + '{urn:schemas-upnp-org:device-1-0}serviceList': {'{urn:schemas-upnp-org:device-1-0}service': [ + {'{urn:schemas-upnp-org:device-1-0}serviceType': 'urn:schemas-microsoft-com:service:OSInfo:1', + '{urn:schemas-upnp-org:device-1-0}serviceId': 'urn:microsoft-com:serviceId:OSInfo1', + '{urn:schemas-upnp-org:device-1-0}controlURL': '/soap.cgi?service=OSInfo1', + '{urn:schemas-upnp-org:device-1-0}eventSubURL': '/gena.cgi?service=OSInfo1', + '{urn:schemas-upnp-org:device-1-0}SCPDURL': '/OSInfo.xml'}, { + '{urn:schemas-upnp-org:device-1-0}serviceType': 'urn:schemas-upnp-org:service:Layer3Forwarding:1', + '{urn:schemas-upnp-org:device-1-0}serviceId': 'urn:upnp-org:serviceId:L3Forwarding1', + '{urn:schemas-upnp-org:device-1-0}controlURL': '/soap.cgi?service=L3Forwarding1', + '{urn:schemas-upnp-org:device-1-0}eventSubURL': '/gena.cgi?service=L3Forwarding1', + '{urn:schemas-upnp-org:device-1-0}SCPDURL': '/Layer3Forwarding.xml'}]}, + '{urn:schemas-upnp-org:device-1-0}deviceList': {'{urn:schemas-upnp-org:device-1-0}device': { + '{urn:schemas-upnp-org:device-1-0}deviceType': 'urn:schemas-upnp-org:device:WANDevice:1', + '{urn:schemas-upnp-org:device-1-0}friendlyName': 'WANDevice', + '{urn:schemas-upnp-org:device-1-0}manufacturer': 'D-Link', + '{urn:schemas-upnp-org:device-1-0}manufacturerURL': 'http://www.dlink.com', + '{urn:schemas-upnp-org:device-1-0}modelDescription': 'WANDevice', + '{urn:schemas-upnp-org:device-1-0}modelName': 'DIR-890L', + '{urn:schemas-upnp-org:device-1-0}modelNumber': '1', + '{urn:schemas-upnp-org:device-1-0}modelURL': 'http://www.dlink.com', + '{urn:schemas-upnp-org:device-1-0}serialNumber': '120', + '{urn:schemas-upnp-org:device-1-0}UDN': 'uuid:11111111-2222-3333-4444-555555555555', + '{urn:schemas-upnp-org:device-1-0}serviceList': {'{urn:schemas-upnp-org:device-1-0}service': { + '{urn:schemas-upnp-org:device-1-0}serviceType': 'urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1', + '{urn:schemas-upnp-org:device-1-0}serviceId': 'urn:upnp-org:serviceId:WANCommonIFC1', + '{urn:schemas-upnp-org:device-1-0}controlURL': '/soap.cgi?service=WANCommonIFC1', + '{urn:schemas-upnp-org:device-1-0}eventSubURL': '/gena.cgi?service=WANCommonIFC1', + '{urn:schemas-upnp-org:device-1-0}SCPDURL': '/WANCommonInterfaceConfig.xml'}}, + '{urn:schemas-upnp-org:device-1-0}deviceList': {'{urn:schemas-upnp-org:device-1-0}device': { + '{urn:schemas-upnp-org:device-1-0}deviceType': 'urn:schemas-upnp-org:device:WANConnectionDevice:1', + '{urn:schemas-upnp-org:device-1-0}friendlyName': 'WANConnectionDevice', + '{urn:schemas-upnp-org:device-1-0}manufacturer': 'D-Link', + '{urn:schemas-upnp-org:device-1-0}manufacturerURL': 'http://www.dlink.com', + '{urn:schemas-upnp-org:device-1-0}modelDescription': 'WanConnectionDevice', + '{urn:schemas-upnp-org:device-1-0}modelName': 'DIR-890L', + '{urn:schemas-upnp-org:device-1-0}modelNumber': '1', + '{urn:schemas-upnp-org:device-1-0}modelURL': 'http://www.dlink.com', + '{urn:schemas-upnp-org:device-1-0}serialNumber': '120', + '{urn:schemas-upnp-org:device-1-0}UDN': 'uuid:11111111-2222-3333-4444-555555555555', + '{urn:schemas-upnp-org:device-1-0}serviceList': { + '{urn:schemas-upnp-org:device-1-0}service': [{ + '{urn:schemas-upnp-org:device-1-0}serviceType': 'urn:schemas-upnp-org:service:WANEthernetLinkConfig:1', + '{urn:schemas-upnp-org:device-1-0}serviceId': 'urn:upnp-org:serviceId:WANEthLinkC1', + '{urn:schemas-upnp-org:device-1-0}controlURL': '/soap.cgi?service=WANEthLinkC1', + '{urn:schemas-upnp-org:device-1-0}eventSubURL': '/gena.cgi?service=WANEthLinkC1', + '{urn:schemas-upnp-org:device-1-0}SCPDURL': '/WANEthernetLinkConfig.xml'}, + { + '{urn:schemas-upnp-org:device-1-0}serviceType': 'urn:schemas-upnp-org:service:WANIPConnection:1', + '{urn:schemas-upnp-org:device-1-0}serviceId': 'urn:upnp-org:serviceId:WANIPConn1', + '{urn:schemas-upnp-org:device-1-0}controlURL': '/soap.cgi?service=WANIPConn1', + '{urn:schemas-upnp-org:device-1-0}eventSubURL': '/gena.cgi?service=WANIPConn1', + '{urn:schemas-upnp-org:device-1-0}SCPDURL': '/WANIPConnection.xml'}]}}}}}, + '{urn:schemas-upnp-org:device-1-0}presentationURL': 'http://10.0.0.1'}}} + ) def test_deserialize_get_response(self): self.assertDictEqual(deserialize_scpd_get_response(self.response), self.expected_parsed) @@ -101,6 +206,14 @@ class TestSCPDSerialization(unittest.TestCase): def test_deserialize_blank(self): self.assertDictEqual(deserialize_scpd_get_response(b''), {}) + def test_fail_to_deserialize_invalid_root_device(self): + with self.assertRaises(UPnPError): + deserialize_scpd_get_response(self.response_bad_root_device_name) + + def test_fail_to_deserialize_invalid_root_xmls(self): + with self.assertRaises(UPnPError): + deserialize_scpd_get_response(self.response_bad_root_xmls) + def test_deserialize_to_device_object(self): devices = [] services = [] @@ -173,3 +286,78 @@ class TestSCPDSerialization(unittest.TestCase): }, 'presentationURL': 'http://10.1.10.1/' } self.assertDictEqual(expected_result, device.as_dict()) + + def test_deserialize_another_device(self): + xml_bytes = b"\n\n\n1\n0\n\n\nurn:schemas-upnp-org:device:InternetGatewayDevice:1\nCGA4131COM\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:11111111-2222-3333-4444-555555555556\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:Layer3Forwarding:1\nurn:upnp-org:serviceId:L3Forwarding1\n/Layer3ForwardingSCPD.xml\n/upnp/control/Layer3Forwarding\n/upnp/event/Layer3Forwarding\n\n\n\n\nurn:schemas-upnp-org:device:WANDevice:1\nWANDevice:1\nCisco\nhttp://www.cisco.com/\nCGA4131COM\nCGA4131COM\nCGA4131COM\nhttp://www.cisco.com\n\nuuid:ebf5a0a0-1dd1-11b2-a92f-603d266f9915\nCGA4131COM\n\n\nurn:schemas-upnp-org:service:WANCommonInterfaceConfig:1\nurn:upnp-org:serviceId:WANCommonIFC1\n/WANCommonInterfaceConfigSCPD.xml\n/upnp/control/WANCommonInterfaceConfig0\n/upnp/event/WANCommonInterfaceConfig0\n\n\n\n \n urn:schemas-upnp-org:device:WANConnectionDevice:1\n WANConnectionDevice:1\n Cisco\n http://www.cisco.com/\n CGA4131COM\n CGA4131COM\n CGA4131COM\n http://www.cisco.com\n \n uuid:11111111-2222-3333-4444-555555555555\n CGA4131COM\n \n \n urn:schemas-upnp-org:service:WANIPConnection:1\n urn:upnp-org:serviceId:WANIPConn1\n /WANIPConnectionServiceSCPD.xml\n /upnp/control/WANIPConnection0\n /upnp/event/WANIPConnection0\n \n \n \n\n\n\nhttp://10.1.10.1/\n\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/' + } + } + self.assertDictEqual(expected_parsed, deserialize_scpd_get_response(xml_bytes)) diff --git a/tests/serialization/test_soap.py b/tests/serialization/test_soap.py index a238ef5..ea3cf04 100644 --- a/tests/serialization/test_soap.py +++ b/tests/serialization/test_soap.py @@ -28,6 +28,26 @@ class TestSOAPSerialization(unittest.TestCase): b"\r\n" \ b"\n\r\n11.22.33.44\r\n\r\n " + blank_response = b"HTTP/1.1 200 OK\r\n" \ + b"CONTENT-LENGTH: 148\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"\n " + + blank_response_body = b"HTTP/1.1 200 OK\r\n" \ + b"CONTENT-LENGTH: 280\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"\n\r\n " + error_response = b"HTTP/1.1 500 Internal Server Error\r\n" \ b"Server: WebServer\r\n" \ b"Date: Thu, 11 Oct 2018 22:16:17 GMT\r\n" \ @@ -43,12 +63,29 @@ class TestSOAPSerialization(unittest.TestCase): self.method, self.param_names, self.st, self.gateway_address, self.path, **self.kwargs ), self.post_bytes) + def test_serialize_post_http_host(self): + self.assertEqual(serialize_soap_post( + self.method, self.param_names, self.st, b'http://' + self.gateway_address, self.path, **self.kwargs + ), 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'} ) + def test_deserialize_error_response_field_not_found(self): + with self.assertRaises(UPnPError) as e: + deserialize_soap_post_response(self.post_response, self.method + 'derp', service_id=self.st.decode()) + self.assertTrue(str(e.exception).startswith('unknown response fields for GetExternalIPAddressderp')) + + def test_deserialize_blank_response(self): + # TODO: these seem like they should error... this test will break and have to be updated + self.assertDictEqual({}, deserialize_soap_post_response(self.blank_response, self.method, + service_id=self.st.decode())) + self.assertDictEqual({}, deserialize_soap_post_response(self.blank_response_body, self.method, + service_id=self.st.decode())) + def test_raise_from_error_response(self): raised = False try: diff --git a/tests/test_cli.py b/tests/test_cli.py index e1c4da5..e72ea59 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,7 +1,6 @@ import contextlib from io import StringIO -from tests import TestBase -from tests.mocks import mock_tcp_and_udp +from tests import AsyncioTestCase, mock_tcp_and_udp from collections import OrderedDict from aioupnp.serialization.ssdp import SSDPDatagram from aioupnp.protocols.m_search_patterns import packet_generator @@ -22,7 +21,7 @@ m_search_cli_result = """{ }\n""" -class TestCLI(TestBase): +class TestCLI(AsyncioTestCase): gateway_address = "10.0.0.1" soap_port = 49152 m_search_args = OrderedDict([ @@ -78,7 +77,7 @@ class TestCLI(TestBase): with contextlib.redirect_stdout(actual_output): with mock_tcp_and_udp(self.loop, '10.0.0.1', tcp_replies=self.scpd_replies, udp_replies=self.udp_replies): main( - (None, '--timeout=1', '--gateway_address=10.0.0.1', '--lan_address=10.0.0.2', 'get-external-ip'), + [None, '--timeout=1', '--gateway_address=10.0.0.1', '--lan_address=10.0.0.2', 'get-external-ip'], self.loop ) self.assertEqual("11.22.33.44\n", actual_output.getvalue()) @@ -89,7 +88,7 @@ class TestCLI(TestBase): with contextlib.redirect_stdout(actual_output): with mock_tcp_and_udp(self.loop, '10.0.0.1', tcp_replies=self.scpd_replies, udp_replies=self.udp_replies): main( - (None, '--timeout=1', '--gateway_address=10.0.0.1', '--lan_address=10.0.0.2', 'm-search'), + [None, '--timeout=1', '--gateway_address=10.0.0.1', '--lan_address=10.0.0.2', 'm-search'], self.loop ) self.assertEqual(timeout_msg, actual_output.getvalue()) @@ -98,7 +97,7 @@ class TestCLI(TestBase): with contextlib.redirect_stdout(actual_output): with mock_tcp_and_udp(self.loop, '10.0.0.1', tcp_replies=self.scpd_replies, udp_replies=self.udp_replies): main( - (None, '--timeout=1', '--gateway_address=10.0.0.1', '--lan_address=10.0.0.2', '--unicast', 'm-search'), + [None, '--timeout=1', '--gateway_address=10.0.0.1', '--lan_address=10.0.0.2', '--unicast', 'm-search'], self.loop ) self.assertEqual(m_search_cli_result, actual_output.getvalue()) diff --git a/tests/test_gateway.py b/tests/test_gateway.py index c105f11..6366e27 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -1,10 +1,7 @@ -import asyncio - from aioupnp.fault import UPnPError -from tests import TestBase -from tests.mocks import mock_tcp_and_udp +from tests import AsyncioTestCase, mock_tcp_and_udp from collections import OrderedDict -from aioupnp.gateway import Gateway +from aioupnp.gateway import Gateway, get_action_list from aioupnp.serialization.ssdp import SSDPDatagram @@ -14,7 +11,137 @@ def gen_get_bytes(location: str, host: str) -> bytes: ).encode() -class TestDiscoverDLinkDIR890L(TestBase): +class TestParseActionList(AsyncioTestCase): + test_action_list = {'actionList': { + 'action': [OrderedDict([('name', 'SetConnectionType'), ('argumentList', OrderedDict([('argument', OrderedDict( + [('name', 'NewConnectionType'), ('direction', 'in'), ('relatedStateVariable', 'ConnectionType')]))]))]), + OrderedDict([('name', 'GetConnectionTypeInfo'), ('argumentList', OrderedDict([('argument', [ + OrderedDict([('name', 'NewConnectionType'), ('direction', 'out'), + ('relatedStateVariable', 'ConnectionType')]), OrderedDict( + [('name', 'NewPossibleConnectionTypes'), ('direction', 'out'), + ('relatedStateVariable', 'PossibleConnectionTypes')])])]))]), + OrderedDict([('name', 'RequestConnection')]), OrderedDict([('name', 'ForceTermination')]), + OrderedDict([('name', 'GetStatusInfo'), ('argumentList', OrderedDict([('argument', [OrderedDict( + [('name', 'NewConnectionStatus'), ('direction', 'out'), + ('relatedStateVariable', 'ConnectionStatus')]), OrderedDict( + [('name', 'NewLastConnectionError'), ('direction', 'out'), + ('relatedStateVariable', 'LastConnectionError')]), OrderedDict( + [('name', 'NewUptime'), ('direction', 'out'), ('relatedStateVariable', 'Uptime')])])]))]), + OrderedDict([('name', 'GetNATRSIPStatus'), ('argumentList', OrderedDict([('argument', [OrderedDict( + [('name', 'NewRSIPAvailable'), ('direction', 'out'), + ('relatedStateVariable', 'RSIPAvailable')]), OrderedDict( + [('name', 'NewNATEnabled'), ('direction', 'out'), + ('relatedStateVariable', 'NATEnabled')])])]))]), OrderedDict( + [('name', 'GetGenericPortMappingEntry'), ('argumentList', OrderedDict([('argument', [OrderedDict( + [('name', 'NewPortMappingIndex'), ('direction', 'in'), + ('relatedStateVariable', 'PortMappingNumberOfEntries')]), OrderedDict( + [('name', 'NewRemoteHost'), ('direction', 'out'), ('relatedStateVariable', 'RemoteHost')]), + OrderedDict( + [('name', 'NewExternalPort'), ('direction', 'out'), ('relatedStateVariable', 'ExternalPort')]), + OrderedDict( + [('name', 'NewProtocol'), ('direction', 'out'), + ('relatedStateVariable', 'PortMappingProtocol')]), + OrderedDict([('name', + 'NewInternalPort'), + ('direction', + 'out'), ( + 'relatedStateVariable', + 'InternalPort')]), + OrderedDict([('name', + 'NewInternalClient'), + ('direction', + 'out'), ( + 'relatedStateVariable', + 'InternalClient')]), + OrderedDict([('name', + 'NewEnabled'), + ('direction', + 'out'), ( + 'relatedStateVariable', + 'PortMappingEnabled')]), + OrderedDict([('name', + 'NewPortMappingDescription'), + ('direction', + 'out'), ( + 'relatedStateVariable', + 'PortMappingDescription')]), + OrderedDict([('name', + 'NewLeaseDuration'), + ('direction', + 'out'), ( + 'relatedStateVariable', + 'PortMappingLeaseDuration')])])]))]), + OrderedDict([('name', 'GetSpecificPortMappingEntry'), ('argumentList', OrderedDict([('argument', [ + OrderedDict( + [('name', 'NewRemoteHost'), ('direction', 'in'), ('relatedStateVariable', 'RemoteHost')]), + OrderedDict([('name', 'NewExternalPort'), ('direction', 'in'), + ('relatedStateVariable', 'ExternalPort')]), OrderedDict( + [('name', 'NewProtocol'), ('direction', 'in'), + ('relatedStateVariable', 'PortMappingProtocol')]), OrderedDict( + [('name', 'NewInternalPort'), ('direction', 'out'), + ('relatedStateVariable', 'InternalPort')]), OrderedDict( + [('name', 'NewInternalClient'), ('direction', 'out'), + ('relatedStateVariable', 'InternalClient')]), OrderedDict( + [('name', 'NewEnabled'), ('direction', 'out'), + ('relatedStateVariable', 'PortMappingEnabled')]), OrderedDict( + [('name', 'NewPortMappingDescription'), ('direction', 'out'), + ('relatedStateVariable', 'PortMappingDescription')]), OrderedDict( + [('name', 'NewLeaseDuration'), ('direction', 'out'), + ('relatedStateVariable', 'PortMappingLeaseDuration')])])]))]), OrderedDict( + [('name', 'AddPortMapping'), ('argumentList', OrderedDict([('argument', [ + OrderedDict( + [('name', 'NewRemoteHost'), ('direction', 'in'), ('relatedStateVariable', 'RemoteHost')]), + OrderedDict( + [('name', 'NewExternalPort'), ('direction', 'in'), ('relatedStateVariable', 'ExternalPort')]), + OrderedDict( + [('name', 'NewProtocol'), ('direction', 'in'), + ('relatedStateVariable', 'PortMappingProtocol')]), + OrderedDict( + [('name', 'NewInternalPort'), ('direction', 'in'), ('relatedStateVariable', 'InternalPort')]), + OrderedDict( + [('name', 'NewInternalClient'), ('direction', 'in'), + ('relatedStateVariable', 'InternalClient')]), + OrderedDict( + [('name', 'NewEnabled'), ('direction', 'in'), ('relatedStateVariable', 'PortMappingEnabled')]), + OrderedDict([('name', 'NewPortMappingDescription'), ('direction', 'in'), + ('relatedStateVariable', 'PortMappingDescription')]), OrderedDict( + [('name', 'NewLeaseDuration'), ('direction', 'in'), + ('relatedStateVariable', 'PortMappingLeaseDuration')])])]))]), OrderedDict( + [('name', 'DeletePortMapping'), ('argumentList', OrderedDict([('argument', [ + OrderedDict( + [('name', 'NewRemoteHost'), ('direction', 'in'), ('relatedStateVariable', 'RemoteHost')]), + OrderedDict( + [('name', 'NewExternalPort'), ('direction', 'in'), ('relatedStateVariable', 'ExternalPort')]), + OrderedDict( + [('name', 'NewProtocol'), ('direction', 'in'), + ('relatedStateVariable', 'PortMappingProtocol')])])]))]), + OrderedDict([('name', 'GetExternalIPAddress'), + ('argumentList', OrderedDict( + [('argument', OrderedDict([('name', 'NewExternalIPAddress'), + ('direction', 'out'), + ('relatedStateVariable', 'ExternalIPAddress')]))]))])]}} + + def test_parse_expected_action_list(self): + expected = [('SetConnectionType', ['NewConnectionType'], []), + ('GetConnectionTypeInfo', [], ['NewConnectionType', 'NewPossibleConnectionTypes']), + ('RequestConnection', [], []), ('ForceTermination', [], []), + ('GetStatusInfo', [], ['NewConnectionStatus', 'NewLastConnectionError', 'NewUptime']), + ('GetNATRSIPStatus', [], ['NewRSIPAvailable', 'NewNATEnabled']), ( + 'GetGenericPortMappingEntry', ['NewPortMappingIndex'], + ['NewRemoteHost', 'NewExternalPort', 'NewProtocol', 'NewInternalPort', 'NewInternalClient', + 'NewEnabled', 'NewPortMappingDescription', 'NewLeaseDuration']), ( + 'GetSpecificPortMappingEntry', ['NewRemoteHost', 'NewExternalPort', 'NewProtocol'], + ['NewInternalPort', 'NewInternalClient', 'NewEnabled', 'NewPortMappingDescription', + 'NewLeaseDuration']), ('AddPortMapping', + ['NewRemoteHost', 'NewExternalPort', 'NewProtocol', 'NewInternalPort', + 'NewInternalClient', 'NewEnabled', 'NewPortMappingDescription', + 'NewLeaseDuration'], []), + ('DeletePortMapping', ['NewRemoteHost', 'NewExternalPort', 'NewProtocol'], []), + ('GetExternalIPAddress', [], ['NewExternalIPAddress'])] + self.assertEqual(expected, get_action_list(self.test_action_list)) + + +class TestDiscoverDLinkDIR890L(AsyncioTestCase): gateway_address = "10.0.0.1" client_address = "10.0.0.2" soap_port = 49152 diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py new file mode 100644 index 0000000..b2a71c2 --- /dev/null +++ b/tests/test_interfaces.py @@ -0,0 +1,55 @@ +import unittest +from unittest import mock + + +class mock_netifaces: + @staticmethod + def gateways(): + return { + "default": { + 2: [ + "192.168.1.1", + "test0" + ] + }, + 2: [ + [ + "192.168.1.1", + "test0", + True + ] + ] + } + @staticmethod + def interfaces(): + return ['test0'] + + @staticmethod + def ifaddresses(interface): + return { + "test0": { + 17: [ + { + "addr": "01:02:03:04:05:06", + "broadcast": "ff:ff:ff:ff:ff:ff" + } + ], + 2: [ + { + "addr": "192.168.1.2", + "netmask": "255.255.255.0", + "broadcast": "192.168.1.255" + } + ], + }, + }[interface] + + +class TestParseInterfaces(unittest.TestCase): + def test_parse_interfaces(self): + with mock.patch('aioupnp.interfaces.get_netifaces') as patch: + patch.return_value = mock_netifaces + import aioupnp.interfaces + gateway, lan = aioupnp.interfaces.get_gateway_and_lan_addresses('test0') + self.assertEqual(gateway, '192.168.1.1') + self.assertEqual(lan, '192.168.1.2')