LBRYCommunity/aioupnp
1
0
Fork 0
Code Issues 7 Pull requests 1 Projects Releases 1 Packages Wiki Activity

mypy

Browse source
This commit is contained in:
Jack Robison 2020-10-30 15:12:39 -04:00
parent 903041b5f6
commit 52ac73bf57
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
4 changed files with 12 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
aioupnp/protocols/ssdp.py
View file

@ -171,8 +171,8 @@ async def multi_m_search(lan_address: str, gateway_address: str, timeout: int =
protocol, gateway_address, lan_address = await listen_ssdp( protocol, gateway_address, lan_address = await listen_ssdp(
lan_address, gateway_address, loop lan_address, gateway_address, loop
) )
fut = asyncio.ensure_future(protocol.send_m_searches( fut = protocol.send_m_searches(
address=gateway_address, datagrams=list(packet_generator()) address=gateway_address, datagrams=list(packet_generator())
), loop=loop) )
loop.call_later(timeout, lambda: None if not fut or fut.done() else fut.cancel()) loop.call_later(timeout, lambda: None if not fut or fut.done() else fut.cancel())
return protocol return protocol

2
aioupnp/serialization/scpd.py
View file

@ -39,7 +39,7 @@ def serialize_scpd_get(path: str, address: str) -> bytes:
def deserialize_scpd_get_response(content: bytes) -> Dict[str, Any]: def deserialize_scpd_get_response(content: bytes) -> Dict[str, Any]:
if XML_VERSION_PREFIX.encode() in content: if XML_VERSION_PREFIX.encode() in content:
parsed: List[Tuple[bytes, bytes]] = CONTENT_PATTERN.findall(content.decode()) parsed: List[Tuple[str, str]] = CONTENT_PATTERN.findall(content.decode())
xml_dict = xml_to_dict('' if not parsed else parsed[0][0]) xml_dict = xml_to_dict('' if not parsed else parsed[0][0])
return parse_device_dict(xml_dict) return parse_device_dict(xml_dict)
return {} return {}

7
aioupnp/upnp.py
View file

@ -260,15 +260,18 @@ class UPnP:
await self.get_redirects() await self.get_redirects()
except UPnPError: except UPnPError:
pass pass
external_port = 0
made_mapping = False
try: try:
external_port = await self.get_next_mapping(1234, 'TCP', 'aioupnp testing') external_port = await self.get_next_mapping(1234, 'TCP', 'aioupnp testing')
made_mapping = True
except UPnPError: except UPnPError:
external_port = None pass
try: try:
await self.get_redirects() await self.get_redirects()
except UPnPError: except UPnPError:
pass pass
if external_port: if made_mapping:
try: try:
await self.delete_port_mapping(external_port, 'TCP') await self.delete_port_mapping(external_port, 'TCP')
except UPnPError: except UPnPError:

8
tests/test_case_insensitive.py
View file

@ -2,7 +2,7 @@ import unittest
from aioupnp.device import CaseInsensitive from aioupnp.device import CaseInsensitive
class TestService(CaseInsensitive): class _TestService(CaseInsensitive):
serviceType = None serviceType = None
serviceId = None serviceId = None
controlURL = None controlURL = None
@ -12,14 +12,14 @@ class TestService(CaseInsensitive):
class TestCaseInsensitive(unittest.TestCase): class TestCaseInsensitive(unittest.TestCase):
def test_initialize(self): def test_initialize(self):
s = TestService( s = _TestService(
serviceType="test", serviceId="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3" serviceType="test", serviceId="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3"
) )
self.assertEqual('test', getattr(s, 'serviceType')) self.assertEqual('test', getattr(s, 'serviceType'))
self.assertEqual('test', getattr(s, 'servicetype')) self.assertEqual('test', getattr(s, 'servicetype'))
self.assertEqual('test', getattr(s, 'SERVICETYPE')) self.assertEqual('test', getattr(s, 'SERVICETYPE'))
s = TestService( s = _TestService(
servicetype="test", serviceid="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3" servicetype="test", serviceid="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3"
) )
self.assertEqual('test', getattr(s, 'serviceType')) self.assertEqual('test', getattr(s, 'serviceType'))
@ -35,7 +35,7 @@ class TestCaseInsensitive(unittest.TestCase):
}, s.as_dict()) }, s.as_dict())
def test_set_attr(self): def test_set_attr(self):
s = TestService( s = _TestService(
serviceType="test", serviceId="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3" serviceType="test", serviceId="test id", controlURL="/test", eventSubURL="/test2", SCPDURL="/test3"
) )
self.assertEqual('test', getattr(s, 'serviceType')) self.assertEqual('test', getattr(s, 'serviceType'))