bumper/__init__.py

@@ -15,7 +15,6 @@ import socket
 import sys
 
 
-
 def strtobool(strbool):
     if str(strbool).lower() in ["true", "1", "t", "y", "on", "yes"]:
         return True
@@ -114,15 +113,12 @@ xmppserverlog.addHandler(xmpp_rotate)
 
 logging.getLogger("asyncio").setLevel(logging.CRITICAL + 1)  # Ignore this logger
 
-mqtt_listen_address = bumper_listen
 mqtt_listen_port = 8883
-conf1_listen_address = bumper_listen
 conf1_listen_port = 443
-conf2_listen_address = bumper_listen
 conf2_listen_port = 8007
-xmpp_listen_address = bumper_listen 
 xmpp_listen_port = 5223
 
+
 async def start():
 
     try:
@@ -157,19 +153,19 @@ async def start():
 
     bumperlog.info("Starting Bumper")
     global mqtt_server
-    mqtt_server = MQTTServer((mqtt_listen_address, mqtt_listen_port))
+    mqtt_server = MQTTServer((bumper_listen, mqtt_listen_port))
     global mqtt_helperbot
-    mqtt_helperbot = MQTTHelperBot((mqtt_listen_address, mqtt_listen_port))
+    mqtt_helperbot = MQTTHelperBot((bumper_listen, mqtt_listen_port))
     global conf_server
     conf_server = ConfServer(
-        (conf1_listen_address, conf1_listen_port), usessl=True, helperbot=mqtt_helperbot
+        (bumper_listen, conf1_listen_port), usessl=True, helperbot=mqtt_helperbot
     )
     global conf_server_2
     conf_server_2 = ConfServer(
-        (conf2_listen_address, conf2_listen_port), usessl=False, helperbot=mqtt_helperbot
+        (bumper_listen, conf2_listen_port), usessl=False, helperbot=mqtt_helperbot
     )
     global xmpp_server
-    xmpp_server = XMPPServer((xmpp_listen_address, xmpp_listen_port))
+    xmpp_server = XMPPServer((bumper_listen, xmpp_listen_port))
 
     # Start web servers
     conf_server.confserver_app()
@@ -438,6 +434,7 @@ class VacBotDevice(object):
             "xmpp_connection": self.xmpp_connection,
         }
 
+
 class GlobalVacBotDevice(VacBotDevice):  # EcoVacs Home
     UILogicId = ""
     ota = True
@@ -944,11 +941,13 @@ API_ERRORS = {
     ERR_WRONG_PWD_FROMATE: "1009",
 }
 
+
 def create_certs():
     import platform
     import os
     import subprocess
     import sys
+
     path = os.path.dirname(sys.modules[__name__].__file__)
     path = os.path.join(path, "..")
     sys.path.insert(0, path)
@@ -958,9 +957,7 @@ def create_certs():
     os.chdir("certs")
     if str(platform.system()).lower() == "windows":
         # run for win
-        subprocess.run(
-            [os.path.join("..", "create_certs", "create_certs_windows.exe")]
-        )
+        subprocess.run([os.path.join("..", "create_certs", "create_certs_windows.exe")])
     elif str(platform.system()).lower() == "darwin":
         # run on mac
         subprocess.run([os.path.join("..", "create_certs", "create_certs_osx")])
@@ -970,12 +967,13 @@ def create_certs():
             subprocess.run([os.path.join("..", "create_certs", "create_certs_rpi")])
         else:
             # run for linux
-            subprocess.run(
-                [os.path.join("..", "create_certs", "create_certs_linux")]
-            )
+            subprocess.run([os.path.join("..", "create_certs", "create_certs_linux")])
 
     else:
-        logging.log(logging.FATAL, "Can't determine platform. Create certs manually and try again.")
+        logging.log(
+            logging.FATAL,
+            "Can't determine platform. Create certs manually and try again.",
+        )
         return
 
     print("Certificates created")
@@ -989,11 +987,13 @@ def create_certs():
     else:
         os.execv(sys.executable, ["python"] + sys.argv)  # Start again
 
+
 def firstrun_input():
     return input(
         "No certificates found, would you like to create them automatically? (y/n): "
     ).lower()
 
+
 def first_run():
     yes = {"yes", "y", "ye", ""}
     print("")
@@ -1001,14 +1001,20 @@ def first_run():
         create_certs()
 
     else:
-        logging.log(logging.FATAL, "Can't continue without certificates, please create some then try again.")
+        logging.log(
+            logging.FATAL,
+            "Can't continue without certificates, please create some then try again.",
+        )
+
 
 def main(argv=None):
     import argparse
+
     global bumper_debug
     global bumper_listen
     global bumper_announce_ip
-
+    if not argv:
+        argv = sys.argv[1:]  # Set argv to argv[1:] if not passed into main
     try:
 
         if not (
@@ -1024,9 +1030,13 @@ def main(argv=None):
             "--listen", type=str, default=None, help="start serving on address"
         )
         parser.add_argument(
-            "--announce", type=str, default=None, help="announce address to bots on checkin"
+            "--announce",
+            type=str,
+            default=None,
+            help="announce address to bots on checkin",
         )
         parser.add_argument("--debug", action="store_true", help="enable debug logs")
+
         args = parser.parse_args(args=argv)
 
         if args.debug:

tests/test_init.py

@@ -35,13 +35,13 @@ async def test_start_stop():
 
         b = bumper
         b.db = "tests/tmp.db"  # Set db location for testing
-        b.conf1_listen_address = "0.0.0.0"
-        b.conf1_listen_port = 443        
+        b.conf1_listen_address = "127.0.0.1"
+        b.conf1_listen_port = 444        
         asyncio.create_task(b.start())
         await asyncio.sleep(0.1)
         l.check_present(("bumper", "INFO", "Starting Bumper"))
         l.clear()
-        assert b.shutting_down == False
+        
         asyncio.create_task(b.shutdown())
         await asyncio.sleep(0.1)
         l.check_present(