diff --git a/mygeotab/api.py b/mygeotab/api.py index 01a9f45..8f7ee49 100644 --- a/mygeotab/api.py +++ b/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 ) diff --git a/mygeotab/cli.py b/mygeotab/cli.py index e34194b..b0806a1 100644 --- a/mygeotab/cli.py +++ b/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 diff --git a/mygeotab/py3/api_async.py b/mygeotab/py3/api_async.py index 7bf80ea..ae406f1 100644 --- a/mygeotab/py3/api_async.py +++ b/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() diff --git a/tests/test_api_async.py b/tests/test_api_async.py index 03d9a8b..357d7b1 100644 --- a/tests/test_api_async.py +++ b/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,25 +142,25 @@ 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: @@ -161,7 +168,7 @@ class TestAsyncServerCallApi: 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) diff --git a/tests/test_api_call.py b/tests/test_api_call.py index 13f840d..e99b71e 100644 --- a/tests/test_api_call.py +++ b/tests/test_api_call.py @@ -1,6 +1,8 @@ # -*- coding: utf-8 -*- import os +import random +import string import pytest @@ -14,7 +16,7 @@ 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" @@ -22,6 +24,19 @@ 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: