"""Unit tests for ImpersonationM2MTokenManager class.""" import datetime import json from typing import Any from unittest.mock import MagicMock, patch import freezegun import pytest from owsclient.constants import PROD_AUTH0_URL, QA_AUTH0_URL from owsclient.m2m.base import M2MToken from owsclient.m2m.impersonation import ( M2M_CLIENT_CREDENTIALS_SECRET_NAME, Auth0TokenException, ClientCredentials, ImpersonationM2MTokenManager, OAuthToken, ) from owsclient.protocols import Cache, SecretsManager TEST_M2M_JWT_ACCESS_TOKEN_STR = json.dumps( { "expires_at": "2024-07-22 16:55:30.13455+0000", "token": "TEST_TOKEN", "client_credentials_secret_arn": "arn:aws:secretsmanager:us-east-1:1234:secret:qa/lambda-test-m2m-client/M2M_AUTH0_CLIENT_CREDENTIALS-H3WvVs", # noqa: E501 } ) @pytest.fixture() def mock_secrets_manager() -> MagicMock: """Create a mock secrets manager instance.""" return MagicMock(spec=SecretsManager) @pytest.fixture() def mock_cache() -> MagicMock: """Create a mock cache instance.""" return MagicMock(spec=Cache) def test_impersonation_m2m_token_manager_init( mock_secrets_manager: MagicMock, ) -> None: """Test ImpersonationM2MTokenManager initialization.""" obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) assert obj.service_name == "my-app" assert obj.environment == "qa" assert obj._cache == {} def test_impersonation_m2m_token_manager_init_with_cache( mock_secrets_manager: MagicMock, mock_cache: MagicMock, ) -> None: """Test ImpersonationM2MTokenManager initialization with cache.""" obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, ) assert obj._cache == mock_cache @pytest.mark.parametrize( "environment, auth0_url, expected_auth0_url", [ pytest.param( "dev", None, QA_AUTH0_URL, id="DEV environment with default QA auth0_url", ), pytest.param( "qa", None, QA_AUTH0_URL, id="QA environment with default QA auth0_url", ), pytest.param( "prod", None, PROD_AUTH0_URL, id="PROD environment with default PROD auth0_url", ), pytest.param( "qa", "https://custom-auth0.example.com", "https://custom-auth0.example.com", id="QA environment with custom auth0_url", ), pytest.param( "prod", "https://custom-auth0.example.com", "https://custom-auth0.example.com", id="PROD environment with custom auth0_url", ), ], ) def test_impersonation_m2m_token_manager_init_auth0_url( mock_secrets_manager: MagicMock, environment: str, auth0_url: str | None, expected_auth0_url: str, ) -> None: """Test ImpersonationM2MTokenManager initialization with various auth0_url configurations.""" kwargs: dict[str, Any] = { "secrets_manager": mock_secrets_manager, "environment": environment, "service_name": "my-app", } if auth0_url is not None: kwargs["auth0_url"] = auth0_url obj = ImpersonationM2MTokenManager(**kwargs) assert obj._auth0_url == expected_auth0_url @pytest.mark.parametrize( "auth0_request_timeout, expected_timeout", [ pytest.param( None, 10.0, id="default timeout", ), pytest.param( 15.0, 15.0, id="custom timeout 15.0", ), ], ) def test_impersonation_m2m_token_manager_init_auth0_request_timeout( mock_secrets_manager: MagicMock, auth0_request_timeout: float | None, expected_timeout: float, ) -> None: """Test ImpersonationM2MTokenManager initialization with various auth0_request_timeout configurations.""" kwargs: dict[str, Any] = { "secrets_manager": mock_secrets_manager, "environment": "qa", "service_name": "my-app", } if auth0_request_timeout is not None: kwargs["auth0_request_timeout"] = auth0_request_timeout obj = ImpersonationM2MTokenManager(**kwargs) assert obj._auth0_request_timeout == expected_timeout def test_impersonation_m2m_token_manager_generate_secret_name( mock_secrets_manager: MagicMock, ) -> None: """Test ImpersonationM2MTokenManager.generate_secret_name().""" obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) secret_name = obj.generate_secret_name() assert secret_name == f"qa/my-app/{M2M_CLIENT_CREDENTIALS_SECRET_NAME}" def test_impersonation_m2m_token_manager_generate_m2m_token_cache_key( mock_secrets_manager: MagicMock, ) -> None: """Test ImpersonationM2MTokenManager.generate_m2m_token_cache_key().""" obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) impersonated_identity_uuid = "f94b0c5a-b520-486b-ac17-e59e9888b8bd" cache_key = obj.generate_m2m_token_cache_key( impersonated_identity_uuid=impersonated_identity_uuid ) assert ( cache_key == "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd" ) def test_generate_auth0_credentials_cache_key( mock_secrets_manager: MagicMock, ) -> None: """Test that the cache key is generated correctly for Auth0 client credentials.""" # Arrange manager = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="test", service_name="test-service", ) cache_key = manager.generate_auth0_credentials_cache_key() assert cache_key == "_auth0_client_credentials|service_name#test-service" @pytest.mark.parametrize( "secret_value", [ pytest.param( { "audience": "https://api.example.com", "client_id": "test_client_id", "client_secret": "test_client_secret", "grant_type": "client_credentials", }, id="dict", ), pytest.param( json.dumps( { "audience": "https://api.example.com", "client_id": "test_client_id", "client_secret": "test_client_secret", "grant_type": "client_credentials", } ), id="json_string", ), ], ) def test_impersonation_m2m_token_manager_get_client_credentials_from_secret_manager( mock_secrets_manager: MagicMock, secret_value: dict[str, Any] | str, ) -> None: """Test ImpersonationM2MTokenManager._get_client_credentials_from_secret_manager().""" mock_secrets_manager.get_secret.return_value = secret_value obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) client_credentials = obj._get_client_credentials_from_secret_manager() mock_secrets_manager.get_secret.assert_called_once_with( secret_name=f"qa/my-app/{M2M_CLIENT_CREDENTIALS_SECRET_NAME}" ) assert isinstance(client_credentials, ClientCredentials) assert client_credentials.audience == "https://api.example.com" assert client_credentials.client_id == "test_client_id" assert client_credentials.client_secret == "test_client_secret" assert client_credentials.grant_type == "client_credentials" @pytest.mark.parametrize( "mock_cache, expected", [ pytest.param({}, None, id="An empty cache dict returns None"), pytest.param( {"not_the_key": 123}, None, id="A cache dict without the key returns None" ), pytest.param( {"_auth0_client_credentials|service_name#my-app": None}, None, id="When the cached value is None, None is returned", ), pytest.param( {"_auth0_client_credentials|service_name#my-app": 123}, None, id="When the cached value is an integer, None is returned", ), pytest.param( {"_auth0_client_credentials|service_name#my-app": True}, None, id="When the cached value is a boolean, None is returned", ), pytest.param( {"_auth0_client_credentials|service_name#my-app": "some string"}, None, id="When the cached value cannot be validated as ClientCredentials, None is returned", ), pytest.param( { "_auth0_client_credentials|service_name#my-app": ClientCredentials( audience="https://api.example.com", client_id="test_client_id", client_secret="test_client_secret", ) }, None, id="When the cached value is a ClientCredentials object, None is returned", ), pytest.param( { "_auth0_client_credentials|service_name#my-app": json.dumps( { "audience": "https://api.example.com", "client_id": "test_client_id", "client_secret": "test_client_secret", "grant_type": "client_credentials", } ) }, ClientCredentials( audience="https://api.example.com", client_id="test_client_id", client_secret="test_client_secret", grant_type="client_credentials", ), id="The cached value is validated as ClientCredentials", ), ], ) def test_impersonation_m2m_token_manager__get_client_credentials_from_cache( mock_secrets_manager: MagicMock, mock_cache: dict[str, Any], expected: ClientCredentials | None, ) -> None: """Test ImpersonationM2MTokenManager._get_client_credentials_from_cache().""" obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, ) actual = obj._get_client_credentials_from_cache() assert actual == expected @freezegun.freeze_time("2022-01-01 00:00:00") @patch("owsclient.m2m.impersonation.generate_jwt_token_with_impersonation_with_retries") @patch.object(ImpersonationM2MTokenManager, "_get_client_credentials") def test_impersonation_m2m_token_manager_generate_auth_token( mock_get_client_credentials: MagicMock, mock_generate_jwt: MagicMock, mock_secrets_manager: MagicMock, ) -> None: """Test generate_auth_token successfully generates an M2M token.""" client_credentials = ClientCredentials( audience="https://api.example.com", client_id="test_client_id", client_secret="test_client_secret", grant_type="client_credentials", ) mock_get_client_credentials.return_value = client_credentials oauth_token = OAuthToken( access_token="test_access_token", expires_in=3600, token_type="Bearer", ) mock_generate_jwt.return_value = oauth_token obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", auth0_request_timeout=15.0, ) impersonated_identity_uuid = "f94b0c5a-b520-486b-ac17-e59e9888b8bd" result = obj.generate_auth_token(impersonated_identity_uuid) assert isinstance(result, M2MToken) assert result == M2MToken( token="test_access_token", expires_at=datetime.datetime(2022, 1, 1, 1, 0, 0, tzinfo=datetime.UTC), ) mock_get_client_credentials.assert_called_once() mock_generate_jwt.assert_called_once_with( client_credentials=client_credentials, impersonated_identity_uuid=impersonated_identity_uuid, timeout=15.0, auth0_url=QA_AUTH0_URL, ) @patch("owsclient.m2m.impersonation.generate_jwt_token_with_impersonation_with_retries") @patch.object(ImpersonationM2MTokenManager, "_get_client_credentials") def test_impersonation_m2m_token_manager_generate_auth_token_raises_on_auth0_error( mock_get_client_credentials: MagicMock, mock_generate_jwt: MagicMock, mock_secrets_manager: MagicMock, ) -> None: """Test generate_auth_token propagates Auth0TokenException.""" client_credentials = ClientCredentials( audience="https://api.example.com", client_id="test_client_id", client_secret="test_client_secret", grant_type="client_credentials", ) mock_get_client_credentials.return_value = client_credentials mock_generate_jwt.side_effect = Auth0TokenException("Auth0 error") obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) with pytest.raises(Auth0TokenException, match="Auth0 error"): obj.generate_auth_token("test-uuid") @pytest.mark.parametrize( "mock_cache, impersonated_identity_uuid, expected", [ pytest.param( {}, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="An empty cache dict returns None", ), pytest.param( {"not_the_key": 123}, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="A cache dict without the key returns None", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd": None }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="When the cached value is None, None is returned", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd": 123 }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="When the cached value is an integer, None is returned", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd": True }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="When the cached value is a boolean, None is returned", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd": "some string" }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="When the cached value cannot be validated as an M2MToken, None is returned", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd": M2MToken.model_validate_json( TEST_M2M_JWT_ACCESS_TOKEN_STR ) }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="When the cached value is an M2MToken object, None is returned", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd": TEST_M2M_JWT_ACCESS_TOKEN_STR }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", M2MToken.model_validate_json(TEST_M2M_JWT_ACCESS_TOKEN_STR), id="The cached value is validated as an M2MToken", ), pytest.param( { "_m2m_token|service_name#my-app|impersonated_identity_uuid#different-uuid": TEST_M2M_JWT_ACCESS_TOKEN_STR }, "f94b0c5a-b520-486b-ac17-e59e9888b8bd", None, id="Cache key for different identity UUID returns None", ), ], ) def test_impersonation_m2m_token_manager__get_token_payload_from_cache( mock_secrets_manager: MagicMock, mock_cache: Cache | dict[str, Any], impersonated_identity_uuid: str, expected: M2MToken | None, ) -> None: """Test ImpersonationM2MTokenManager._get_token_payload_from_cache().""" obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, ) actual = obj._get_token_payload_from_cache(impersonated_identity_uuid) assert actual == expected @freezegun.freeze_time("2022-01-01 00:00:00") @patch.object(ImpersonationM2MTokenManager, "generate_auth_token") @patch.object(ImpersonationM2MTokenManager, "_get_token_payload_from_cache") def test_impersonation_m2m_token_manager_get_token_string( mock_get_from_cache: MagicMock, mock_generate: MagicMock, mock_secrets_manager: MagicMock, mock_cache: MagicMock, ) -> None: """Test get_token_string generates new token when cache is empty.""" mock_get_from_cache.return_value = None new_token = M2MToken( token="new_access_token", expires_at=datetime.datetime(2022, 1, 1, 1, 0, 0, tzinfo=datetime.UTC), ) mock_generate.return_value = new_token obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, ) result = obj.get_token_string("f94b0c5a-b520-486b-ac17-e59e9888b8bd") assert result == "new_access_token" mock_get_from_cache.assert_called_once() mock_generate.assert_called_once_with( impersonated_identity_uuid="f94b0c5a-b520-486b-ac17-e59e9888b8bd" ) cache_key = "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd" expected_cached_value = new_token.model_dump_json() mock_cache.set.assert_called_once_with( cache_key, value=expected_cached_value, timeout=0 ) @freezegun.freeze_time("2022-01-01 00:00:00") @patch.object(ImpersonationM2MTokenManager, "generate_auth_token") @patch.object(ImpersonationM2MTokenManager, "_get_token_payload_from_cache") def test_impersonation_m2m_token_manager_get_token_string_uses_cached_token( mock_get_from_cache: MagicMock, mock_generate: MagicMock, mock_secrets_manager: MagicMock, ) -> None: """Test get_token_string returns cached token when valid and not expired.""" cached_token = M2MToken( token="cached_access_token", expires_at=datetime.datetime(2022, 1, 1, 1, 0, 0, tzinfo=datetime.UTC), ) mock_get_from_cache.return_value = cached_token obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) result = obj.get_token_string("f94b0c5a-b520-486b-ac17-e59e9888b8bd") assert result == "cached_access_token" mock_get_from_cache.assert_called_once_with( impersonated_identity_uuid="f94b0c5a-b520-486b-ac17-e59e9888b8bd" ) mock_generate.assert_not_called() @freezegun.freeze_time("2022-01-01 00:59:00") @patch.object(ImpersonationM2MTokenManager, "generate_auth_token") @patch.object(ImpersonationM2MTokenManager, "_get_token_payload_from_cache") def test_impersonation_m2m_token_manager_get_token_string_generates_new_token_when_cached_expired( mock_get_from_cache: MagicMock, mock_generate: MagicMock, mock_secrets_manager: MagicMock, mock_cache: MagicMock, ) -> None: """Test get_token_string generates new token when cached token is expired.""" expired_token = M2MToken( token="expired_token", expires_at=datetime.datetime(2022, 1, 1, 1, 0, 0, tzinfo=datetime.UTC), ) mock_get_from_cache.return_value = expired_token new_token = M2MToken( token="refreshed_access_token", expires_at=datetime.datetime(2022, 1, 1, 2, 0, 0, tzinfo=datetime.UTC), ) mock_generate.return_value = new_token obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, leeway_seconds=120, ) result = obj.get_token_string("f94b0c5a-b520-486b-ac17-e59e9888b8bd") assert result == "refreshed_access_token" mock_generate.assert_called_once() mock_get_from_cache.assert_called_once() @freezegun.freeze_time("2022-01-01 00:00:00") @patch.object(ImpersonationM2MTokenManager, "generate_auth_token") @patch.object(ImpersonationM2MTokenManager, "_get_token_payload_from_cache") def test_impersonation_m2m_token_manager_get_token_string_caches_new_token_with_dict_cache( mock_get_from_cache: MagicMock, mock_generate: MagicMock, mock_secrets_manager: MagicMock, ) -> None: """Test get_token_string caches new token when using dict cache.""" mock_get_from_cache.return_value = None new_token = M2MToken( token="new_access_token", expires_at=datetime.datetime(2022, 1, 1, 1, 0, 0, tzinfo=datetime.UTC), ) mock_generate.return_value = new_token obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) result = obj.get_token_string("f94b0c5a-b520-486b-ac17-e59e9888b8bd") assert result == "new_access_token" cache_key = "_m2m_token|service_name#my-app|impersonated_identity_uuid#f94b0c5a-b520-486b-ac17-e59e9888b8bd" assert isinstance(obj._cache, dict) assert cache_key in obj._cache assert obj._cache[cache_key] == new_token.model_dump_json() @patch.object( ImpersonationM2MTokenManager, "_get_client_credentials_from_secret_manager" ) @patch.object(ImpersonationM2MTokenManager, "_get_client_credentials_from_cache") def test_impersonation_m2m_token_manager_get_client_credentials__cached( mock_get_from_cache: MagicMock, mock_get_from_secrets: MagicMock, mock_secrets_manager: MagicMock, mock_cache: MagicMock, ) -> None: """Test _get_client_credentials returns cached credentials when available.""" cached_credentials = ClientCredentials( audience="https://api.example.com", client_id="cached_client_id", client_secret="cached_client_secret", grant_type="client_credentials", ) mock_get_from_cache.return_value = cached_credentials obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, ) result = obj._get_client_credentials() assert result == cached_credentials mock_get_from_cache.assert_called_once() mock_get_from_secrets.assert_not_called() mock_cache.set.assert_not_called() @patch.object( ImpersonationM2MTokenManager, "_get_client_credentials_from_secret_manager" ) @patch.object(ImpersonationM2MTokenManager, "_get_client_credentials_from_cache") def test_impersonation_m2m_token_manager_get_client_credentials__not_cached( mock_get_from_cache: MagicMock, mock_get_from_secrets: MagicMock, mock_secrets_manager: MagicMock, mock_cache: MagicMock, ) -> None: """Test _get_client_credentials fetches from secrets manager when cache is empty.""" credentials = ClientCredentials( audience="https://api.example.com", client_id="secret_client_id", client_secret="secret_client_secret", grant_type="client_credentials", ) mock_get_from_cache.return_value = None mock_get_from_secrets.return_value = credentials obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", cache=mock_cache, ) result = obj._get_client_credentials() assert result == credentials mock_get_from_cache.assert_called_once() mock_get_from_secrets.assert_called_once() mock_cache.set.assert_called_once() cache_key = "_auth0_client_credentials|service_name#my-app" expected_cached_value = credentials.model_dump_json() mock_cache.set.assert_called_once_with( cache_key, value=expected_cached_value, timeout=0 ) @patch.object( ImpersonationM2MTokenManager, "_get_client_credentials_from_secret_manager" ) @patch.object(ImpersonationM2MTokenManager, "_get_client_credentials_from_cache") def test_impersonation_m2m_token_manager_get_client_credentials_with_dict_cache( mock_get_from_cache: MagicMock, mock_get_from_secrets: MagicMock, mock_secrets_manager: MagicMock, ) -> None: """Test _get_client_credentials works with dict cache.""" credentials = ClientCredentials( audience="https://api.example.com", client_id="secret_client_id", client_secret="secret_client_secret", grant_type="client_credentials", ) mock_get_from_cache.return_value = None mock_get_from_secrets.return_value = credentials obj = ImpersonationM2MTokenManager( secrets_manager=mock_secrets_manager, environment="qa", service_name="my-app", ) result = obj._get_client_credentials() assert result == credentials cache_key = "_auth0_client_credentials|service_name#my-app" assert isinstance(obj._cache, dict) assert cache_key in obj._cache assert obj._cache[cache_key] == credentials.model_dump_json()