diff --git a/tests/test_api_async.py b/tests/test_api_async.py index d6923d2..56da70e 100644 --- a/tests/test_api_async.py +++ b/tests/test_api_async.py @@ -1,77 +1,78 @@ # -*- coding: utf-8 -*- -import os +from unittest.mock import AsyncMock, patch import pytest from mygeotab import API, server_call_async from mygeotab.exceptions import AuthenticationException, MyGeotabException, TimeoutException from tests.test_api_call import ( - CER_FILE, DATABASE, - KEY_FILE, PASSWORD, - PEM_FILE, SERVER, USERNAME, ZONETYPE_NAME, generate_fake_credentials, + mock_authenticate_response, + mock_user_response, + mock_version_response, + mock_zonetype_response, ) 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) - -@pytest.fixture(scope="session") -def async_populated_api(): - cert = None - if CER_FILE and KEY_FILE: - cert = (CER_FILE, KEY_FILE) - elif PEM_FILE: - cert = PEM_FILE - if USERNAME and PASSWORD: - session = API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER, cert=cert) - try: - session.authenticate() - except (MyGeotabException, AuthenticationException) as exception: - pytest.fail(str(exception)) - return - yield session - else: - pytest.skip( - "Can't make calls to the API without the MYGEOTAB_USERNAME and MYGEOTAB_PASSWORD environment variables being set" - ) +@pytest.fixture +def mock_async_query(): + """Fixture to mock the async _query function.""" + with patch("mygeotab.api_async._query", new_callable=AsyncMock) as mock: + yield mock + + +@pytest.fixture +def mock_sync_query(): + """Fixture to mock the sync _query function for authentication.""" + with patch("mygeotab.api._query") as mock: + yield mock + + +@pytest.fixture +def async_populated_api(mock_sync_query): + """Create an async API instance with mocked credentials.""" + mock_sync_query.return_value = mock_authenticate_response() + session = API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER) + session.authenticate() + mock_sync_query.return_value = None + yield session -@pytest.fixture(scope="session") -def async_populated_api_entity(async_populated_api): - def clean_zonetypes(): - zonetypes = async_populated_api.get("ZoneType", name=ASYNC_ZONETYPE_NAME) - for zonetype in zonetypes: - async_populated_api.remove("ZoneType", zonetype) - clean_zonetypes() +@pytest.fixture +def async_populated_api_entity(async_populated_api, mock_async_query): + """Create an async API instance for entity tests.""" + mock_async_query.return_value = [] yield async_populated_api - clean_zonetypes() + mock_async_query.return_value = [] class TestAsyncCallApi: @pytest.mark.asyncio - async def test_get_version(self, async_populated_api): + async def test_get_version(self, async_populated_api, mock_async_query): + mock_async_query.return_value = mock_version_response() version = await async_populated_api.call_async("GetVersion") version_len = len(version.split(".")) assert 3 <= version_len <= 4 @pytest.mark.asyncio - async def test_get_user(self, async_populated_api): + async def test_get_user(self, async_populated_api, mock_async_query): + mock_async_query.return_value = mock_user_response() user = await async_populated_api.get_async("User", name=USERNAME) assert len(user) == 1 user = user[0] assert user["name"] == USERNAME @pytest.mark.asyncio - async def test_multi_call(self, async_populated_api): + async def test_multi_call(self, async_populated_api, mock_async_query): + mock_async_query.return_value = [mock_user_response(), mock_version_response()] calls = [["Get", dict(typeName="User", search=dict(name="{0}".format(USERNAME)))], ["GetVersion"]] results = await async_populated_api.multi_call_async(calls) assert len(results) == 2 @@ -83,29 +84,34 @@ async def test_multi_call(self, async_populated_api): assert 3 <= version_len <= 4 @pytest.mark.asyncio - async def test_pythonic_parameters(self, async_populated_api): + async def test_pythonic_parameters(self, async_populated_api, mock_sync_query, mock_async_query): + mock_sync_query.return_value = mock_user_response() + mock_async_query.return_value = mock_user_response() users = async_populated_api.get("User") count_users = await async_populated_api.call_async("Get", type_name="User") assert len(count_users) >= 1 assert len(count_users) == len(users) @pytest.mark.asyncio - async def test_api_from_credentials(self, async_populated_api): + async def test_api_from_credentials(self, async_populated_api, mock_async_query): + mock_async_query.return_value = mock_user_response() new_api = API.from_credentials(async_populated_api.credentials) users = await new_api.get_async("User") assert len(users) >= 1 @pytest.mark.asyncio - async def test_results_limit(self, async_populated_api): + async def test_results_limit(self, async_populated_api, mock_async_query): + mock_async_query.return_value = mock_user_response() users = await async_populated_api.get_async("User", resultsLimit=1) assert len(users) == 1 @pytest.mark.asyncio - async def test_session_expired(self, async_populated_api): + async def test_session_expired(self, async_populated_api, mock_async_query): credentials = async_populated_api.credentials credentials.password = PASSWORD credentials.session_id = "abc123" test_api = API.from_credentials(credentials) + mock_async_query.return_value = mock_user_response() users = await test_api.get_async("User") assert len(users) >= 1 @@ -115,54 +121,73 @@ async def test_missing_method(self, async_populated_api): await async_populated_api.call_async(None) @pytest.mark.asyncio - async def test_call_without_credentials(self): + async def test_call_without_credentials(self, mock_sync_query, mock_async_query): + mock_sync_query.return_value = mock_authenticate_response() new_api = API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER) + mock_async_query.return_value = mock_user_response() user = await new_api.get_async("User", name="{0}".format(USERNAME)) assert len(user) == 1 @pytest.mark.asyncio - async def test_bad_parameters(self, async_populated_api): + async def test_bad_parameters(self, async_populated_api, mock_async_query): + mock_async_query.side_effect = MyGeotabException( + {"errors": [{"name": "MissingMethodException", "message": "NonExistentMethod could not be found"}]} + ) with pytest.raises(MyGeotabException) as excinfo: await async_populated_api.call_async("NonExistentMethod", not_a_property="abc123") assert "NonExistentMethod" in str(excinfo.value) @pytest.mark.asyncio - async def test_get_search_parameter(self, async_populated_api): + async def test_get_search_parameter(self, async_populated_api, mock_async_query): + mock_async_query.return_value = mock_user_response() user = await async_populated_api.get_async("User", search=dict(name=USERNAME)) assert len(user) == 1 user = user[0] assert user["name"] == USERNAME @pytest.mark.asyncio - async def test_add_edit_remove(self, async_populated_api_entity): - async def get_zonetypes(): - zonetypes = await async_populated_api_entity.get_async("ZoneType", name=ASYNC_ZONETYPE_NAME) - assert len(zonetypes) == 1 - return zonetypes[0] + async def test_add_edit_remove(self, async_populated_api_entity, mock_sync_query, mock_async_query): + zonetype_id = "zt123" + # Get user for company groups (sync) + mock_sync_query.return_value = [{"id": "u123", "name": USERNAME, "companyGroups": [{"id": "g123"}]}] user = async_populated_api_entity.get("User", name=USERNAME)[0] + + # Mock async add + mock_async_query.return_value = zonetype_id 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() + + # Mock async get + mock_async_query.return_value = [mock_zonetype_response(ASYNC_ZONETYPE_NAME, zonetype_id)] + zonetypes = await async_populated_api_entity.get_async("ZoneType", name=ASYNC_ZONETYPE_NAME) + assert len(zonetypes) == 1 + zonetype = zonetypes[0] assert zonetype["name"] == ASYNC_ZONETYPE_NAME + + # Mock async set comment = "some comment" zonetype["comment"] = comment + mock_async_query.return_value = None await async_populated_api_entity.set_async("ZoneType", zonetype) - zonetype = await get_zonetypes() + + # Mock async get after set + mock_async_query.return_value = [mock_zonetype_response(ASYNC_ZONETYPE_NAME, zonetype_id, comment)] + zonetype = (await async_populated_api_entity.get_async("ZoneType", name=ASYNC_ZONETYPE_NAME))[0] assert zonetype["comment"] == comment + + # Mock async remove + mock_async_query.return_value = None await async_populated_api_entity.remove_async("ZoneType", zonetype) + + # Mock async get after remove + mock_async_query.return_value = [] 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) == 3 - @pytest.mark.asyncio async def test_invalid_server_call(self): with pytest.raises(Exception) as excinfo1: @@ -173,7 +198,8 @@ async def test_invalid_server_call(self): assert "server" in str(excinfo2.value) @pytest.mark.asyncio - async def test_timeout(self): + async def test_timeout(self, mock_async_query): + mock_async_query.side_effect = TimeoutException("my36.geotab.com") 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) @@ -181,7 +207,7 @@ async def test_timeout(self): class TestAsyncAuthentication: @pytest.mark.asyncio - async def test_invalid_session(self): + async def test_invalid_session(self, mock_async_query): fake_credentials = generate_fake_credentials() test_api = API( fake_credentials["username"], @@ -190,6 +216,10 @@ async def test_invalid_session(self): ) assert fake_credentials["username"] in str(test_api.credentials) assert fake_credentials["database"] in str(test_api.credentials) + + mock_async_query.side_effect = MyGeotabException( + {"errors": [{"name": "InvalidUserException", "message": "Invalid user"}]} + ) with pytest.raises(AuthenticationException) as excinfo: await test_api.get_async("User") assert "Cannot authenticate" in str(excinfo.value) diff --git a/tests/test_api_call.py b/tests/test_api_call.py index e954f48..ff96035 100644 --- a/tests/test_api_call.py +++ b/tests/test_api_call.py @@ -1,21 +1,19 @@ # -*- coding: utf-8 -*- -import os import random import string +from unittest.mock import patch import pytest from mygeotab import api from mygeotab.exceptions import AuthenticationException, TimeoutException -USERNAME = os.environ.get("MYGEOTAB_USERNAME") -PASSWORD = os.environ.get("MYGEOTAB_PASSWORD") -DATABASE = os.environ.get("MYGEOTAB_DATABASE") -SERVER = os.environ.get("MYGEOTAB_SERVER") -CER_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_CER") -KEY_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_KEY") -PEM_FILE = os.environ.get("MYGEOTAB_CERTIFICATE_PEM") +USERNAME = "test@example.com" +PASSWORD = "testpassword" +DATABASE = "testdatabase" +SERVER = "my3.geotab.com" +SESSION_ID = "abc123sessionid" ZONETYPE_NAME = "mygeotab-python test zonetype" FAKE_USERNAME = "fakeusername" @@ -37,52 +35,77 @@ def generate_fake_credentials(): ) -@pytest.fixture(scope="session") -def populated_api(): - cert = None - if CER_FILE and KEY_FILE: - cert = (CER_FILE, KEY_FILE) - elif PEM_FILE: - cert = PEM_FILE - if USERNAME and PASSWORD: - session = api.API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER, cert=cert) - try: - session.authenticate() - except (api.MyGeotabException, api.AuthenticationException) as exception: - pytest.fail(str(exception)) - return - yield session - else: - pytest.skip( - "Can't make calls to the API without the MYGEOTAB_USERNAME and MYGEOTAB_PASSWORD environment variables being set" - ) +def mock_authenticate_response(): + """Return a mock authentication response.""" + return { + "path": SERVER, + "credentials": { + "userName": USERNAME, + "sessionId": SESSION_ID, + "database": DATABASE, + }, + } + + +def mock_user_response(): + """Return a mock user response.""" + return [{"id": "b123", "name": USERNAME}] + + +def mock_version_response(): + """Return a mock version response.""" + return "8.0.1234" + +def mock_zonetype_response(name, zonetype_id="zt123", comment=None): + """Return a mock zonetype response.""" + result = {"id": zonetype_id, "name": name} + if comment: + result["comment"] = comment + return result -@pytest.fixture(scope="session") -def populated_api_entity(populated_api): - def clean_zonetypes(): - zonetypes = populated_api.get("ZoneType", name=ZONETYPE_NAME) - for zonetype in zonetypes: - populated_api.remove("ZoneType", zonetype) - clean_zonetypes() +@pytest.fixture +def mock_query(): + """Fixture to mock the _query function.""" + with patch("mygeotab.api._query") as mock: + yield mock + + +@pytest.fixture +def populated_api(mock_query): + """Create an API instance with mocked credentials.""" + mock_query.return_value = mock_authenticate_response() + session = api.API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER) + session.authenticate() + mock_query.return_value = None + yield session + + +@pytest.fixture +def populated_api_entity(populated_api, mock_query): + """Create an API instance for entity tests.""" + mock_query.return_value = [] yield populated_api - clean_zonetypes() + mock_query.return_value = [] class TestCallApi: - def test_get_version(self, populated_api): + def test_get_version(self, populated_api, mock_query): + mock_query.return_value = mock_version_response() version = populated_api.call("GetVersion") version_len = len(version.split(".")) assert 3 <= version_len <= 4 - def test_get_user(self, populated_api): + def test_get_user(self, populated_api, mock_query): + mock_query.return_value = mock_user_response() user = populated_api.get("User", name=USERNAME) assert len(user) == 1 user = user[0] assert user["name"] == USERNAME - def test_multi_call(self, populated_api): + def test_multi_call(self, populated_api, mock_query): + mock_query.return_value = [mock_user_response(), mock_version_response()] calls = [["Get", dict(typeName="User", search=dict(name="{0}".format(USERNAME)))], ["GetVersion"]] results = populated_api.multi_call(calls) assert len(results) == 2 @@ -93,26 +116,30 @@ def test_multi_call(self, populated_api): version_len = len(results[1].split(".")) assert 3 <= version_len <= 4 - def test_pythonic_parameters(self, populated_api): + def test_pythonic_parameters(self, populated_api, mock_query): + mock_query.return_value = mock_user_response() users = populated_api.get("User") count_users = populated_api.call("Get", type_name="User") assert len(count_users) >= 1 assert len(count_users) == len(users) - def test_api_from_credentials(self, populated_api): + def test_api_from_credentials(self, populated_api, mock_query): + mock_query.return_value = mock_user_response() new_api = api.API.from_credentials(populated_api.credentials) users = new_api.get("User") assert len(users) >= 1 - def test_results_limit(self, populated_api): + def test_results_limit(self, populated_api, mock_query): + mock_query.return_value = mock_user_response() users = populated_api.get("User", resultsLimit=1) assert len(users) == 1 - def test_session_expired(self, populated_api): + def test_session_expired(self, populated_api, mock_query): credentials = populated_api.credentials credentials.password = PASSWORD credentials.session_id = "abc123" test_api = api.API.from_credentials(credentials) + mock_query.return_value = mock_user_response() users = test_api.get("User") assert len(users) >= 1 @@ -120,23 +147,26 @@ def test_missing_method(self, populated_api): with pytest.raises(Exception): populated_api.call(None) - def test_call_without_credentials(self): - if not (USERNAME and PASSWORD): - pytest.skip( - "Can't make calls to the API without the " - "MYGEOTAB_USERNAME and MYGEOTAB_PASSWORD " - "environment variables being set" - ) + def test_call_without_credentials(self, mock_query): + # Use side_effect to return different values for consecutive calls + mock_query.side_effect = [ + mock_authenticate_response(), # First call: authentication + mock_user_response() # Second call: get user + ] new_api = api.API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER) user = new_api.get("User", name="{0}".format(USERNAME)) assert len(user) == 1 - def test_bad_parameters(self, populated_api): + def test_bad_parameters(self, populated_api, mock_query): + mock_query.side_effect = api.MyGeotabException( + {"errors": [{"name": "MissingMethodException", "message": "NonExistentMethod could not be found"}]} + ) with pytest.raises(api.MyGeotabException) as excinfo: populated_api.call("NonExistentMethod", not_a_property="abc123") assert "NonExistentMethod" in str(excinfo.value) - def test_get_search_parameter(self, populated_api): + def test_get_search_parameter(self, populated_api, mock_query): + mock_query.return_value = mock_user_response() user = populated_api.get("User", search=dict(name=USERNAME)) assert len(user) == 1 user = user[0] @@ -144,29 +174,45 @@ def test_get_search_parameter(self, populated_api): class TestEntity: - def test_add_edit_remove(self, populated_api_entity): - def get_zonetype(): - zonetypes = populated_api_entity.get("ZoneType", name=ZONETYPE_NAME) - assert len(zonetypes) == 1 - return zonetypes[0] + def test_add_edit_remove(self, populated_api_entity, mock_query): + zonetype_id = "zt123" + # Mock add + mock_query.return_value = zonetype_id zonetype = {"name": ZONETYPE_NAME} zonetype["id"] = populated_api_entity.add("ZoneType", zonetype) assert zonetype["id"] is not None - zonetype = get_zonetype() + + # Mock get after add + mock_query.return_value = [mock_zonetype_response(ZONETYPE_NAME, zonetype_id)] + zonetypes = populated_api_entity.get("ZoneType", name=ZONETYPE_NAME) + assert len(zonetypes) == 1 + zonetype = zonetypes[0] assert zonetype["name"] == ZONETYPE_NAME + + # Mock set comment = "some comment" zonetype["comment"] = comment + mock_query.return_value = None populated_api_entity.set("ZoneType", zonetype) - zonetype = get_zonetype() + + # Mock get after set + mock_query.return_value = [mock_zonetype_response(ZONETYPE_NAME, zonetype_id, comment)] + zonetype = populated_api_entity.get("ZoneType", name=ZONETYPE_NAME)[0] assert zonetype["comment"] == comment + + # Mock remove + mock_query.return_value = None populated_api_entity.remove("ZoneType", zonetype) + + # Mock get after remove + mock_query.return_value = [] zonetypes = populated_api_entity.get("ZoneType", name=ZONETYPE_NAME) assert len(zonetypes) == 0 class TestAuthentication: - def test_invalid_session(self): + def test_invalid_session(self, mock_query): fake_credentials = generate_fake_credentials() test_api = api.API( fake_credentials["username"], @@ -175,6 +221,10 @@ def test_invalid_session(self): ) assert fake_credentials["username"] in str(test_api.credentials) assert fake_credentials["database"] in str(test_api.credentials) + + mock_query.side_effect = api.MyGeotabException( + {"errors": [{"name": "InvalidUserException", "message": "Invalid user"}]} + ) with pytest.raises(AuthenticationException) as excinfo: test_api.get("User") assert "Cannot authenticate" in str(excinfo.value) @@ -190,19 +240,24 @@ def test_username_password_exists(self): assert "username" in str(excinfo1.value) assert "password" in str(excinfo2.value) - def test_call_authenticate_sessionid(self, populated_api): - credentials = populated_api.authenticate() + def test_call_authenticate_sessionid(self, mock_query): + mock_query.return_value = mock_authenticate_response() + session = api.API(USERNAME, password=PASSWORD, database=DATABASE, server=SERVER) + credentials = session.authenticate() assert credentials.username == USERNAME assert credentials.database == DATABASE assert credentials.session_id is not None - def test_call_authenticate_invalid_sessionid(self): + def test_call_authenticate_invalid_sessionid(self, mock_query): fake_credentials = generate_fake_credentials() test_api = api.API( fake_credentials["username"], session_id=fake_credentials["sessionid"], database=fake_credentials["database"], ) + mock_query.side_effect = api.MyGeotabException( + {"errors": [{"name": "InvalidUserException", "message": "Invalid user"}]} + ) with pytest.raises(AuthenticationException) as excinfo: test_api.authenticate() assert "Cannot authenticate" in str(excinfo.value) @@ -210,11 +265,6 @@ def test_call_authenticate_invalid_sessionid(self): class TestServerCallApi: - def test_get_version(self): - version = api.server_call("GetVersion", server="my3.geotab.com") - version_split = version.split(".") - assert len(version_split) == 3 - def test_invalid_server_call(self): with pytest.raises(Exception) as excinfo1: api.server_call(None, None) @@ -223,7 +273,8 @@ def test_invalid_server_call(self): assert "method" in str(excinfo1.value) assert "server" in str(excinfo2.value) - def test_timeout(self): + def test_timeout(self, mock_query): + mock_query.side_effect = TimeoutException("my36.geotab.com") with pytest.raises(TimeoutException) as excinfo: api.server_call("GetVersion", server="my36.geotab.com", timeout=0.01) assert "Request timed out @ my36.geotab.com" in str(excinfo.value) diff --git a/tests/test_api_live.py b/tests/test_api_live.py new file mode 100644 index 0000000..491ce5b --- /dev/null +++ b/tests/test_api_live.py @@ -0,0 +1,32 @@ +# -*- coding: utf-8 -*- + +""" +Live API tests - only unauthenticated GetVersion calls. + +These tests actually hit the MyGeotab API servers and require network access. +""" + +import pytest + +from mygeotab import api, server_call_async + + +class TestLiveServerCall: + """Test unauthenticated server_call to live API.""" + + def test_get_version(self): + """Test GetVersion call using sync server_call.""" + version = api.server_call("GetVersion", server="my.geotab.com") + version_split = version.split(".") + assert len(version_split) == 3 + + +class TestLiveServerCallAsync: + """Test unauthenticated server_call_async to live API.""" + + @pytest.mark.asyncio + async def test_get_version(self): + """Test GetVersion call using async server_call_async.""" + version = await server_call_async("GetVersion", server="my.geotab.com") + version_split = version.split(".") + assert len(version_split) == 3