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

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,