"""Test ClientCredentialsManager.""" from unittest.mock import MagicMock, patch import pytest from m2mconfig.connectors.assume_role_secrets_manager import ( AssumeRoleSecretsManager, SecretScheduledForDeletionError, ) from m2mconfig.connectors.client_credentials_manager import ( ClientCredentialsDoNotExist, ClientCredentialsInvalidFormat, ClientCredentialsManager, ClientCredentialsManagerException, ClientCredentialsSecretScheduledForDeletion, ) from m2mconfig.schemas import ClientCredentials @pytest.fixture() def mock_environment() -> str: """Return reusable environment""" return "test" @pytest.fixture() def mock_machine_name() -> str: """Return reusable machine name.""" return "more-dogs" @pytest.fixture() def expected_secret_name(mock_environment: str, mock_machine_name: str) -> str: """Return expected secret name.""" return f"{mock_environment}/{mock_machine_name}/M2M_AUTH0_CLIENT_CREDENTIALS" @pytest.fixture() def expected_audience() -> str: """Return expected audience.""" return "meyouus" @pytest.fixture() def expected_client_id() -> str: """Return expected client id.""" return "hello" @pytest.fixture() def expected_client_secret() -> str: """Return expected client secret.""" return "dogs" @pytest.fixture() def expected_client_credentials( expected_audience: str, expected_client_id: str, expected_client_secret: str, ) -> ClientCredentials: """Return expected ClientCredentials.""" return ClientCredentials( audience=expected_audience, client_id=expected_client_id, client_secret=expected_client_secret, grant_type="client_credentials", ) @pytest.fixture() def expected_client_credentials_str( expected_audience: str, expected_client_id: str, expected_client_secret: str, ) -> str: """Return expected ClientCredentials, stringified.""" return f'{{"audience":"{expected_audience}","client_id":"{expected_client_id}","client_secret":"{expected_client_secret}","grant_type":"client_credentials"}}' # noqa: E501 @pytest.fixture() def mock_assume_role_secrets_manager() -> MagicMock: """Return mock for secrets manager.""" return MagicMock(spec=AssumeRoleSecretsManager) @pytest.fixture() def client_credentials_manager( mock_environment: str, mock_machine_name: str, mock_assume_role_secrets_manager: MagicMock, ) -> ClientCredentialsManager: """Return reusable ClientCredentialsManager.""" return ClientCredentialsManager( environment=mock_environment, machine_name=mock_machine_name, secrets_manager=mock_assume_role_secrets_manager, ) def test_get_secret_name( client_credentials_manager: ClientCredentialsManager, ) -> None: """Test secret name is returned.""" actual = client_credentials_manager._get_secret_name() assert actual == "test/more-dogs/M2M_AUTH0_CLIENT_CREDENTIALS" @patch.object(ClientCredentialsManager, "_get_secret_name") @pytest.mark.parametrize( "does_secret_exist_value, expected", [ pytest.param(True, True, id="the secret exists, return True"), pytest.param(False, False, id="the secret does not exist, return False"), ], ) def test_exists( mock_get_secret_name: MagicMock, does_secret_exist_value: bool, expected: bool, expected_secret_name: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test ClientCredentials exists.""" mock_get_secret_name.return_value = expected_secret_name mock_assume_role_secrets_manager.does_secret_exist.return_value = ( does_secret_exist_value ) actual = client_credentials_manager.exists() assert actual == expected mock_assume_role_secrets_manager.does_secret_exist.assert_called_once_with( expected_secret_name ) @patch.object(ClientCredentialsManager, "_get") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_is_valid_returns_true( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock__get: MagicMock, expected_secret_name: str, expected_client_credentials: ClientCredentials, client_credentials_manager: ClientCredentialsManager, ) -> None: """Test is_valid returns True when _get succeeds.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock__get.return_value = expected_client_credentials assert client_credentials_manager.is_valid() @patch.object(ClientCredentialsManager, "_get") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_is_valid_returns_false( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock__get: MagicMock, expected_secret_name: str, client_credentials_manager: ClientCredentialsManager, ) -> None: """Test is_valid returns False when ClientCredentialsInvalidFormat is raised.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock__get.side_effect = ClientCredentialsInvalidFormat assert not client_credentials_manager.is_valid() @patch.object(ClientCredentialsManager, "_get") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_is_valid_raises_exceptions( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock__get: MagicMock, expected_secret_name: str, client_credentials_manager: ClientCredentialsManager, ) -> None: """Test is_valid raises on other Exceptions.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock__get.side_effect = ClientCredentialsManagerException with pytest.raises(Exception): # noqa: B017 client_credentials_manager.is_valid() @patch.object(ClientCredentials, "model_validate_json") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_get( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock_model_validate_json: MagicMock, expected_secret_name: str, expected_client_credentials: ClientCredentials, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test get happy path.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock_assume_role_secrets_manager.get_secret_string.return_value = "valid_string" mock_model_validate_json.return_value = expected_client_credentials actual = client_credentials_manager._get() assert actual == expected_client_credentials mock_exists.assert_called_once() mock_assume_role_secrets_manager.get_secret_string.assert_called_once_with( expected_secret_name ) @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_get_secret_does_not_exist( mock_get_secret_name: MagicMock, mock_exists: MagicMock, expected_secret_name: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test _get when secret does not exist.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = False with pytest.raises( ClientCredentialsDoNotExist, match=f"{expected_secret_name} does not exist", ): client_credentials_manager._get() mock_exists.assert_called_once() mock_assume_role_secrets_manager.get_secret_string.assert_not_called() # noqa: E501 @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_get_invalid_format( mock_get_secret_name: MagicMock, mock_exists: MagicMock, expected_secret_name: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test _get when ValidationError.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock_assume_role_secrets_manager.get_secret_string.return_value = "invalid" with pytest.raises( ClientCredentialsInvalidFormat, match=f"{expected_secret_name} could not be parsed", ): client_credentials_manager._get() mock_exists.assert_called_once() mock_assume_role_secrets_manager.get_secret_string.assert_called_once_with( expected_secret_name ) @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_get_secret_scheduled_for_deletion( mock_get_secret_name: MagicMock, mock_exists: MagicMock, expected_secret_name: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test _get when SecretScheduledForDeletionError is raised.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock_assume_role_secrets_manager.get_secret_string.side_effect = ( SecretScheduledForDeletionError ) with pytest.raises( ClientCredentialsSecretScheduledForDeletion, match=f"Secret name, {expected_secret_name}, is scheduled for deletion.", ): client_credentials_manager._get() mock_exists.assert_called_once() mock_assume_role_secrets_manager.get_secret_string.assert_called_once_with( expected_secret_name ) @patch("m2mconfig.connectors.client_credentials_manager.get_auth0_audience") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_set( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock_get_auth0_audience: MagicMock, mock_environment: str, expected_secret_name: str, expected_client_id: str, expected_client_secret: str, expected_audience: str, expected_client_credentials_str: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test set ClientCredentials happy path.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock_assume_role_secrets_manager.save_secret_string.return_value = True mock_get_auth0_audience.return_value = expected_audience client_credentials_manager.set( expected_client_id, expected_client_secret, ) mock_get_secret_name.assert_called_once() mock_exists.assert_called_once() mock_get_auth0_audience.assert_called_once_with(mock_environment) mock_assume_role_secrets_manager.save_secret_string.assert_called_once_with( expected_secret_name, expected_client_credentials_str, ) @patch("m2mconfig.connectors.client_credentials_manager.get_auth0_audience") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_set_secret_does_not_exist( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock_get_auth0_audience: MagicMock, expected_secret_name: str, expected_client_id: str, expected_client_secret: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test set ClientCredentials raises when secret does not exist.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = False with pytest.raises( ClientCredentialsDoNotExist, match=f"{expected_secret_name} does not exist", ): client_credentials_manager.set( expected_client_id, expected_client_secret, ) mock_get_secret_name.assert_called_once() mock_exists.assert_called_once() mock_get_auth0_audience.assert_not_called() mock_assume_role_secrets_manager.save_secret_string.assert_not_called() @patch("m2mconfig.connectors.client_credentials_manager.get_auth0_audience") @patch.object(ClientCredentialsManager, "exists") @patch.object(ClientCredentialsManager, "_get_secret_name") def test_set_not_saved( mock_get_secret_name: MagicMock, mock_exists: MagicMock, mock_get_auth0_audience: MagicMock, mock_environment: str, expected_secret_name: str, expected_client_id: str, expected_client_secret: str, expected_audience: str, expected_client_credentials_str: str, client_credentials_manager: ClientCredentialsManager, mock_assume_role_secrets_manager: MagicMock, ) -> None: """Test set ClientCredentials when save_secret_string returns False.""" mock_get_secret_name.return_value = expected_secret_name mock_exists.return_value = True mock_assume_role_secrets_manager.save_secret_string.return_value = False mock_get_auth0_audience.return_value = expected_audience with pytest.raises(ClientCredentialsManagerException): # noqa client_credentials_manager.set( expected_client_id, expected_client_secret, ) mock_get_secret_name.assert_called_once() mock_exists.assert_called_once() mock_get_auth0_audience.assert_called_once_with(mock_environment) mock_assume_role_secrets_manager.save_secret_string.assert_called_once_with( expected_secret_name, expected_client_credentials_str, )