Remove sleep from tests

Author: carl-alexander-adams

noxfile.py

@@ -3,7 +3,19 @@
 nox.options.stop_on_first_error = True
 nox.options.reuse_existing_virtualenvs = False
 
-nox.options.sessions = ["test", "lint"]
+# Default sessions - all tests, but not packaging
+nox.options.sessions = [
+    "black_lint",
+    "pytest",
+    "semgrep_src",
+    "mypy",
+    "pyflakes_src",
+    "pyflakes_examples",
+    "pyflakes_tests",
+    "pylint_src",
+    "pylint_examples",
+    "pylint_tests",
+]
 
 _DEFAULT_PYTHON = "3.13"
 _ALL_PYTHON = ["3.9", "3.10", "3.11", "3.12", "3.13"]

pyproject.toml

@@ -55,6 +55,7 @@ examples = [
 tests = [
     "black",
     "coverage[toml]",
+    "freezegun",
     "mypy",
     "nox",
     "pyflakes",

src/planet_auth/oidc/api_clients/token_api_client.py

@@ -75,6 +75,7 @@ def _polling_checked_call(self, token_params, timeout, poll_interval, auth_enric
                 else:
                     raise oe
             if sleep_counter < timeout:
+                # FIXME: a weakness here is we are not counting the time spent in the HTTP call against the timeout.
                 time.sleep(poll_interval)
                 sleep_counter += poll_interval
             else:

tests/test_planet_auth/unit/auth/auth_clients/oidc/api_clients/test_token_api_client.py

@@ -24,6 +24,8 @@
 from planet_auth.oidc.api_clients.token_api_client import TokenApiClient, TokenApiException
 from planet_auth.oidc.util import create_pkce_challenge_verifier_pair
 
+from tests.test_planet_auth.util import mock_sleep_skip
+
 TEST_API_ENDPOINT = "https://blackhole.unittest.planet.com/api"
 TEST_CLIENT_ID = "__test_client_id__"
 TEST_REDIRECT_URI = "__test_redirect_uri__"
@@ -224,6 +226,7 @@ def test_token_from_device_code(self, mock_post):
         )
         self.assertEqual(API_RESPONSE_VALID, token_response)
 
+    @mock.patch("time.sleep", mock_sleep_skip)
     @mock.patch("requests.sessions.Session.post", side_effect=mocked_response_pending_then_valid)
     def test_token_from_device_code_pending_then_ok(self, mock_post):
         under_test = TokenApiClient(token_uri=TEST_API_ENDPOINT)
@@ -237,8 +240,11 @@ def test_token_from_device_code_pending_then_ok(self, mock_post):
         self.assertEqual(API_RESPONSE_VALID, token_response)
         self.assertEqual(mock_post.call_count, 2)
 
+    @mock.patch("time.sleep", mock_sleep_skip)
     @mock.patch("requests.sessions.Session.post", side_effect=mocked_response_slow_down)
     def test_token_from_device_code_slow_down(self, mock_post):
+        # FIXME: we get away with only mocking sleep() and not the
+        #     the clock because the timout implementation is a little weak.
         under_test = TokenApiClient(token_uri=TEST_API_ENDPOINT)
         with self.assertRaises(TokenApiException):
             under_test.poll_for_token_from_device_code(
@@ -255,6 +261,7 @@ def test_token_from_device_code_slow_down(self, mock_post):
         self.assertEqual(mock_post.call_count, 3)
         # TODO check wall-clock time?
 
+    @mock.patch("time.sleep", mock_sleep_skip)
     @mock.patch("requests.sessions.Session.post", side_effect=mocked_response_pending)
     def test_token_from_device_code_client_provided_timeout(self, mock_post):
         under_test = TokenApiClient(token_uri=TEST_API_ENDPOINT)

tests/test_planet_auth/unit/auth/auth_clients/oidc/auth_clients/test_device_code.py

@@ -23,7 +23,7 @@
     DeviceCodeAuthClientException,
 )
 from tests.test_planet_auth.unit.auth.util import StubOidcClientConfig, StubOidcAuthClient
-from tests.test_planet_auth.util import tdata_resource_file_path
+from tests.test_planet_auth.util import tdata_resource_file_path, mock_sleep_skip
 
 TEST_AUTH_SERVER = "https://blackhole.unittest.planet.com/fake_authserver"
 TEST_CLIENT_ID = "fake_test_client_id"
@@ -86,6 +86,7 @@ def setUp(self):
             )
         )
 
