Handle DBUnavailableException as an authentication issue

FIx version tests
Use ZoneType insead of Trailer for entity tests
Formatted with black

mygeotab/api.py

@@ -40,7 +40,7 @@ def __init__(
         server="my.geotab.com",
         timeout=DEFAULT_TIMEOUT,
         proxies=None,
-        cert=None
+        cert=None,
     ):
         """Initialize the MyGeotab API object with credentials.
 
@@ -111,17 +111,20 @@ def call(self, method, **parameters):
         try:
             result = _query(
                 self._server,
-                method, params,
+                method,
+                params,
                 self.timeout,
                 verify_ssl=self._is_verify_ssl,
                 proxies=self._proxies,
-                cert=self._cert
+                cert=self._cert,
             )
             if result is not None:
                 self.__reauthorize_count = 0
             return result
         except MyGeotabException as exception:
-            if exception.name == "InvalidUserException":
+            if exception.name == "InvalidUserException" or (
+                exception.name == "DbUnavailableException" and "Initializing" in exception.message
+            ):
                 if self.__reauthorize_count == 0 and self.credentials.password:
                     self.__reauthorize_count += 1
                     self.authenticate()
@@ -229,7 +232,7 @@ def authenticate(self):
                 self.timeout,
                 verify_ssl=self._is_verify_ssl,
                 proxies=self._proxies,
-                cert=self._cert
+                cert=self._cert,
             )
             if result:
                 if "path" not in result and self.credentials.session_id:
@@ -245,7 +248,9 @@ def authenticate(self):
                 )
                 return self.credentials
         except MyGeotabException as exception:
-            if exception.name == "InvalidUserException":
+            if exception.name == "InvalidUserException" or (
+                exception.name == "DbUnavailableException" and "Initializing" in exception.message
+            ):
                 raise AuthenticationException(
                     self.credentials.username, self.credentials.database, self.credentials.server
                 )

mygeotab/cli.py

@@ -194,7 +194,7 @@ def remove(session, database):
 def console(session, database=None, user=None, password=None, server=None):
     """An interactive Python API console for MyGeotab
 
-    If either IPython or ptpython are installed, it will launch an interactive console using those libraries instead of 
+    If either IPython or ptpython are installed, it will launch an interactive console using those libraries instead of
     the built-in Python console. Using IPython or ptpython has numerous advantages over the stock Python console,
     including: colors, pretty printing, command auto-completion, and more.
 
@@ -217,8 +217,12 @@ def console(session, database=None, user=None, password=None, server=None):
         def configure(repl):
             repl.prompt_style = "ipython"
 
-        embed(globals=globals(), locals=local_vars, title="{0}. {1}".format(myg_console_version, auth_line),
-              configure=configure)
+        embed(
+            globals=globals(),
+            locals=local_vars,
+            title="{0}. {1}".format(myg_console_version, auth_line),
+            configure=configure,
+        )
     except ImportError:
         try:
             from IPython import embed

mygeotab/py3/api_async.py

