diff --git a/aioupnp/commands.py b/aioupnp/commands.py index 163ad44..34e4dc2 100644 --- a/aioupnp/commands.py +++ b/aioupnp/commands.py @@ -35,6 +35,16 @@ class GetGenericPortMappingEntryResponse(typing.NamedTuple): lease_time: int +class SCPDRequestDebuggingInfo(typing.NamedTuple): + method: str + kwargs: typing.Dict[str, typing.Union[str, int, bool]] + response_xml: bytes + result: typing.Optional[typing.Union[str, int, bool, GetSpecificPortMappingEntryResponse, + GetGenericPortMappingEntryResponse]] + err: typing.Optional[Exception] + ts: float + + def recast_return(return_annotation, result: typing.Dict[str, typing.Union[int, str]], result_keys: typing.List[str]) -> typing.Optional[ typing.Union[str, int, bool, GetSpecificPortMappingEntryResponse, GetGenericPortMappingEntryResponse]]: @@ -108,11 +118,7 @@ class SOAPCommands: self._base_address = base_address self._port = port - self._requests: typing.List[typing.Tuple[str, typing.Dict[str, typing.Any], bytes, - typing.Optional[typing.Union[str, int, bool, - GetSpecificPortMappingEntryResponse, - GetGenericPortMappingEntryResponse]], - typing.Optional[Exception], float]] = [] + self._request_debug_infos: typing.List[SCPDRequestDebuggingInfo] = [] def is_registered(self, name: str) -> bool: if name not in self.SOAP_COMMANDS: @@ -147,11 +153,17 @@ class SOAPCommands: ) if err is not None: assert isinstance(xml_bytes, bytes) - self._requests.append((name, kwargs, xml_bytes, None, err, time.time())) + self._request_debug_infos.append(SCPDRequestDebuggingInfo(name, kwargs, xml_bytes, None, err, time.time())) raise err assert 'return' in annotations - result = recast_return(annotations['return'], response, output_names) - self._requests.append((name, kwargs, xml_bytes, result, None, time.time())) + try: + result = recast_return(annotations['return'], response, output_names) + self._request_debug_infos.append(SCPDRequestDebuggingInfo(name, kwargs, xml_bytes, result, None, time.time())) + except Exception as err: + if isinstance(err, asyncio.CancelledError): + raise + self._request_debug_infos.append(SCPDRequestDebuggingInfo(name, kwargs, xml_bytes, None, err, time.time())) + raise UPnPError(f"Raised {str(type(err).__name__)}({str(err)}) parsing response for {name}") return result if not len(list(k for k in annotations if k != 'return')): diff --git a/aioupnp/gateway.py b/aioupnp/gateway.py index cd2a68c..26eb64b 100644 --- a/aioupnp/gateway.py +++ b/aioupnp/gateway.py @@ -6,7 +6,7 @@ from collections import OrderedDict from typing import Dict, List from aioupnp.util import get_dict_val_case_insensitive from aioupnp.constants import SPEC_VERSION, SERVICE -from aioupnp.commands import SOAPCommands +from aioupnp.commands import SOAPCommands, SCPDRequestDebuggingInfo from aioupnp.device import Device, Service from aioupnp.protocols.ssdp import fuzzy_m_search, m_search from aioupnp.protocols.scpd import scpd_get @@ -103,17 +103,6 @@ class Gateway: self._registered_commands: Dict[str, str] = {} self.commands = SOAPCommands(self._loop, self.base_ip, self.port) - # def gateway_descriptor(self) -> dict: - # r = { - # 'server': self.server.decode(), - # 'urlBase': self.url_base, - # 'location': self.location.decode(), - # "specVersion": self.spec_version, - # 'usn': self.usn.decode(), - # 'urn': self.urn.decode(), - # } - # return r - @property def manufacturer_string(self) -> str: manufacturer_string = "UNKNOWN GATEWAY" @@ -147,37 +136,26 @@ class Gateway: # return service # return None - # @property - # def soap_requests(self) -> typing.List[typing.Tuple[str, typing.Dict[str, typing.Any], bytes, - # typing.Optional[typing.Tuple], - # typing.Optional[Exception], float]]: - # soap_call_infos: typing.List[typing.Tuple[str, typing.Dict[str, typing.Any], bytes, - # typing.Optional[typing.Tuple], - # typing.Optional[Exception], float]] = [] - # soap_call_infos.extend([ - # (name, request_args, raw_response, decoded_response, soap_error, ts) - # for ( - # name, request_args, raw_response, decoded_response, soap_error, ts - # ) in self.commands._requests - # ]) - # soap_call_infos.sort(key=lambda x: x[5]) - # return soap_call_infos - - # def debug_gateway(self) -> Dict[str, Union[str, bytes, int, Dict, List]]: - # return { - # 'manufacturer_string': self.manufacturer_string, - # 'gateway_address': self.base_ip, - # 'gateway_descriptor': self.gateway_descriptor(), - # 'gateway_xml': self._xml_response, - # 'services_xml': self._service_descriptors, - # 'services': {service.SCPDURL: service.as_dict() for service in self._services}, - # 'm_search_args': [(k, v) for (k, v) in self._m_search_args.items()], - # 'reply': self._ok_packet.as_dict(), - # 'soap_port': self.port, - # 'registered_soap_commands': self._registered_commands, - # 'unsupported_soap_commands': self._unsupported_actions, - # 'soap_requests': self.soap_requests - # } + def debug_gateway(self) -> Dict[str, typing.Union[str, bytes, int, Dict, List]]: + return { + 'manufacturer_string': self.manufacturer_string, + 'gateway_address': self.base_ip, + 'server': self.server.decode(), + 'urlBase': self.url_base or '', + 'location': self.location.decode(), + "specVersion": self.spec_version or '', + 'usn': self.usn.decode(), + 'urn': self.urn.decode(), + 'gateway_xml': self._xml_response, + 'services_xml': self._service_descriptors, + 'services': {service.SCPDURL: service.as_dict() for service in self._services}, + 'm_search_args': [(k, v) for (k, v) in self._m_search_args.items()], + 'reply': self._ok_packet.as_dict(), + 'soap_port': self.port, + 'registered_soap_commands': self._registered_commands, + 'unsupported_soap_commands': self._unsupported_actions, + 'soap_requests': list(self.commands._request_debug_infos) + } @classmethod async def _discover_gateway(cls, lan_address: str, gateway_address: str, timeout: int = 30,