+    @mock.patch("time.sleep", mock_sleep_skip)
     @mock.patch(
         "planet_auth.oidc.api_clients.token_api_client.TokenApiClient.poll_for_token_from_device_code",
         mocked_tokenapi_from_device_code,
@@ -108,6 +109,7 @@ def test_login_1(self):
         self.assertIsInstance(test_result, FileBackedOidcCredential)
         # self.assertEqual(MOCK_TOKEN, test_result.data())
 
+    @mock.patch("time.sleep", mock_sleep_skip)
     @mock.patch(
         "planet_auth.oidc.api_clients.token_api_client.TokenApiClient.poll_for_token_from_device_code",
         mocked_tokenapi_from_device_code,

tests/test_planet_auth/unit/auth/auth_clients/oidc/test_oidc_request_authenticator.py

@@ -16,7 +16,7 @@
 
 import pathlib
 import tempfile
-import time
+import freezegun
 import unittest
 
 from planet_auth.oidc.oidc_credential import FileBackedOidcCredential
@@ -164,7 +164,8 @@ def mock_api_call(under_test):
         # pre_request_hook()
         under_test.pre_request_hook()
 
-    def test_happy_path_1(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_happy_path_1(self, frozen_time):
         # The first API call should trigger a token load from disk.
         # If the token is current, the auth client should be untouched.
         under_test = self.under_test_happy_path()
@@ -185,13 +186,14 @@ def test_happy_path_1(self):
 
         # When the token reaches the 3/4 life, the authenticator should
         # attempt a token refresh
-        time.sleep(((3 * TEST_TOKEN_TTL) / 4) + 2)
+        frozen_time.tick(((3 * TEST_TOKEN_TTL) / 4) + 2)
         self.mock_api_call(under_test)
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
         access_token_t3 = under_test._credential.access_token()
         self.assertNotEqual(access_token_t1, access_token_t3)
 
-    def test_happy_path_1_in_memory(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_happy_path_1_in_memory(self, frozen_time):
         # The first API call should trigger a token load from disk.
         # If the token is current, the auth client should be untouched.
         under_test = self.under_test_happy_path_in_memory()
@@ -212,24 +214,26 @@ def test_happy_path_1_in_memory(self):
 
         # When the token reaches the 3/4 life, the authenticator should
         # attempt a token refresh
-        time.sleep(((3 * TEST_TOKEN_TTL) / 4) + 2)
+        frozen_time.tick(((3 * TEST_TOKEN_TTL) / 4) + 2)
         self.mock_api_call(under_test)
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
         access_token_t3 = under_test._credential.access_token()
         self.assertNotEqual(access_token_t1, access_token_t3)
 
-    def test_happy_path_2(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_happy_path_2(self, frozen_time):
         # The first API call should trigger a token load.
         # If the token is past the refresh time, a token refresh should be
         # attempted before the first use
         under_test = self.under_test_happy_path()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
         self.mock_api_call(under_test)
         self.assertIsNotNone(under_test._credential.data())
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
 
-    def test_refresh_fails(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_refresh_fails(self, frozen_time):
         # Refresh could fail.  If it does, this should be buried and we
         # should try to carry on with the token we have.
         # (this is why we refresh before expiry, so transient failures
@@ -242,15 +246,16 @@ def test_refresh_fails(self):
         self.assertEqual(0, under_test._auth_client.refresh.call_count)
         access_token_t1 = under_test._credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         self.mock_api_call(under_test)
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
         access_token_t2 = under_test._credential.access_token()
 
         self.assertEqual(access_token_t1, access_token_t2)
 
-    def test_no_refresh_token(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_no_refresh_token(self, frozen_time):
         # if we have no refresh token, what happens when we expect to
         # refresh? It's not expected that a user use the refreshing
         # authenticator when not using a refresh token (there is another
@@ -265,22 +270,23 @@ def test_no_refresh_token(self):
         access_token_t1 = under_test._credential.access_token()
         self.assertEqual(0, under_test._auth_client.refresh.call_count)
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         self.mock_api_call(under_test)
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
         access_token_t2 = under_test._credential.access_token()
         self.assertEqual(access_token_t1, access_token_t2)
 
-    def test_no_auth_client(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_no_auth_client(self, frozen_time):
         # when no auth client is provided, just authenticate with what we
         # have.
         under_test = self.under_test_no_auth_client()
 
         self.mock_api_call(under_test)
         access_token_t1 = under_test._credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         self.mock_api_call(under_test)
         access_token_t2 = under_test._credential.access_token()
@@ -304,7 +310,8 @@ def test_no_access_token(self):
         self.mock_api_call(under_test)
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
 
-    def test_out_of_band_update_1(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_out_of_band_update_1(self, frozen_time):
         # Out of band credential update.  We expect the refresher to reload
         # the credential file without attempting a refresh.
         under_test = self.under_test_happy_path()
@@ -315,7 +322,7 @@ def test_out_of_band_update_1(self):
         self.assertEqual(0, under_test._auth_client.refresh.call_count)
         access_token_t1 = under_test._credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         credential_t1 = under_test._credential
         oob_credential = self.stub_auth_client.refresh(credential_t1.refresh_token())
@@ -330,7 +337,8 @@ def test_out_of_band_update_1(self):
         self.assertNotEqual(access_token_t1, access_token_t2)
         self.assertEqual(access_token_oob, access_token_t2)
 
-    def test_out_of_band_update_2(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_out_of_band_update_2(self, frozen_time):
         # Out of band credential update.  Something updated the credential
         # on disk underneath us. We expect the refresher to reload
         # the credential file without attempting a refresh.
@@ -348,7 +356,7 @@ def test_out_of_band_update_2(self):
         oob_credential.save()
         access_token_oob = oob_credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         # The new token is outside TTL, even after the token is reloaded, this
         # should be detected and a refresh should still occur.
@@ -358,7 +366,8 @@ def test_out_of_band_update_2(self):
         self.assertNotEqual(access_token_t1, access_token_t2)
         self.assertNotEqual(access_token_oob, access_token_t2)
 
-    def test_side_band_update_credential(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_side_band_update_credential(self, frozen_time):
         # a side-band update of the credential should reset our "refresh at" time,
         # cascading behavior from there.  Where the above happens when someone
         # modifies a credential on disk, this happens when something in memory
@@ -378,7 +387,7 @@ def test_side_band_update_credential(self):
         self.assertEqual(0, under_test._auth_client.refresh.call_count)
 
         # Get the sideband credential when we would have expected a refresh to probe behavior.
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
         sideband_credential_path = self.tmp_dir_path / "test_sideband.json"
         sideband_credential = self.mock_auth_login_and_command_initialize(
             credential_path=sideband_credential_path, auth_client=self.stub_auth_client
@@ -461,7 +470,8 @@ def mock_api_call(self, under_test):
         # pre_request_hook()
         under_test.pre_request_hook()
 
-    def test_refresh_token_calls_refresh(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_refresh_token_calls_refresh(self, frozen_time):
         under_test = self.under_test_with_refresh_token()
 
         self.assertIsNone(under_test._credential.data())
@@ -471,15 +481,16 @@ def test_refresh_token_calls_refresh(self):
         self.assertEqual(0, under_test._auth_client.login.call_count)
         access_token_t1 = under_test._credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         self.mock_api_call(under_test)
         self.assertEqual(1, under_test._auth_client.refresh.call_count)
         self.assertEqual(0, under_test._auth_client.login.call_count)
         access_token_t2 = under_test._credential.access_token()
         self.assertNotEqual(access_token_t1, access_token_t2)
 
-    def test_no_refresh_token_calls_login(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_no_refresh_token_calls_login(self, frozen_time):
         under_test = self.under_test_without_refresh_token()
 
         self.assertIsNone(under_test._credential.data())
@@ -489,23 +500,24 @@ def test_no_refresh_token_calls_login(self):
         self.assertEqual(0, under_test._auth_client.login.call_count)
         access_token_t1 = under_test._credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         self.mock_api_call(under_test)
         self.assertEqual(0, under_test._auth_client.refresh.call_count)
         self.assertEqual(1, under_test._auth_client.login.call_count)
         access_token_t2 = under_test._credential.access_token()
         self.assertNotEqual(access_token_t1, access_token_t2)
 
-    def test_no_auth_client(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_no_auth_client(self, frozen_time):
         # when no auth client is provided, just authenticate with what we
         # have.
         under_test = self.under_test_no_auth_client()
 
         self.mock_api_call(under_test)
         access_token_t1 = under_test._credential.access_token()
 
-        time.sleep(TEST_TOKEN_TTL + 2)
+        frozen_time.tick(TEST_TOKEN_TTL + 2)
 
         self.mock_api_call(under_test)
         access_token_t2 = under_test._credential.access_token()

tests/test_planet_auth/unit/auth/auth_clients/oidc/test_token_validator.py

@@ -15,7 +15,7 @@
 import json
 import jwt.utils
 import secrets
-import time
+import freezegun
 import unittest
 from unittest.mock import MagicMock
 
@@ -447,13 +447,14 @@ def test_access_token_multiple_audiences(self):
                 token_str=access_token, issuer=self.token_builder_1.issuer, audience="extra_audience_3"
             )
 
-    def test_access_token_expired(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_access_token_expired(self, frozen_time):
         # QE TC2
         under_test = self.under_test_1
         access_token = self.token_builder_1.construct_oidc_access_token_rfc8693(
             username=TEST_TOKEN_USER, requested_scopes=TEST_TOKEN_SCOPES, ttl=3
         )
-        time.sleep(5)
+        frozen_time.tick(5)
         with self.assertRaises(ExpiredTokenException):
             under_test.validate_token(
                 access_token,
@@ -631,7 +632,8 @@ def test_validate_id_token_multiple_audiences(self):
         with self.assertRaises(InvalidTokenException):
             under_test.validate_id_token(id_token, issuer=self.token_builder_1.issuer, client_id="test_client_id")
 
-    def test_min_jwks_fetch_interval(self):
+    @freezegun.freeze_time(as_kwarg="frozen_time")
+    def test_min_jwks_fetch_interval(self, frozen_time):
         under_test = self.under_test_2
         token_known_signer = self.token_builder_1.construct_oidc_access_token_rfc8693(
             username=TEST_TOKEN_USER, requested_scopes=TEST_TOKEN_SCOPES, ttl=TEST_TOKEN_TTL
@@ -680,7 +682,7 @@ def test_min_jwks_fetch_interval(self):
         # t4 - after the interval has passed, a we should reload for any
         # request.  If a miss comes in before the next interval, it should not
         # trigger a reload.  Also, it's not automatic, it's on demand.
-        time.sleep(self.MIN_JWKS_FETCH_INTERVAL + 2)
+        frozen_time.tick(self.MIN_JWKS_FETCH_INTERVAL + 2)
         self.assertEqual(1, under_test._jwks_client.jwks.call_count)
         under_test.validate_token(
             token_known_signer,