From ab0b9707f2afbf3430b99475ffefecf6b6a792ae Mon Sep 17 00:00:00 2001 From: Jack Robison Date: Thu, 25 Oct 2018 15:28:00 -0400 Subject: [PATCH] test cli --- aioupnp/__main__.py | 8 +- aioupnp/gateway.py | 2 +- aioupnp/upnp.py | 27 +++--- setup.py | 2 +- tests/__init__.py | 113 ++++++++++++++++++++++++++ tests/{protocols => }/mocks.py | 60 ++++++++++++++ tests/protocols/__init__.py | 113 -------------------------- tests/protocols/test_scpd.py | 4 +- tests/protocols/test_ssdp.py | 4 +- tests/test_cli.py | 104 ++++++++++++++++++++++++ tests/{protocols => }/test_gateway.py | 5 +- 11 files changed, 306 insertions(+), 136 deletions(-) create mode 100644 tests/__init__.py rename tests/{protocols => }/mocks.py (54%) create mode 100644 tests/test_cli.py rename tests/{protocols => }/test_gateway.py (99%) diff --git a/aioupnp/__main__.py b/aioupnp/__main__.py index 0a64981..cd8fae8 100644 --- a/aioupnp/__main__.py +++ b/aioupnp/__main__.py @@ -24,7 +24,8 @@ def get_help(command): ) -def main(): +def main(argv=None, loop=None): + argv = argv or sys.argv commands = [n for n in dir(UPnP) if hasattr(getattr(UPnP, n, None), "_cli")] help_str = "\n".join(textwrap.wrap( " | ".join(commands), 100, initial_indent=' ', subsequent_indent=' ', break_long_words=False @@ -40,7 +41,7 @@ def main(): "For help with a specific command:" \ " aioupnp help \n" % (base_usage, help_str) - args = sys.argv[1:] + args = argv[1:] if args[0] in ['help', '-h', '--help']: if len(args) > 1: if args[1] in commands: @@ -53,6 +54,7 @@ def main(): 'gateway_address': '', 'lan_address': '', 'timeout': 30, + 'unicast': False } options = OrderedDict() @@ -88,7 +90,7 @@ def main(): UPnP.run_cli( command.replace('-', '_'), options, options.pop('lan_address'), options.pop('gateway_address'), - options.pop('timeout'), options.pop('interface'), kwargs + options.pop('timeout'), options.pop('interface'), options.pop('unicast'), kwargs, loop ) diff --git a/aioupnp/gateway.py b/aioupnp/gateway.py index 4310123..0554213 100644 --- a/aioupnp/gateway.py +++ b/aioupnp/gateway.py @@ -181,7 +181,7 @@ class Gateway: not_met = [ required for required in required_commands if required not in gateway._registered_commands ] - log.warning("found gateway %s at %s, but it does not implement required soap commands: %s", + log.debug("found gateway %s at %s, but it does not implement required soap commands: %s", gateway.manufacturer_string, gateway.location, not_met) ignored.add(datagram.location) continue diff --git a/aioupnp/upnp.py b/aioupnp/upnp.py index 8c86cd9..d5b7d4e 100644 --- a/aioupnp/upnp.py +++ b/aioupnp/upnp.py @@ -60,18 +60,20 @@ class UPnP: @classmethod @cli async def m_search(cls, lan_address: str = '', gateway_address: str = '', timeout: int = 1, - igd_args: OrderedDict = None, interface_name: str = 'default') -> Dict: - try: - lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name) - assert gateway_address and lan_address - except Exception as err: - raise UPnPError("failed to get lan and gateway addresses for interface \"%s\": %s" % (interface_name, - str(err))) + igd_args: OrderedDict = None, unicast: bool = True, interface_name: str = 'default', + loop=None) -> Dict: + if not lan_address or not gateway_address: + try: + lan_address, gateway_address = cls.get_lan_and_gateway(lan_address, gateway_address, interface_name) + assert gateway_address and lan_address + except Exception as err: + raise UPnPError("failed to get lan and gateway addresses for interface \"%s\": %s" % (interface_name, + str(err))) if not igd_args: - igd_args, datagram = await fuzzy_m_search(lan_address, gateway_address, timeout) + igd_args, datagram = await fuzzy_m_search(lan_address, gateway_address, timeout, loop, unicast=unicast) else: igd_args = OrderedDict(igd_args) - datagram = await m_search(lan_address, gateway_address, igd_args, timeout) + datagram = await m_search(lan_address, gateway_address, igd_args, timeout, loop, unicast=unicast) return { 'lan_address': lan_address, 'gateway_address': gateway_address, @@ -340,7 +342,7 @@ class UPnP: @classmethod def run_cli(cls, method, igd_args: OrderedDict, lan_address: str = '', gateway_address: str = '', timeout: int = 30, - interface_name: str = 'default', kwargs: dict = None) -> None: + interface_name: str = 'default', unicast: bool = True, kwargs: dict = None, loop=None) -> None: """ :param method: the command name :param igd_args: ordered case sensitive M-SEARCH headers, if provided all headers to be used must be provided @@ -349,18 +351,19 @@ class UPnP: :param timeout: timeout, in seconds :param interface_name: name of the network interface, the default is aliased to 'default' :param kwargs: keyword arguments for the command + :param loop: EventLoop, used for testing """ kwargs = kwargs or {} igd_args = igd_args timeout = int(timeout) - loop = asyncio.get_event_loop_policy().get_event_loop() + loop = loop or asyncio.get_event_loop_policy().get_event_loop() fut: asyncio.Future = asyncio.Future() async def wrapper(): # wrap the upnp setup and call of the command in a coroutine if method == 'm_search': # if we're only m_searching don't do any device discovery fn = lambda *_a, **_kw: cls.m_search( - lan_address, gateway_address, timeout, igd_args, interface_name + lan_address, gateway_address, timeout, igd_args, unicast, interface_name, loop ) else: # automatically discover the gateway try: diff --git a/setup.py b/setup.py index 093d062..8325f40 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ setup( long_description_content_type='text/markdown', url="https://github.com/lbryio/aioupnp", license=__license__, - packages=find_packages(), + packages=find_packages(exclude=('tests',)), entry_points={'console_scripts': console_scripts}, install_requires=[ 'netifaces', diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..dd299fc --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,113 @@ +import asyncio +import unittest +from unittest.case import _Outcome + +try: + from asyncio.runners import _cancel_all_tasks +except ImportError: + # this is only available in py3.7 + def _cancel_all_tasks(loop): + pass + + +class TestBase(unittest.TestCase): + # Implementation inspired by discussion: + # https://bugs.python.org/issue32972 + + async def asyncSetUp(self): + pass + + async def asyncTearDown(self): + pass + + async def doAsyncCleanups(self): + pass + + def run(self, result=None): + orig_result = result + if result is None: + result = self.defaultTestResult() + startTestRun = getattr(result, 'startTestRun', None) + if startTestRun is not None: + startTestRun() + + result.startTest(self) + + testMethod = getattr(self, self._testMethodName) + if (getattr(self.__class__, "__unittest_skip__", False) or + getattr(testMethod, "__unittest_skip__", False)): + # If the class or method was skipped. + try: + skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') + or getattr(testMethod, '__unittest_skip_why__', '')) + self._addSkip(result, self, skip_why) + finally: + result.stopTest(self) + return + expecting_failure_method = getattr(testMethod, + "__unittest_expecting_failure__", False) + expecting_failure_class = getattr(self, + "__unittest_expecting_failure__", False) + expecting_failure = expecting_failure_class or expecting_failure_method + outcome = _Outcome(result) + 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() + 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.doCleanups() + + for test, reason in outcome.skipped: + self._addSkip(result, test, reason) + self._feedErrorsToResult(result, outcome.errors) + if outcome.success: + if expecting_failure: + if outcome.expectedFailure: + self._addExpectedFailure(result, outcome.expectedFailure) + else: + self._addUnexpectedSuccess(result) + else: + result.addSuccess(self) + return result + finally: + result.stopTest(self) + if orig_result is None: + stopTestRun = getattr(result, 'stopTestRun', None) + if stopTestRun is not None: + stopTestRun() + + # explicitly break reference cycles: + # outcome.errors -> frame -> outcome -> outcome.errors + # outcome.expectedFailure -> frame -> outcome -> outcome.expectedFailure + outcome.errors.clear() + outcome.expectedFailure = None + + # clear the outcome, no more needed + self._outcome = None + + def setUp(self): + self.loop = asyncio.get_event_loop_policy().get_event_loop() diff --git a/tests/protocols/mocks.py b/tests/mocks.py similarity index 54% rename from tests/protocols/mocks.py rename to tests/mocks.py index 52f1d5e..c74c691 100644 --- a/tests/protocols/mocks.py +++ b/tests/mocks.py @@ -75,3 +75,63 @@ def mock_tcp_endpoint_factory(loop, replies=None, delay_reply=0.0, sent_packets= mock_socket.return_value = mock_sock loop.create_connection = create_connection yield + + +@contextlib.contextmanager +def mock_tcp_and_udp(loop, udp_expected_addr, udp_replies=None, udp_delay_reply=0.0, sent_udp_packets=None, + tcp_replies=None, tcp_delay_reply=0.0, tcp_sent_packets=None): + sent_udp_packets = sent_udp_packets if sent_udp_packets is not None else [] + udp_replies = udp_replies or {} + + tcp_sent_packets = tcp_sent_packets if tcp_sent_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): + tcp_sent_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/__init__.py b/tests/protocols/__init__.py index dd299fc..e69de29 100644 --- a/tests/protocols/__init__.py +++ b/tests/protocols/__init__.py @@ -1,113 +0,0 @@ -import asyncio -import unittest -from unittest.case import _Outcome - -try: - from asyncio.runners import _cancel_all_tasks -except ImportError: - # this is only available in py3.7 - def _cancel_all_tasks(loop): - pass - - -class TestBase(unittest.TestCase): - # Implementation inspired by discussion: - # https://bugs.python.org/issue32972 - - async def asyncSetUp(self): - pass - - async def asyncTearDown(self): - pass - - async def doAsyncCleanups(self): - pass - - def run(self, result=None): - orig_result = result - if result is None: - result = self.defaultTestResult() - startTestRun = getattr(result, 'startTestRun', None) - if startTestRun is not None: - startTestRun() - - result.startTest(self) - - testMethod = getattr(self, self._testMethodName) - if (getattr(self.__class__, "__unittest_skip__", False) or - getattr(testMethod, "__unittest_skip__", False)): - # If the class or method was skipped. - try: - skip_why = (getattr(self.__class__, '__unittest_skip_why__', '') - or getattr(testMethod, '__unittest_skip_why__', '')) - self._addSkip(result, self, skip_why) - finally: - result.stopTest(self) - return - expecting_failure_method = getattr(testMethod, - "__unittest_expecting_failure__", False) - expecting_failure_class = getattr(self, - "__unittest_expecting_failure__", False) - expecting_failure = expecting_failure_class or expecting_failure_method - outcome = _Outcome(result) - 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() - 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.doCleanups() - - for test, reason in outcome.skipped: - self._addSkip(result, test, reason) - self._feedErrorsToResult(result, outcome.errors) - if outcome.success: - if expecting_failure: - if outcome.expectedFailure: - self._addExpectedFailure(result, outcome.expectedFailure) - else: - self._addUnexpectedSuccess(result) - else: - result.addSuccess(self) - return result - finally: - result.stopTest(self) - if orig_result is None: - stopTestRun = getattr(result, 'stopTestRun', None) - if stopTestRun is not None: - stopTestRun() - - # explicitly break reference cycles: - # outcome.errors -> frame -> outcome -> outcome.errors - # outcome.expectedFailure -> frame -> outcome -> outcome.expectedFailure - outcome.errors.clear() - outcome.expectedFailure = None - - # clear the outcome, no more needed - self._outcome = None - - def setUp(self): - self.loop = asyncio.get_event_loop_policy().get_event_loop() diff --git a/tests/protocols/test_scpd.py b/tests/protocols/test_scpd.py index 5ba0af9..b0f4d28 100644 --- a/tests/protocols/test_scpd.py +++ b/tests/protocols/test_scpd.py @@ -1,7 +1,7 @@ from aioupnp.fault import UPnPError from aioupnp.protocols.scpd import scpd_post, scpd_get -from . import TestBase -from .mocks import mock_tcp_endpoint_factory +from tests import TestBase +from tests.mocks import mock_tcp_endpoint_factory class TestSCPDGet(TestBase): diff --git a/tests/protocols/test_ssdp.py b/tests/protocols/test_ssdp.py index 79df31b..47c342b 100644 --- a/tests/protocols/test_ssdp.py +++ b/tests/protocols/test_ssdp.py @@ -4,8 +4,8 @@ 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 . import TestBase -from .mocks import mock_datagram_endpoint_factory +from tests import TestBase +from tests.mocks import mock_datagram_endpoint_factory class TestSSDP(TestBase): diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..91def3e --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,104 @@ +import contextlib +from io import StringIO +from tests import TestBase +from tests.mocks import mock_tcp_and_udp +from collections import OrderedDict +from aioupnp.serialization.ssdp import SSDPDatagram +from aioupnp.protocols.m_search_patterns import packet_generator +from aioupnp.__main__ import main + + +m_search_cli_result = """{ + "lan_address": "10.0.0.2", + "gateway_address": "10.0.0.1", + "m_search_kwargs": "--HOST=239.255.255.250:1900 --MAN=ssdp:discover --MX=1 --ST=urn:schemas-upnp-org:device:WANDevice:1", + "discover_reply": { + "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:device:WANDevice:1", + "USN": "uuid:22222222-3333-4444-5555-666666666666::urn:schemas-upnp-org:device:WANDevice:1" + } +}\n""" + + +class TestCLI(TestBase): + gateway_address = "10.0.0.1" + soap_port = 49152 + m_search_args = OrderedDict([ + ("HOST", "239.255.255.250:1900"), + ("MAN", "ssdp:discover"), + ("MX", 1), + ("ST", "urn:schemas-upnp-org:device:WANDevice:1") + ]) + reply = SSDPDatagram("OK", OrderedDict([ + ("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:device:WANDevice:1"), + ("USN", "uuid:11111111-2222-3333-4444-555555555555::urn:schemas-upnp-org:device:WANDevice:1") + ])) + + scpd_replies = { + b'GET /InternetGatewayDevice.xml HTTP/1.1\r\nAccept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n': b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Thu, 11 Oct 2018 22:16:16 GMT\r\nContent-Type: text/xml\r\nContent-Length: 3921\r\nLast-Modified: Thu, 09 Aug 2018 12:41:07 GMT\r\nConnection: close\r\n\r\n\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", + b'GET /OSInfo.xml HTTP/1.1\r\nAccept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n': b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Thu, 11 Oct 2018 22:16:16 GMT\r\nContent-Type: text/xml\r\nContent-Length: 219\r\nLast-Modified: Thu, 09 Aug 2018 12:41:07 GMT\r\nConnection: close\r\n\r\n\n\n\t\n\t\t1\n\t\t0\n\t\n\t\n\t\n\t\n\t\n\n", + b'GET /Layer3Forwarding.xml HTTP/1.1\r\nAccept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n': b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Thu, 11 Oct 2018 22:16:16 GMT\r\nContent-Type: text/xml\r\nContent-Length: 920\r\nLast-Modified: Thu, 09 Aug 2018 12:41:07 GMT\r\nConnection: close\r\n\r\n\n\n\t\n\t\t1\n\t\t0\n\t\n\t\n\t\t\n\t\t\tGetDefaultConnectionService\n\t\t\t\n\t\t\t\t\n\t\t\t\t\tNewDefaultConnectionService\n\t\t\t\t\tout\n\t\t\t\t\tDefaultConnectionService\n\t\t\t\t\n\t\t\t\n\t\t\n\t\t\n\t\t\tSetDefaultConnectionService\n\t\t\t\n\t\t\t\t\n\t\t\t\t\tNewDefaultConnectionService\n\t\t\t\t\tin\n\t\t\t\t\tDefaultConnectionService\n\t\t\t\t\n\t\t\t\n\t\t\n\t\n\t\n\t\t\n\t\t\tDefaultConnectionService\n\t\t\tstring\n\t\t\n\t\n\n", + b'GET /WANCommonInterfaceConfig.xml HTTP/1.1\r\nAccept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n': b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Thu, 11 Oct 2018 22:16:16 GMT\r\nContent-Type: text/xml\r\nContent-Length: 5343\r\nLast-Modified: Thu, 09 Aug 2018 12:41:07 GMT\r\nConnection: close\r\n\r\n\r\n\r\n\t\r\n\t\t1\r\n\t\t0\r\n\t\r\n\t\r\n\t\t\r\n\t\t\tGetCommonLinkProperties\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewWANAccessType\r\n\t\t\t\t\tout\r\n\t\t\t\t\tWANAccessType\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewLayer1UpstreamMaxBitRate\r\n\t\t\t\t\tout\r\n\t\t\t\t\tLayer1UpstreamMaxBitRate\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewLayer1DownstreamMaxBitRate\r\n\t\t\t\t\tout\r\n\t\t\t\t\tLayer1DownstreamMaxBitRate\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewPhysicalLinkStatus\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPhysicalLinkStatus\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetTotalBytesSent\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewTotalBytesSent\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalBytesSent\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetTotalBytesReceived\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewTotalBytesReceived\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalBytesReceived\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetTotalPacketsSent\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewTotalPacketsSent\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalPacketsSent\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetTotalPacketsReceived\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewTotalPacketsReceived\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalPacketsReceived\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tX_GetICSStatistics\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tTotalBytesSent\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalBytesSent\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tTotalBytesReceived\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalBytesReceived\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tTotalPacketsSent\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalPacketsSent\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tTotalPacketsReceived\r\n\t\t\t\t\tout\r\n\t\t\t\t\tTotalPacketsReceived\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tLayer1DownstreamMaxBitRate\r\n\t\t\t\t\tout\r\n\t\t\t\t\tLayer1DownstreamMaxBitRate\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tUptime\r\n\t\t\t\t\tout\r\n\t\t\t\t\tX_Uptime\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\r\n\t\r\n\t\t\r\n\t\t\tWANAccessType\r\n\t\t\tstring\r\n\t\t\t\r\n\t\t\t\tDSL\r\n\t\t\t\tPOTS\r\n\t\t\t\tCable\r\n\t\t\t\tEthernet\r\n\t\t\t\tOther\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tLayer1UpstreamMaxBitRate\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tLayer1DownstreamMaxBitRate\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tPhysicalLinkStatus\r\n\t\t\tstring\r\n\t\t\t\r\n\t\t\t\tUp\r\n\t\t\t\tDown\r\n\t\t\t\tInitializing\r\n\t\t\t\tUnavailable\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tWANAccessProvider\r\n\t\t\tstring\r\n\t\t\r\n\t\t\r\n\t\t\tMaximumActiveConnections\r\n\t\t\tui2\r\n\t\t\t\r\n\t\t\t\t1\r\n\t\t\t\t\r\n\t\t\t\t1\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tTotalBytesSent\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tTotalBytesReceived\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tTotalPacketsSent\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tTotalPacketsReceived\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tX_PersonalFirewallEnabled\r\n\t\t\tboolean\r\n\t\t\r\n\t\t\r\n\t\t\tX_Uptime\r\n\t\t\tui4\r\n\t\t\r\n\t\r\n\r\n", + b'GET /WANEthernetLinkConfig.xml HTTP/1.1\r\nAccept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n': b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Thu, 11 Oct 2018 22:16:16 GMT\r\nContent-Type: text/xml\r\nContent-Length: 773\r\nLast-Modified: Thu, 09 Aug 2018 12:41:07 GMT\r\nConnection: close\r\n\r\n\n\n\t\n\t\t1\n\t\t0\n\t\n\t\n\t\t\n\t\t\tGetEthernetLinkStatus\n\t\t\t\n\t\t\t\t\n\t\t\t\t\tNewEthernetLinkStatus\n\t\t\t\t\tout\n\t\t\t\t\tEthernetLinkStatus\n\t\t\t\t\n\t\t\t\n\t\t\n\t\n\t\n\t\t\n\t\t\tEthernetLinkStatus\n\t\t\tstring\n\t\t\t\n\t\t\t\tUp\n\t\t\t\tDown\n\t\t\t\tUnavailable\n\t\t\t\n\t\t\n\t\n\n", + b'GET /WANIPConnection.xml HTTP/1.1\r\nAccept-Encoding: gzip\r\nHost: 10.0.0.1\r\nConnection: Close\r\n\r\n': b"HTTP/1.1 200 OK\r\nServer: WebServer\r\nDate: Thu, 11 Oct 2018 22:16:16 GMT\r\nContent-Type: text/xml\r\nContent-Length: 12078\r\nLast-Modified: Thu, 09 Aug 2018 12:41:07 GMT\r\nConnection: close\r\n\r\n\r\n\r\n\t\r\n\t\t1\r\n\t\t0\r\n\t\r\n\t\r\n\t\t\r\n\t\t\tSetConnectionType\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewConnectionType\r\n\t\t\t\t\tin\r\n\t\t\t\t\tConnectionType\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t \r\n\t\t\r\n\t\t\tGetConnectionTypeInfo\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewConnectionType\r\n\t\t\t\t\tout\r\n\t\t\t\t\tConnectionType\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewPossibleConnectionTypes\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPossibleConnectionTypes\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tRequestConnection\r\n\t\t\r\n\t\t\r\n\t\t\tForceTermination\r\n\t\t\r\n\t\t\r\n\t\t\tGetStatusInfo\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewConnectionStatus\r\n\t\t\t\t\tout\r\n\t\t\t\t\tConnectionStatus\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewLastConnectionError\r\n\t\t\t\t\tout\r\n\t\t\t\t\tLastConnectionError\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewUptime\r\n\t\t\t\t\tout\r\n\t\t\t\t\tUptime\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetNATRSIPStatus\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewRSIPAvailable\r\n\t\t\t\t\tout\r\n\t\t\t\t\tRSIPAvailable\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewNATEnabled\r\n\t\t\t\t\tout\r\n\t\t\t\t\tNATEnabled\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetGenericPortMappingEntry\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewPortMappingIndex\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingNumberOfEntries\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewRemoteHost\r\n\t\t\t\t\tout\r\n\t\t\t\t\tRemoteHost\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewExternalPort\r\n\t\t\t\t\tout\r\n\t\t\t\t\tExternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewProtocol\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingProtocol\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewInternalPort\r\n\t\t\t\t\tout\r\n\t\t\t\t\tInternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewInternalClient\r\n\t\t\t\t\tout\r\n\t\t\t\t\tInternalClient\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewEnabled\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingEnabled\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewPortMappingDescription\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingDescription\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewLeaseDuration\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingLeaseDuration\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetSpecificPortMappingEntry\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewRemoteHost\r\n\t\t\t\t\tin\r\n\t\t\t\t\tRemoteHost\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewExternalPort\r\n\t\t\t\t\tin\r\n\t\t\t\t\tExternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewProtocol\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingProtocol\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewInternalPort\r\n\t\t\t\t\tout\r\n\t\t\t\t\tInternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewInternalClient\r\n\t\t\t\t\tout\r\n\t\t\t\t\tInternalClient\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewEnabled\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingEnabled\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewPortMappingDescription\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingDescription\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewLeaseDuration\r\n\t\t\t\t\tout\r\n\t\t\t\t\tPortMappingLeaseDuration\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tAddPortMapping\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewRemoteHost\r\n\t\t\t\t\tin\r\n\t\t\t\t\tRemoteHost\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewExternalPort\r\n\t\t\t\t\tin\r\n\t\t\t\t\tExternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewProtocol\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingProtocol\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewInternalPort\r\n\t\t\t\t\tin\r\n\t\t\t\t\tInternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewInternalClient\r\n\t\t\t\t\tin\r\n\t\t\t\t\tInternalClient\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewEnabled\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingEnabled\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewPortMappingDescription\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingDescription\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewLeaseDuration\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingLeaseDuration\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tDeletePortMapping\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewRemoteHost\r\n\t\t\t\t\tin\r\n\t\t\t\t\tRemoteHost\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewExternalPort\r\n\t\t\t\t\tin\r\n\t\t\t\t\tExternalPort\r\n\t\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewProtocol\r\n\t\t\t\t\tin\r\n\t\t\t\t\tPortMappingProtocol\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tGetExternalIPAddress\r\n\t\t\t\r\n\t\t\t\t\r\n\t\t\t\t\tNewExternalIPAddress\r\n\t\t\t\t\tout\r\n\t\t\t\t\tExternalIPAddress\r\n\t\t\t\t\r\n\t\t\t\r\n\t\t\r\n\t\r\n\t\r\n\t\t\r\n\t\t\tConnectionType\r\n\t\t\tstring\r\n\t\t\tUnconfigured\r\n\t\t\r\n\t\t\r\n\t\t\tPossibleConnectionTypes\r\n\t\t\tstring\r\n\t\t\t\r\n\t\t\t\tUnconfigured\r\n\t\t\t\tIP_Routed\r\n\t\t\t\tIP_Bridged\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tConnectionStatus\r\n\t\t\tstring\r\n\t\t\tUnconfigured\r\n\t\t\t\r\n\t\t\t\tUnconfigured\r\n\t\t\t\tConnecting\r\n\t\t\t\tAuthenticating\r\n\t\t\t\tPendingDisconnect\r\n\t\t\t\tDisconnecting\r\n\t\t\t\tDisconnected\r\n\t\t\t\tConnected\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tUptime\r\n\t\t\tui4\r\n\t\t\t0\r\n\t\t\t\r\n\t\t\t\t0\r\n\t\t\t\t\r\n\t\t\t\t1\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tRSIPAvailable\r\n\t\tboolean\r\n\t\t\t0\r\n\t\t\r\n\t\t\r\n\t\t\tNATEnabled\r\n\t\t\tboolean\r\n\t\t\t1\r\n\t\t \r\n\t\t\r\n\t\t\tX_Name\r\n\t\t\tstring\r\n\t\t\r\n\t\t\r\n\t\t\tLastConnectionError\r\n\t\t\tstring\r\n\t\t\tERROR_NONE\r\n\t\t\t\r\n\t\t\t\tERROR_NONE\r\n\t\t\t\tERROR_ISP_TIME_OUT\r\n\t\t\t\tERROR_COMMAND_ABORTED\r\n\t\t\t\tERROR_NOT_ENABLED_FOR_INTERNET\r\n\t\t\t\tERROR_BAD_PHONE_NUMBER\r\n\t\t\t\tERROR_USER_DISCONNECT\r\n\t\t\t\tERROR_ISP_DISCONNECT\r\n\t\t\t\tERROR_IDLE_DISCONNECT\r\n\t\t\t\tERROR_FORCED_DISCONNECT\r\n\t\t\t\tERROR_SERVER_OUT_OF_RESOURCES\r\n\t\t\t\tERROR_RESTRICTED_LOGON_HOURS\r\n\t\t\t\tERROR_ACCOUNT_DISABLED\r\n\t\t\t\tERROR_ACCOUNT_EXPIRED\r\n\t\t\t\tERROR_PASSWORD_EXPIRED\r\n\t\t\t\tERROR_AUTHENTICATION_FAILURE\r\n\t\t\t\tERROR_NO_DIALTONE\r\n\t\t\t\tERROR_NO_CARRIER\r\n\t\t\t\tERROR_NO_ANSWER\r\n\t\t\t\tERROR_LINE_BUSY\r\n\t\t\t\tERROR_UNSUPPORTED_BITSPERSECOND\r\n\t\t\t\tERROR_TOO_MANY_LINE_ERRORS\r\n\t\t\t\tERROR_IP_CONFIGURATION\r\n\t\t\t\tERROR_UNKNOWN\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tExternalIPAddress\r\n\t\t\tstring\r\n\t\t\r\n\t\t\r\n\t\t\tRemoteHost\r\n\t\t\tstring\r\n\t\t\r\n\t\t\r\n\t\t\tExternalPort\r\n\t\t\tui2\r\n\t\t\r\n\t\t\r\n\t\t\tInternalPort\r\n\t\t\tui2\r\n\t\t\r\n\t\t\r\n\t\t\tPortMappingProtocol\r\n\t\t\tstring\r\n\t\t\t\r\n\t\t\t\tTCP\r\n\t\t\t\tUDP\r\n\t\t\t\r\n\t\t\r\n\t\t\r\n\t\t\tInternalClient\r\n\t\t\tstring\r\n\t\t\r\n\t\t\r\n\t\t\tPortMappingDescription\r\n\t\t\tstring\r\n\t\t\r\n\t\t\r\n\t\t\tPortMappingEnabled\r\n\t\t\tboolean\r\n\t\t\r\n\t\t\r\n\t\t\tPortMappingLeaseDuration\r\n\t\t\tui4\r\n\t\t\r\n\t\t\r\n\t\t\tPortMappingNumberOfEntries\r\n\t\t\tui2\r\n\t\t\r\n\t\r\n\r\n", + b'POST /soap.cgi?service=WANIPConn1 HTTP/1.1\r\nHost: 10.0.0.1\r\nUser-Agent: python3/aioupnp, UPnP/1.0, MiniUPnPc/1.9\r\nContent-Length: 285\r\nContent-Type: text/xml\r\nSOAPAction: "urn:schemas-upnp-org:service:WANIPConnection:1#GetExternalIPAddress"\r\nConnection: Close\r\nCache-Control: no-cache\r\nPragma: no-cache\r\n\r\n\r\n\r\n': b"HTTP/1.1 200 OK\r\nCONTENT-LENGTH: 340\r\nCONTENT-TYPE: text/xml; charset=\"utf-8\"\r\nDATE: Thu, 18 Oct 2018 01:20:23 GMT\r\nEXT:\r\nSERVER: Linux/3.14.28-Prod_17.2, UPnP/1.0, Portable SDK for UPnP devices/1.6.22\r\nX-User-Agent: redsonic\r\n\r\n\n\r\n11.22.33.44\r\n\r\n " + } + + packet_args = list(packet_generator()) + byte_packets = [SSDPDatagram("M-SEARCH", p).encode().encode() for p in packet_args] + + successful_args = OrderedDict([ + ("HOST", "239.255.255.250:1900"), + ("MAN", "ssdp:discover"), + ("MX", 1), + ("ST", "urn:schemas-upnp-org:device:WANDevice:1") + ]) + query_packet = SSDPDatagram("M-SEARCH", successful_args) + + reply_args = OrderedDict([ + ("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:device:WANDevice:1"), + ("USN", "uuid:22222222-3333-4444-5555-666666666666::urn:schemas-upnp-org:device:WANDevice:1") + ]) + reply_packet = SSDPDatagram("OK", reply_args) + + udp_replies = { + (query_packet.encode().encode(), ("10.0.0.1", 1900)): reply_packet.encode().encode() + } + + def test_get_external_ip(self): + actual_output = StringIO() + 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'), + self.loop + ) + self.assertEqual("11.22.33.44\n", actual_output.getvalue()) + + def test_m_search(self): + actual_output = StringIO() + timeout_msg = "aioupnp encountered an error:\nM-SEARCH for 10.0.0.1:1900 timed out\n" + 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'), + self.loop + ) + self.assertEqual(timeout_msg, actual_output.getvalue()) + + actual_output = StringIO() + 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'), + self.loop + ) + self.assertEqual(m_search_cli_result, actual_output.getvalue()) diff --git a/tests/protocols/test_gateway.py b/tests/test_gateway.py similarity index 99% rename from tests/protocols/test_gateway.py rename to tests/test_gateway.py index 462a97f..6d73f98 100644 --- a/tests/protocols/test_gateway.py +++ b/tests/test_gateway.py @@ -1,11 +1,12 @@ from aioupnp.fault import UPnPError from aioupnp.protocols.scpd import scpd_post, scpd_get -from . import TestBase -from .mocks import mock_tcp_endpoint_factory +from tests import TestBase +from tests.mocks import mock_tcp_endpoint_factory from collections import OrderedDict from aioupnp.gateway import Gateway from aioupnp.serialization.ssdp import SSDPDatagram + class TestDiscoverCommands(TestBase): gateway_address = "10.0.0.1" soap_port = 49152