mypy refactor, improve coverage #12

Merged
jackrobison merged 9 commits from mypy-refactor into master 2019-05-22 09:05:10 +02:00
13 changed files with 582 additions and 125 deletions
Showing only changes of commit 836271c6e0 - Show all commits

View file

@ -4,3 +4,4 @@ __author__ = "Jack Robison"
__maintainer__ = "Jack Robison" __maintainer__ = "Jack Robison"
__license__ = "MIT" __license__ = "MIT"
__email__ = "jackrobison@lbry.io" __email__ = "jackrobison@lbry.io"

View file

@ -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()

View file

@ -36,8 +36,9 @@ version = '0.10.7'
# functions # functions
def gateways(*args, **kwargs) -> typing.Dict[typing.Union[str, int],
def gateways(*args, **kwargs) -> typing.List: # real signature unknown 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. Obtain a list of the gateways on this machine.
@ -56,7 +57,7 @@ def gateways(*args, **kwargs) -> typing.List: # real signature unknown
pass 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. Obtain information about the specified network interface.
@ -67,7 +68,7 @@ def ifaddresses(*args, **kwargs) -> typing.Dict: # real signature unknown
pass 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. """ """ Obtain a list of the interfaces available on this machine. """
pass pass

View file

@ -1,7 +1,11 @@
import asyncio import asyncio
import unittest import unittest
import contextlib
import socket
from unittest import mock
from unittest.case import _Outcome from unittest.case import _Outcome
try: try:
from asyncio.runners import _cancel_all_tasks from asyncio.runners import _cancel_all_tasks
except ImportError: except ImportError:
@ -10,30 +14,89 @@ except ImportError:
pass 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: # Implementation inspired by discussion:
# https://bugs.python.org/issue32972 # https://bugs.python.org/issue32972
async def asyncSetUp(self): maxDiff = None
async def asyncSetUp(self): # pylint: disable=C0103
pass pass
async def asyncTearDown(self): async def asyncTearDown(self): # pylint: disable=C0103
pass pass
async def doAsyncCleanups(self): def run(self, result=None): # pylint: disable=R0915
pass
def run(self, result=None):
orig_result = result orig_result = result
if result is None: if result is None:
result = self.defaultTestResult() result = self.defaultTestResult()
startTestRun = getattr(result, 'startTestRun', None) startTestRun = getattr(result, 'startTestRun', None) # pylint: disable=C0103
if startTestRun is not None: if startTestRun is not None:
startTestRun() startTestRun()
result.startTest(self) result.startTest(self)
testMethod = getattr(self, self._testMethodName) testMethod = getattr(self, self._testMethodName) # pylint: disable=C0103
if (getattr(self.__class__, "__unittest_skip__", False) or if (getattr(self.__class__, "__unittest_skip__", False) or
getattr(testMethod, "__unittest_skip__", False)): getattr(testMethod, "__unittest_skip__", False)):
# If the class or method was skipped. # If the class or method was skipped.
@ -50,36 +113,36 @@ class TestBase(unittest.TestCase):
"__unittest_expecting_failure__", False) "__unittest_expecting_failure__", False)
expecting_failure = expecting_failure_class or expecting_failure_method expecting_failure = expecting_failure_class or expecting_failure_method
outcome = _Outcome(result) outcome = _Outcome(result)
self.loop = asyncio.new_event_loop() # pylint: disable=W0201
asyncio.set_event_loop(self.loop)
self.loop.set_debug(True)
try: try:
self._outcome = outcome self._outcome = outcome
loop = asyncio.new_event_loop() with outcome.testPartExecutor(self):
try: self.setUp()
asyncio.set_event_loop(loop) self.loop.run_until_complete(self.asyncSetUp())
loop.set_debug(True) 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): with outcome.testPartExecutor(self):
self.setUp() self.loop.run_until_complete(self.asyncTearDown())
loop.run_until_complete(self.asyncSetUp()) self.tearDown()
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() 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: for test, reason in outcome.skipped:
self._addSkip(result, test, reason) self._addSkip(result, test, reason)
@ -96,9 +159,9 @@ class TestBase(unittest.TestCase):
finally: finally:
result.stopTest(self) result.stopTest(self)
if orig_result is None: if orig_result is None:
stopTestRun = getattr(result, 'stopTestRun', None) stopTestRun = getattr(result, 'stopTestRun', None) # pylint: disable=C0103
if stopTestRun is not None: if stopTestRun is not None:
stopTestRun() stopTestRun() # pylint: disable=E1102
# explicitly break reference cycles: # explicitly break reference cycles:
# outcome.errors -> frame -> outcome -> outcome.errors # outcome.errors -> frame -> outcome -> outcome.errors
@ -109,5 +172,11 @@ class TestBase(unittest.TestCase):
# clear the outcome, no more needed # clear the outcome, no more needed
self._outcome = None self._outcome = None
def setUp(self): def doAsyncCleanups(self): # pylint: disable=C0103
self.loop = asyncio.get_event_loop_policy().get_event_loop() 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)

View file

@ -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

View file

@ -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()

View file

@ -1,10 +1,9 @@
from aioupnp.fault import UPnPError from aioupnp.fault import UPnPError
from aioupnp.protocols.scpd import scpd_post, scpd_get from aioupnp.protocols.scpd import scpd_post, scpd_get
from tests import TestBase from tests import AsyncioTestCase, mock_tcp_and_udp
from tests.mocks import mock_tcp_and_udp
class TestSCPDGet(TestBase): class TestSCPDGet(AsyncioTestCase):
path, lan_address, port = '/IGDdevicedesc_brlan0.xml', '10.1.10.1', 49152 path, lan_address, port = '/IGDdevicedesc_brlan0.xml', '10.1.10.1', 49152
get_request = b'GET /IGDdevicedesc_brlan0.xml HTTP/1.1\r\n' \ 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' 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')) self.assertTrue(str(err).startswith('too many bytes written'))
class TestSCPDPost(TestBase): class TestSCPDPost(AsyncioTestCase):
param_names: list = [] param_names: list = []
kwargs: dict = {} kwargs: dict = {}
method, gateway_address, port = "GetExternalIPAddress", '10.0.0.1', 49152 method, gateway_address, port = "GetExternalIPAddress", '10.0.0.1', 49152

View file

@ -4,11 +4,10 @@ from aioupnp.protocols.m_search_patterns import packet_generator
from aioupnp.serialization.ssdp import SSDPDatagram from aioupnp.serialization.ssdp import SSDPDatagram
from aioupnp.constants import SSDP_IP_ADDRESS from aioupnp.constants import SSDP_IP_ADDRESS
from aioupnp.protocols.ssdp import fuzzy_m_search, m_search from aioupnp.protocols.ssdp import fuzzy_m_search, m_search
from tests import TestBase from tests import AsyncioTestCase, mock_tcp_and_udp
from tests.mocks import mock_tcp_and_udp
class TestSSDP(TestBase): class TestSSDP(AsyncioTestCase):
packet_args = list(packet_generator()) packet_args = list(packet_generator())
byte_packets = [SSDPDatagram("M-SEARCH", p).encode().encode() for p in packet_args] byte_packets = [SSDPDatagram("M-SEARCH", p).encode().encode() for p in packet_args]

View file

@ -1,5 +1,7 @@
import unittest import unittest
from aioupnp.fault import UPnPError
from aioupnp.serialization.scpd import serialize_scpd_get, deserialize_scpd_get_response 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.device import Device
from aioupnp.util import get_dict_val_case_insensitive from aioupnp.util import get_dict_val_case_insensitive
@ -20,6 +22,28 @@ class TestSCPDSerialization(unittest.TestCase):
b"\r\n" \ b"\r\n" \
b"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp-org:device-1-0\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevice:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n <device>\n <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n <friendlyName>WANConnectionDevice:1</friendlyName>\n <manufacturer>Cisco</manufacturer>\n <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n <modelDescription>CGA4131COM</modelDescription>\n <modelName>CGA4131COM</modelName>\n <modelNumber>CGA4131COM</modelNumber>\n <modelURL>http://www.cisco.com</modelURL>\n <serialNumber></serialNumber>\n <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n <UPC>CGA4131COM</UPC>\n <serviceList>\n <service>\n <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n <controlURL>/upnp/control/WANIPConnection0</controlURL>\n <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n </service>\n </serviceList>\n </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n" b"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp-org:device-1-0\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevice:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n <device>\n <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n <friendlyName>WANConnectionDevice:1</friendlyName>\n <manufacturer>Cisco</manufacturer>\n <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n <modelDescription>CGA4131COM</modelDescription>\n <modelName>CGA4131COM</modelName>\n <modelNumber>CGA4131COM</modelNumber>\n <modelURL>http://www.cisco.com</modelURL>\n <serialNumber></serialNumber>\n <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n <UPC>CGA4131COM</UPC>\n <serviceList>\n <service>\n <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n <controlURL>/upnp/control/WANIPConnection0</controlURL>\n <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n </service>\n </serviceList>\n </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n"
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"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp-org:device-1-?\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevic3:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n <device>\n <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n <friendlyName>WANConnectionDevice:1</friendlyName>\n <manufacturer>Cisco</manufacturer>\n <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n <modelDescription>CGA4131COM</modelDescription>\n <modelName>CGA4131COM</modelName>\n <modelNumber>CGA4131COM</modelNumber>\n <modelURL>http://www.cisco.com</modelURL>\n <serialNumber></serialNumber>\n <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n <UPC>CGA4131COM</UPC>\n <serviceList>\n <service>\n <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n <controlURL>/upnp/control/WANIPConnection0</controlURL>\n <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n </service>\n </serviceList>\n </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n"
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"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp--org:device-1-0\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevic3:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n <device>\n <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n <friendlyName>WANConnectionDevice:1</friendlyName>\n <manufacturer>Cisco</manufacturer>\n <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n <modelDescription>CGA4131COM</modelDescription>\n <modelName>CGA4131COM</modelName>\n <modelNumber>CGA4131COM</modelNumber>\n <modelURL>http://www.cisco.com</modelURL>\n <serialNumber></serialNumber>\n <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n <UPC>CGA4131COM</UPC>\n <serviceList>\n <service>\n <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n <controlURL>/upnp/control/WANIPConnection0</controlURL>\n <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n </service>\n </serviceList>\n </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n"
expected_parsed = { expected_parsed = {
'specVersion': {'major': '1', 'minor': '0'}, 'specVersion': {'major': '1', 'minor': '0'},
'device': { 'device': {
@ -94,6 +118,87 @@ class TestSCPDSerialization(unittest.TestCase):
def test_serialize_get(self): 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, 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('<?xml version="1.0"?>\n<root xmlns="urn:schemas-upnp-org:device-1-0">\n\t<specVersion>\n\t\t<major>1</major>\n\t\t<minor>0</minor>\n\t</specVersion>\n\t<URLBase>http://10.0.0.1:49152</URLBase>\n\t<device>\n\t\t<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevice:1</deviceType>\n\t\t<friendlyName>Wireless Broadband Router</friendlyName>\n\t\t<manufacturer>D-Link Corporation</manufacturer>\n\t\t<manufacturerURL>http://www.dlink.com</manufacturerURL>\n\t\t<modelDescription>D-Link Router</modelDescription>\n\t\t<modelName>D-Link Router</modelName>\n\t\t<modelNumber>DIR-890L</modelNumber>\n\t\t<modelURL>http://www.dlink.com</modelURL>\n\t\t<serialNumber>120</serialNumber>\n\t\t<UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n\t\t<iconList>\n\t\t\t<icon>\n\t\t\t\t<mimetype>image/gif</mimetype>\n\t\t\t\t<width>118</width>\n\t\t\t\t<height>119</height>\n\t\t\t\t<depth>8</depth>\n\t\t\t\t<url>/ligd.gif</url>\n\t\t\t</icon>\n\t\t</iconList>\n\t\t<serviceList>\n\t\t\t<service>\n\t\t\t\t<serviceType>urn:schemas-microsoft-com:service:OSInfo:1</serviceType>\n\t\t\t\t<serviceId>urn:microsoft-com:serviceId:OSInfo1</serviceId>\n\t\t\t\t<controlURL>/soap.cgi?service=OSInfo1</controlURL>\n\t\t\t\t<eventSubURL>/gena.cgi?service=OSInfo1</eventSubURL>\n\t\t\t\t<SCPDURL>/OSInfo.xml</SCPDURL>\n\t\t\t</service>\n\t\t\t<service>\n\t\t\t\t<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n\t\t\t\t<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n\t\t\t\t<controlURL>/soap.cgi?service=L3Forwarding1</controlURL>\n\t\t\t\t<eventSubURL>/gena.cgi?service=L3Forwarding1</eventSubURL>\n\t\t\t\t<SCPDURL>/Layer3Forwarding.xml</SCPDURL>\n\t\t\t</service>\n\t\t</serviceList>\n\t\t<deviceList>\n\t\t\t<device>\n\t\t\t\t<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n\t\t\t\t<friendlyName>WANDevice</friendlyName>\n\t\t\t\t<manufacturer>D-Link</manufacturer>\n\t\t\t\t<manufacturerURL>http://www.dlink.com</manufacturerURL>\n\t\t\t\t<modelDescription>WANDevice</modelDescription>\n\t\t\t\t<modelName>DIR-890L</modelName>\n\t\t\t\t<modelNumber>1</modelNumber>\n\t\t\t\t<modelURL>http://www.dlink.com</modelURL>\n\t\t\t\t<serialNumber>120</serialNumber>\n\t\t\t\t<UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n\t\t\t\t<serviceList>\n\t\t\t\t\t<service>\n\t\t\t\t\t\t<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n\t\t\t\t\t\t<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n\t\t\t\t\t\t<controlURL>/soap.cgi?service=WANCommonIFC1</controlURL>\n\t\t\t\t\t\t<eventSubURL>/gena.cgi?service=WANCommonIFC1</eventSubURL>\n\t\t\t\t\t\t<SCPDURL>/WANCommonInterfaceConfig.xml</SCPDURL>\n\t\t\t\t\t</service>\n\t\t\t\t</serviceList>\n\t\t\t\t<deviceList>\n\t\t\t\t\t<device>\n\t\t\t\t\t\t<deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n\t\t\t\t\t\t<friendlyName>WANConnectionDevice</friendlyName>\n\t\t\t\t\t\t<manufacturer>D-Link</manufacturer>\n\t\t\t\t\t\t<manufacturerURL>http://www.dlink.com</manufacturerURL>\n\t\t\t\t\t\t<modelDescription>WanConnectionDevice</modelDescription>\n\t\t\t\t\t\t<modelName>DIR-890L</modelName>\n\t\t\t\t\t\t<modelNumber>1</modelNumber>\n\t\t\t\t\t\t<modelURL>http://www.dlink.com</modelURL>\n\t\t\t\t\t\t<serialNumber>120</serialNumber>\n\t\t\t\t\t\t<UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n\t\t\t\t\t\t<serviceList>\n\t\t\t\t\t\t\t<service>\n\t\t\t\t\t\t\t\t<serviceType>urn:schemas-upnp-org:service:WANEthernetLinkConfig:1</serviceType>\n\t\t\t\t\t\t\t\t<serviceId>urn:upnp-org:serviceId:WANEthLinkC1</serviceId>\n\t\t\t\t\t\t\t\t<controlURL>/soap.cgi?service=WANEthLinkC1</controlURL>\n\t\t\t\t\t\t\t\t<eventSubURL>/gena.cgi?service=WANEthLinkC1</eventSubURL>\n\t\t\t\t\t\t\t\t<SCPDURL>/WANEthernetLinkConfig.xml</SCPDURL>\n\t\t\t\t\t\t\t</service>\n\t\t\t\t\t\t\t<service>\n\t\t\t\t\t\t\t\t<serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n\t\t\t\t\t\t\t\t<serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n\t\t\t\t\t\t\t\t<controlURL>/soap.cgi?service=WANIPConn1</controlURL>\n\t\t\t\t\t\t\t\t<eventSubURL>/gena.cgi?service=WANIPConn1</eventSubURL>\n\t\t\t\t\t\t\t\t<SCPDURL>/WANIPConnection.xml</SCPDURL>\n\t\t\t\t\t\t\t</service>\n\t\t\t\t\t\t</serviceList>\n\t\t\t\t\t</device>\n\t\t\t\t</deviceList>\n\t\t\t</device>\n\t\t</deviceList>\n\t\t<presentationURL>http://10.0.0.1</presentationURL>\n\t</device>\n</root>\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): def test_deserialize_get_response(self):
self.assertDictEqual(deserialize_scpd_get_response(self.response), self.expected_parsed) self.assertDictEqual(deserialize_scpd_get_response(self.response), self.expected_parsed)
@ -101,6 +206,14 @@ class TestSCPDSerialization(unittest.TestCase):
def test_deserialize_blank(self): def test_deserialize_blank(self):
self.assertDictEqual(deserialize_scpd_get_response(b''), {}) 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): def test_deserialize_to_device_object(self):
devices = [] devices = []
services = [] services = []
@ -173,3 +286,78 @@ class TestSCPDSerialization(unittest.TestCase):
}, 'presentationURL': 'http://10.1.10.1/' }, 'presentationURL': 'http://10.1.10.1/'
} }
self.assertDictEqual(expected_result, device.as_dict()) self.assertDictEqual(expected_result, device.as_dict())
def test_deserialize_another_device(self):
xml_bytes = b"<?xml version=\"1.0\"?>\n<root xmlns=\"urn:schemas-upnp-org:device-1-0\">\n<specVersion>\n<major>1</major>\n<minor>0</minor>\n</specVersion>\n<device>\n<deviceType>urn:schemas-upnp-org:device:InternetGatewayDevice:1</deviceType>\n<friendlyName>CGA4131COM</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:11111111-2222-3333-4444-555555555556</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:Layer3Forwarding:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:L3Forwarding1</serviceId>\n<SCPDURL>/Layer3ForwardingSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/Layer3Forwarding</controlURL>\n<eventSubURL>/upnp/event/Layer3Forwarding</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n<device>\n<deviceType>urn:schemas-upnp-org:device:WANDevice:1</deviceType>\n<friendlyName>WANDevice:1</friendlyName>\n<manufacturer>Cisco</manufacturer>\n<manufacturerURL>http://www.cisco.com/</manufacturerURL>\n<modelDescription>CGA4131COM</modelDescription>\n<modelName>CGA4131COM</modelName>\n<modelNumber>CGA4131COM</modelNumber>\n<modelURL>http://www.cisco.com</modelURL>\n<serialNumber></serialNumber>\n<UDN>uuid:ebf5a0a0-1dd1-11b2-a92f-603d266f9915</UDN>\n<UPC>CGA4131COM</UPC>\n<serviceList>\n<service>\n<serviceType>urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1</serviceType>\n<serviceId>urn:upnp-org:serviceId:WANCommonIFC1</serviceId>\n<SCPDURL>/WANCommonInterfaceConfigSCPD.xml</SCPDURL>\n<controlURL>/upnp/control/WANCommonInterfaceConfig0</controlURL>\n<eventSubURL>/upnp/event/WANCommonInterfaceConfig0</eventSubURL>\n</service>\n</serviceList>\n<deviceList>\n <device>\n <deviceType>urn:schemas-upnp-org:device:WANConnectionDevice:1</deviceType>\n <friendlyName>WANConnectionDevice:1</friendlyName>\n <manufacturer>Cisco</manufacturer>\n <manufacturerURL>http://www.cisco.com/</manufacturerURL>\n <modelDescription>CGA4131COM</modelDescription>\n <modelName>CGA4131COM</modelName>\n <modelNumber>CGA4131COM</modelNumber>\n <modelURL>http://www.cisco.com</modelURL>\n <serialNumber></serialNumber>\n <UDN>uuid:11111111-2222-3333-4444-555555555555</UDN>\n <UPC>CGA4131COM</UPC>\n <serviceList>\n <service>\n <serviceType>urn:schemas-upnp-org:service:WANIPConnection:1</serviceType>\n <serviceId>urn:upnp-org:serviceId:WANIPConn1</serviceId>\n <SCPDURL>/WANIPConnectionServiceSCPD.xml</SCPDURL>\n <controlURL>/upnp/control/WANIPConnection0</controlURL>\n <eventSubURL>/upnp/event/WANIPConnection0</eventSubURL>\n </service>\n </serviceList>\n </device>\n</deviceList>\n</device>\n</deviceList>\n<presentationURL>http://10.1.10.1/</presentationURL></device>\n</root>\n"
expected_parsed = {
'specVersion': {'major': '1', 'minor': '0'},
'device': {
'deviceType': 'urn:schemas-upnp-org:device:InternetGatewayDevice:1',
'friendlyName': 'CGA4131COM',
'manufacturer': 'Cisco',
'manufacturerURL': 'http://www.cisco.com/',
'modelDescription': 'CGA4131COM',
'modelName': 'CGA4131COM',
'modelNumber': 'CGA4131COM',
'modelURL': 'http://www.cisco.com',
'UDN': 'uuid:11111111-2222-3333-4444-555555555556',
'UPC': 'CGA4131COM',
'serviceList': {
'service': {
'serviceType': 'urn:schemas-upnp-org:service:Layer3Forwarding:1',
'serviceId': 'urn:upnp-org:serviceId:L3Forwarding1',
'SCPDURL': '/Layer3ForwardingSCPD.xml',
'controlURL': '/upnp/control/Layer3Forwarding',
'eventSubURL': '/upnp/event/Layer3Forwarding'
}
},
'deviceList': {
'device': {
'deviceType': 'urn:schemas-upnp-org:device:WANDevice:1',
'friendlyName': 'WANDevice:1',
'manufacturer': 'Cisco',
'manufacturerURL': 'http://www.cisco.com/',
'modelDescription': 'CGA4131COM',
'modelName': 'CGA4131COM',
'modelNumber': 'CGA4131COM',
'modelURL': 'http://www.cisco.com',
'UDN': 'uuid:ebf5a0a0-1dd1-11b2-a92f-603d266f9915',
'UPC': 'CGA4131COM',
'serviceList': {
'service': {
'serviceType': 'urn:schemas-upnp-org:service:WANCommonInterfaceConfig:1',
'serviceId': 'urn:upnp-org:serviceId:WANCommonIFC1',
'SCPDURL': '/WANCommonInterfaceConfigSCPD.xml',
'controlURL': '/upnp/control/WANCommonInterfaceConfig0',
'eventSubURL': '/upnp/event/WANCommonInterfaceConfig0'
}
},
'deviceList': {
'device': {
'deviceType': 'urn:schemas-upnp-org:device:WANConnectionDevice:1',
'friendlyName': 'WANConnectionDevice:1',
'manufacturer': 'Cisco',
'manufacturerURL': 'http://www.cisco.com/',
'modelDescription': 'CGA4131COM',
'modelName': 'CGA4131COM',
'modelNumber': 'CGA4131COM',
'modelURL': 'http://www.cisco.com',
'UDN': 'uuid:11111111-2222-3333-4444-555555555555',
'UPC': 'CGA4131COM',
'serviceList': {
'service': {
'serviceType': 'urn:schemas-upnp-org:service:WANIPConnection:1',
'serviceId': 'urn:upnp-org:serviceId:WANIPConn1',
'SCPDURL': '/WANIPConnectionServiceSCPD.xml',
'controlURL': '/upnp/control/WANIPConnection0',
'eventSubURL': '/upnp/event/WANIPConnection0'
}
}
}
}
}
},
'presentationURL': 'http://10.1.10.1/'
}
}
self.assertDictEqual(expected_parsed, deserialize_scpd_get_response(xml_bytes))

View file

@ -28,6 +28,26 @@ class TestSOAPSerialization(unittest.TestCase):
b"\r\n" \ b"\r\n" \
b"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\"><s:Body>\n<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\">\r\n<NewExternalIPAddress>11.22.33.44</NewExternalIPAddress>\r\n</u:GetExternalIPAddressResponse>\r\n</s:Body> </s:Envelope>" b"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\"><s:Body>\n<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\">\r\n<NewExternalIPAddress>11.22.33.44</NewExternalIPAddress>\r\n</u:GetExternalIPAddressResponse>\r\n</s:Body> </s:Envelope>"
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"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\"><s:Body>\n</s:Body> </s:Envelope>"
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"<s:Envelope xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\" s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\"><s:Body>\n<u:GetExternalIPAddressResponse xmlns:u=\"urn:schemas-upnp-org:service:WANIPConnection:1\"></u:GetExternalIPAddressResponse>\r\n</s:Body> </s:Envelope>"
error_response = b"HTTP/1.1 500 Internal Server Error\r\n" \ error_response = b"HTTP/1.1 500 Internal Server Error\r\n" \
b"Server: WebServer\r\n" \ b"Server: WebServer\r\n" \
b"Date: Thu, 11 Oct 2018 22:16:17 GMT\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.method, self.param_names, self.st, self.gateway_address, self.path, **self.kwargs
), self.post_bytes) ), 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): def test_deserialize_post_response(self):
self.assertDictEqual( self.assertDictEqual(
deserialize_soap_post_response(self.post_response, self.method, service_id=self.st.decode()), deserialize_soap_post_response(self.post_response, self.method, service_id=self.st.decode()),
{'NewExternalIPAddress': '11.22.33.44'} {'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): def test_raise_from_error_response(self):
raised = False raised = False
try: try:

View file

@ -1,7 +1,6 @@
import contextlib import contextlib
from io import StringIO from io import StringIO
from tests import TestBase from tests import AsyncioTestCase, mock_tcp_and_udp
from tests.mocks import mock_tcp_and_udp
from collections import OrderedDict from collections import OrderedDict
from aioupnp.serialization.ssdp import SSDPDatagram from aioupnp.serialization.ssdp import SSDPDatagram
from aioupnp.protocols.m_search_patterns import packet_generator from aioupnp.protocols.m_search_patterns import packet_generator
@ -22,7 +21,7 @@ m_search_cli_result = """{
}\n""" }\n"""
class TestCLI(TestBase): class TestCLI(AsyncioTestCase):
gateway_address = "10.0.0.1" gateway_address = "10.0.0.1"
soap_port = 49152 soap_port = 49152
m_search_args = OrderedDict([ m_search_args = OrderedDict([
@ -78,7 +77,7 @@ class TestCLI(TestBase):
with contextlib.redirect_stdout(actual_output): 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): with mock_tcp_and_udp(self.loop, '10.0.0.1', tcp_replies=self.scpd_replies, udp_replies=self.udp_replies):
main( 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.loop
) )
self.assertEqual("11.22.33.44\n", actual_output.getvalue()) self.assertEqual("11.22.33.44\n", actual_output.getvalue())
@ -89,7 +88,7 @@ class TestCLI(TestBase):
with contextlib.redirect_stdout(actual_output): 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): with mock_tcp_and_udp(self.loop, '10.0.0.1', tcp_replies=self.scpd_replies, udp_replies=self.udp_replies):
main( 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.loop
) )
self.assertEqual(timeout_msg, actual_output.getvalue()) self.assertEqual(timeout_msg, actual_output.getvalue())
@ -98,7 +97,7 @@ class TestCLI(TestBase):
with contextlib.redirect_stdout(actual_output): 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): with mock_tcp_and_udp(self.loop, '10.0.0.1', tcp_replies=self.scpd_replies, udp_replies=self.udp_replies):
main( 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.loop
) )
self.assertEqual(m_search_cli_result, actual_output.getvalue()) self.assertEqual(m_search_cli_result, actual_output.getvalue())

View file

@ -1,10 +1,7 @@
import asyncio
from aioupnp.fault import UPnPError from aioupnp.fault import UPnPError
from tests import TestBase from tests import AsyncioTestCase, mock_tcp_and_udp
from tests.mocks import mock_tcp_and_udp
from collections import OrderedDict from collections import OrderedDict
from aioupnp.gateway import Gateway from aioupnp.gateway import Gateway, get_action_list
from aioupnp.serialization.ssdp import SSDPDatagram from aioupnp.serialization.ssdp import SSDPDatagram
@ -14,7 +11,137 @@ def gen_get_bytes(location: str, host: str) -> bytes:
).encode() ).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" gateway_address = "10.0.0.1"
client_address = "10.0.0.2" client_address = "10.0.0.2"
soap_port = 49152 soap_port = 49152

55
tests/test_interfaces.py Normal file
View file

@ -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')