@@ -35,7 +35,7 @@ def __init__(
         server="my.geotab.com",
         timeout=DEFAULT_TIMEOUT,
         proxies=None,
-        cert=None
+        cert=None,
     ):
         """
         Initialize the asynchronous MyGeotab API object with credentials.
@@ -75,7 +75,9 @@ async def call_async(self, method, **parameters):
                 self.__reauthorize_count = 0
             return result
         except MyGeotabException as exception:
-            if exception.name == "InvalidUserException":
+            if exception.name == "InvalidUserException" or (
+                exception.name == "DbUnavailableException" and "Initializing" in exception.message
+            ):
                 if self.__reauthorize_count == 0 and self.credentials.password:
                     self.__reauthorize_count += 1
                     self.authenticate()

tests/test_api_async.py

@@ -7,10 +7,20 @@
 import sys
 
 from mygeotab import API, server_call_async
-from mygeotab.exceptions import MyGeotabException, TimeoutException
-from tests.test_api_call import SERVER, USERNAME, PASSWORD, DATABASE, CER_FILE, KEY_FILE, PEM_FILE, TRAILER_NAME
-
-ASYNC_TRAILER_NAME = "async {name}".format(name=TRAILER_NAME)
+from mygeotab.exceptions import AuthenticationException, MyGeotabException, TimeoutException
+from tests.test_api_call import (
+    SERVER,
+    USERNAME,
+    PASSWORD,
+    DATABASE,
+    CER_FILE,
+    KEY_FILE,
+    PEM_FILE,
+    ZONETYPE_NAME,
+    generate_fake_credentials,
+)
+
+ASYNC_ZONETYPE_NAME = "async {name}".format(name=ZONETYPE_NAME)
 
 USERNAME = os.environ.get("MYGEOTAB_USERNAME_ASYNC", USERNAME)
 PASSWORD = os.environ.get("MYGEOTAB_PASSWORD_ASYNC", PASSWORD)
@@ -43,17 +53,14 @@ def async_populated_api():
 
 @pytest.fixture(scope="session")
 def async_populated_api_entity(async_populated_api):
-    def clean_trailers():
-        try:
-            trailers = async_populated_api.get("Trailer", name=ASYNC_TRAILER_NAME)
-            for trailer in trailers:
-                async_populated_api.remove("Trailer", trailer)
-        except Exception:
-            pass
+    def clean_zonetypes():
+        zonetypes = async_populated_api.get("ZoneType", name=ASYNC_ZONETYPE_NAME)
+        for zonetype in zonetypes:
+            async_populated_api.remove("ZoneType", zonetype)
 
-    clean_trailers()
+    clean_zonetypes()
     yield async_populated_api
-    clean_trailers()
+    clean_zonetypes()
 
 
 class TestAsyncCallApi:
@@ -135,33 +142,33 @@ async def test_get_search_parameter(self, async_populated_api):
 
     @pytest.mark.asyncio
     async def test_add_edit_remove(self, async_populated_api_entity):
-        async def get_trailer():
-            trailers = await async_populated_api_entity.get_async("Trailer", name=ASYNC_TRAILER_NAME)
-            assert len(trailers) == 1
-            return trailers[0]
+        async def get_zonetypes():
+            zonetypes = await async_populated_api_entity.get_async("ZoneType", name=ASYNC_ZONETYPE_NAME)
+            assert len(zonetypes) == 1
+            return zonetypes[0]
 
         user = async_populated_api_entity.get("User", name=USERNAME)[0]
-        trailer = {"name": ASYNC_TRAILER_NAME, "groups": user["companyGroups"]}
-        trailer_id = await async_populated_api_entity.add_async("Trailer", trailer)
-        trailer["id"] = trailer_id
-        trailer = await get_trailer()
-        assert trailer["name"] == ASYNC_TRAILER_NAME
+        zonetype = {"name": ASYNC_ZONETYPE_NAME, "groups": user["companyGroups"]}
+        zonetype_id = await async_populated_api_entity.add_async("ZoneType", zonetype)
+        zonetype["id"] = zonetype_id
+        zonetype = await get_zonetypes()
+        assert zonetype["name"] == ASYNC_ZONETYPE_NAME
         comment = "some comment"
-        trailer["comment"] = comment
-        await async_populated_api_entity.set_async("Trailer", trailer)
-        trailer = await get_trailer()
-        assert trailer["comment"] == comment
-        await async_populated_api_entity.remove_async("Trailer", trailer)
-        trailers = await async_populated_api_entity.get_async("Trailer", name=ASYNC_TRAILER_NAME)
-        assert len(trailers) == 0
+        zonetype["comment"] = comment
+        await async_populated_api_entity.set_async("ZoneType", zonetype)
+        zonetype = await get_zonetypes()
+        assert zonetype["comment"] == comment
+        await async_populated_api_entity.remove_async("ZoneType", zonetype)
+        zonetypes = await async_populated_api_entity.get_async("ZoneType", name=ASYNC_ZONETYPE_NAME)
+        assert len(zonetypes) == 0
 
 
 class TestAsyncServerCallApi:
     @pytest.mark.asyncio
     async def test_get_version(self):
         version = await server_call_async("GetVersion", server="my3.geotab.com")
         version_split = version.split(".")
-        assert len(version_split) == 4
+        assert len(version_split) == 3
 
     @pytest.mark.asyncio
     async def test_invalid_server_call(self):
@@ -177,3 +184,21 @@ async def test_timeout(self):
         with pytest.raises(TimeoutException) as excinfo:
             await server_call_async("GetVersion", server="my36.geotab.com", timeout=0.01)
         assert "Request timed out @ my36.geotab.com" in str(excinfo.value)
+
+
+class TestAsyncAuthentication:
+    @pytest.mark.asyncio
+    async def test_invalid_session(self):
+        fake_credentials = generate_fake_credentials()
+        test_api = API(
+            fake_credentials["username"],
+            session_id=fake_credentials["sessionid"],
+            database=fake_credentials["database"],
+        )
+        assert fake_credentials["username"] in str(test_api.credentials)
+        assert fake_credentials["database"] in str(test_api.credentials)
+        with pytest.raises(AuthenticationException) as excinfo:
+            await test_api.get_async("User")
+        assert "Cannot authenticate" in str(excinfo.value)
+        assert fake_credentials["database"] in str(excinfo.value)
+        assert fake_credentials["username"] in str(excinfo.value)

tests/test_api_call.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 
 import os
+import random
+import string
 
 import pytest
 
@@ -14,14 +16,27 @@
 CER_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_CER")
 KEY_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_KEY")
 PEM_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_PEM")
-TRAILER_NAME = "mygeotab-python test trailer"
+ZONETYPE_NAME = "mygeotab-python test zonetype"
 
 FAKE_USERNAME = "fakeusername"
 FAKE_PASSWORD = "fakepassword"
 FAKE_DATABASE = "fakedatabase"
 FAKE_SESSIONID = "3n8943bsdf768"
 
 
+def get_random_str(str_length):
+    return "".join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(str_length))
+
+
+def generate_fake_credentials():
+    return dict(
+        username=FAKE_USERNAME + get_random_str(20),
+        password=FAKE_PASSWORD + get_random_str(20),
+        database=FAKE_DATABASE + get_random_str(20),
+        sessionid=FAKE_SESSIONID + get_random_str(30),
+    )
+
+
 @pytest.fixture(scope="session")
 def populated_api():
     cert = None
@@ -47,17 +62,14 @@ def populated_api():
 
 @pytest.fixture(scope="session")
 def populated_api_entity(populated_api):
-    def clean_trailers():
-        try:
-            trailers = populated_api.get("Trailer", name=TRAILER_NAME)
-            for trailer in trailers:
-                populated_api.remove("Trailer", trailer)
-        except Exception:
-            pass
+    def clean_zonetypes():
+        zonetypes = populated_api.get("ZoneType", name=ZONETYPE_NAME)
+        for zonetype in zonetypes:
+            populated_api.remove("ZoneType", zonetype)
 
-    clean_trailers()
+    clean_zonetypes()
     yield populated_api
-    clean_trailers()
+    clean_zonetypes()
 
 
 class TestCallApi:
@@ -135,43 +147,48 @@ def test_get_search_parameter(self, populated_api):
 
 class TestEntity:
     def test_add_edit_remove(self, populated_api_entity):
-        def get_trailer():
-            trailers = populated_api_entity.get("Trailer", name=TRAILER_NAME)
-            assert len(trailers) == 1
-            return trailers[0]
-
-        user = populated_api_entity.get("User", name=USERNAME)[0]
-        trailer = {"name": TRAILER_NAME, "groups": user["companyGroups"]}
-        trailer["id"] = populated_api_entity.add("Trailer", trailer)
-        assert trailer["id"] is not None
-        trailer = get_trailer()
-        assert trailer["name"] == TRAILER_NAME
+        def get_zonetype():
+            zonetypes = populated_api_entity.get("ZoneType", name=ZONETYPE_NAME)
+            assert len(zonetypes) == 1
+            return zonetypes[0]
+
+        zonetype = {"name": ZONETYPE_NAME}
+        zonetype["id"] = populated_api_entity.add("ZoneType", zonetype)
+        assert zonetype["id"] is not None
+        zonetype = get_zonetype()
+        assert zonetype["name"] == ZONETYPE_NAME
         comment = "some comment"
-        trailer["comment"] = comment
-        populated_api_entity.set("Trailer", trailer)
-        trailer = get_trailer()
-        assert trailer["comment"] == comment
-        populated_api_entity.remove("Trailer", trailer)
-        trailers = populated_api_entity.get("Trailer", name=TRAILER_NAME)
-        assert len(trailers) == 0
+        zonetype["comment"] = comment
+        populated_api_entity.set("ZoneType", zonetype)
+        zonetype = get_zonetype()
+        assert zonetype["comment"] == comment
+        populated_api_entity.remove("ZoneType", zonetype)
+        zonetypes = populated_api_entity.get("ZoneType", name=ZONETYPE_NAME)
+        assert len(zonetypes) == 0
 
 
 class TestAuthentication:
     def test_invalid_session(self):
-        test_api = api.API(FAKE_USERNAME, session_id=FAKE_SESSIONID, database=FAKE_DATABASE)
-        assert FAKE_USERNAME in str(test_api.credentials)
-        assert FAKE_DATABASE in str(test_api.credentials)
+        fake_credentials = generate_fake_credentials()
+        test_api = api.API(
+            fake_credentials["username"],
+            session_id=fake_credentials["sessionid"],
+            database=fake_credentials["database"],
+        )
+        assert fake_credentials["username"] in str(test_api.credentials)
+        assert fake_credentials["database"] in str(test_api.credentials)
         with pytest.raises(AuthenticationException) as excinfo:
             test_api.get("User")
         assert "Cannot authenticate" in str(excinfo.value)
-        assert FAKE_DATABASE in str(excinfo.value)
-        assert FAKE_USERNAME in str(excinfo.value)
+        assert fake_credentials["database"] in str(excinfo.value)
+        assert fake_credentials["username"] in str(excinfo.value)
 
     def test_username_password_exists(self):
+        fake_credentials = generate_fake_credentials()
         with pytest.raises(Exception) as excinfo1:
             api.API(None)
         with pytest.raises(Exception) as excinfo2:
-            api.API(FAKE_USERNAME)
+            api.API(fake_credentials["username"])
         assert "username" in str(excinfo1.value)
         assert "password" in str(excinfo2.value)
 
@@ -182,18 +199,23 @@ def test_call_authenticate_sessionid(self, populated_api):
         assert credentials.session_id is not None
 
     def test_call_authenticate_invalid_sessionid(self):
-        test_api = api.API(FAKE_USERNAME, session_id=FAKE_SESSIONID, database=FAKE_DATABASE)
+        fake_credentials = generate_fake_credentials()
+        test_api = api.API(
+            fake_credentials["username"],
+            session_id=fake_credentials["sessionid"],
+            database=fake_credentials["database"],
+        )
         with pytest.raises(AuthenticationException) as excinfo:
             test_api.authenticate()
         assert "Cannot authenticate" in str(excinfo.value)
-        assert FAKE_DATABASE in str(excinfo.value)
+        assert fake_credentials["database"] in str(excinfo.value)
 
 
 class TestServerCallApi:
     def test_get_version(self):
         version = api.server_call("GetVersion", server="my3.geotab.com")
         version_split = version.split(".")
-        assert len(version_split) == 4
+        assert len(version_split) == 3
 
     def test_invalid_server_call(self):
         with pytest.raises(Exception) as excinfo1: