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 2019-10-25 14:15:24 -04:00
parent 9570a843e3
commit 1b25c009ca
No known key found for this signature in database
GPG key ID: DF25C68FE0239BB2
3 changed files with 8 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
aioupnp/gateway.py
View file

@ -184,8 +184,9 @@ class Gateway:
return None
@classmethod
async def _gateway_from_igd_args(cls, lan_address: str, gateway_address: str, timeout: int = 30,
igd_args: Optional[typing.Dict[str, typing.Union[int, str]]] = None,
async def _gateway_from_igd_args(cls, lan_address: str, gateway_address: str,
igd_args: typing.Dict[str, typing.Union[int, str]],
timeout: int = 30,
loop: Optional[asyncio.AbstractEventLoop] = None) -> 'Gateway':
datagram = await m_search(lan_address, gateway_address, igd_args, timeout, loop, set())
gateway = await cls._try_gateway_from_ssdp(datagram, lan_address, gateway_address, loop)
@ -208,7 +209,7 @@ class Gateway:
gateway = await cls._try_gateway_from_ssdp(datagram, lan_address, gateway_address, loop)
if gateway:
return gateway
else:
elif datagram.location:
ignored.add(datagram.location)
finally:
ssdp_proto.disconnect()
@ -219,7 +220,7 @@ class Gateway:
loop: Optional[asyncio.AbstractEventLoop] = None) -> 'Gateway':
loop = loop or asyncio.get_event_loop()
if igd_args:
return await cls._gateway_from_igd_args(lan_address, gateway_address, timeout, igd_args, loop)
return await cls._gateway_from_igd_args(lan_address, gateway_address, igd_args, timeout, loop)
try:
return await asyncio.wait_for(loop.create_task(
cls._discover_gateway(lan_address, gateway_address, timeout, loop)

4
aioupnp/protocols/ssdp.py
View file

@ -80,12 +80,10 @@ class SSDPProtocol(MulticastProtocol):
if not fut.done():
fut.set_exception(UPnPError("SSDP transport not connected"))
return
if fut.done():
return
assert packet.st is not None
self._pending_searches.append(
PendingSearch(address, packet.st, fut)
)
self.transport.sendto(packet.encode().encode(), (SSDP_IP_ADDRESS, SSDP_PORT))
# also send unicast

2
aioupnp/serialization/ssdp.py
View file

@ -167,7 +167,7 @@ class SSDPDatagram:
packet = cls._from_string(datagram.decode())
if packet is None:
raise UPnPError(
"failed to decode datagram: {}".format(binascii.hexlify(datagram))
f"failed to decode datagram: {binascii.hexlify(datagram).decode()}"
)
for attr_name in packet._required_fields[packet._packet_type]:
if getattr(packet, attr_name, None) is None: