debug xml

txupnp/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.0.1a2"
+__version__ = "0.0.1a3"
 __name__ = "txupnp"
 __author__ = "Jack Robison"
 __maintainer__ = "Jack Robison"

txupnp/cli.py

@@ -7,28 +7,50 @@ from txupnp.upnp import UPnP
 log = logging.getLogger("txupnp")
 
 
+def debug_device(u, include_gateway_xml=False, *_):
+    print(u.get_debug_info(include_gateway_xml=include_gateway_xml))
+    return defer.succeed(None)
+
+
 @defer.inlineCallbacks
-def run_command(found, u, command):
-    if not found:
-        print("failed to find gateway")
-        reactor.callLater(0, reactor.stop)
-        return
-    if command == "debug_device":
-        print(u.get_debug_info())
-    elif command == "get_external_ip":
+def get_external_ip(u, *_):
     ip = yield u.get_external_ip()
     print(ip)
-    elif command == "list_mappings":
+
+
+@defer.inlineCallbacks
+def list_mappings(u, *_):
+    print(u.get_debug_info(include_gateway_xml=True))
     redirects = yield u.get_redirects()
     print("found {} redirects".format(len(redirects)))
     for redirect in redirects:
         print("\t", redirect)
 
 
+cli_commands = {
+    "debug_device": debug_device,
+    "get_external_ip": get_external_ip,
+    "list_mappings": list_mappings
+}
+
+
+@defer.inlineCallbacks
+def run_command(found, u, command, debug_xml):
+    if not found:
+        print("failed to find gateway")
+        reactor.callLater(0, reactor.stop)
+        return
+    if command not in cli_commands:
+        print("unrecognized command: valid commands: %s" % list(cli_commands.keys()))
+    else:
+        yield cli_commands[command](u, debug_xml)
+
+
 def main():
     parser = argparse.ArgumentParser(description="upnp command line utility")
-    parser.add_argument(dest="command", type=str, help="debug_gateway | list_mappings")
+    parser.add_argument(dest="command", type=str, help="debug_gateway | list_mappings | get_external_ip")
     parser.add_argument("--debug_logging", dest="debug_logging", default=False, action="store_true")
+    parser.add_argument("--include_igd_xml", dest="include_igd_xml", default=False, action="store_true")
     args = parser.parse_args()
     if args.debug_logging:
         from twisted.python import log as tx_log
@@ -36,15 +58,13 @@ def main():
         observer.start()
         log.setLevel(logging.DEBUG)
     command = args.command
-    if command not in ['debug_device', 'list_mappings', 'get_external_ip']:
-        return sys.exit(0)
 
     def show(err):
         print("error: {}".format(err))
 
     u = UPnP(reactor)
     d = u.discover()
-    d.addCallback(run_command, u, command)
+    d.addCallback(run_command, u, command, args.include_igd_xml)
     d.addErrback(show)
     d.addBoth(lambda _: reactor.callLater(0, reactor.stop))
     reactor.run()

txupnp/soap.py

@@ -52,11 +52,11 @@ class SOAPServiceManager(object):
     def get_available_runners(self):
         return self._command_runners.keys()
 
-    def debug(self):
+    def debug(self, include_gateway_xml=False):
         results = []
         for runner in self._command_runners.values():
             gateway = runner._gateway
-            info = gateway.debug_device()
+            info = gateway.debug_device(include_xml=include_gateway_xml)
             commands = runner.debug_commands()
             service_result = []
             for service in info['services']:

txupnp/upnp.py

@@ -162,13 +162,13 @@ class UPnP(object):
         )
         defer.returnValue(port)
 
-    def get_debug_info(self):
+    def get_debug_info(self, include_gateway_xml=False):
         def default_byte(x):
             if isinstance(x, bytes):
                 return x.decode()
             return x
         return json.dumps({
-            'txupnp': self.soap_manager.debug(),
+            'txupnp': self.soap_manager.debug(include_gateway_xml=include_gateway_xml),
             'miniupnpc_igd_url': self._miniupnpc_igd_url
             },
             indent=2, default=default_byte