"""Lambda test module.""" from dataclasses import dataclass import logging from typing import Any from unittest.mock import MagicMock, call, patch import pytest from freezegun import freeze_time from src import app as index from src.models import M2MToken, ClientCredentials, OAuthToken, create_dummy_m2m_token @pytest.fixture def lambda_context() -> object: @dataclass class LambdaContext: function_name: str = "test" memory_limit_in_mb: int = 128 invoked_function_arn: str = "arn:aws:lambda:eu-west-1:809313241:function:test" aws_request_id: str = "52fdfc07-2182-154f-163f-5f0f9a621d72" return LambdaContext() @pytest.fixture() def mock_client_request_token() -> str: """Reusable ClientRequestToken.""" return "some token" @pytest.fixture() def mock_secret_id() -> str: """Reusable Secret ID/ARN.""" return "qa/my-machine/AUTH0_M2M_JWT_TOKEN" class MockException(Exception): pass @patch("src.app.boto3") def test_handler_raises_when_rotation_is_not_enabled( mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler raises when rotation is not enabled.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "createSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": False, } mock_boto3.client.return_value = mock_service_client with pytest.raises(ValueError): index.handler(mock_event, lambda_context) @patch("src.app.boto3") def test_handler_raises_when_token_not_in_versions( mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler raises when token is not in VersionIdsToStages.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "createSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { "other token": ["AWSCURRENT", "AWSPENDING"], }, } mock_boto3.client.return_value = mock_service_client with pytest.raises(ValueError): index.handler(mock_event, lambda_context) @patch("src.app.boto3") def test_handler_does_nothing_when_token_is_awscurrent( mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler does nothing when token is awscurrent. When the secret to operate on already is the AWSCURRENT stage, then return None. """ mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "createSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { mock_client_request_token: ["AWSCURRENT"], }, } mock_boto3.client.return_value = mock_service_client result = index.handler(mock_event, lambda_context) assert not result @patch("src.app.boto3") def test_handler_raises_when_token_not_pending( mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler raises when token is not AWSPENDING.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "createSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { mock_client_request_token: ["OTHER STAGE"], }, } mock_boto3.client.return_value = mock_service_client with pytest.raises(ValueError): index.handler(mock_event, lambda_context) @patch("src.app.boto3") @patch("src.app.create_secret") def test_handler_calls_calls_create_secret( mock_create_secret: MagicMock, mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler calls create_secret.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "createSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { mock_client_request_token: ["AWSPENDING"], }, } mock_boto3.client.return_value = mock_service_client result = index.handler(mock_event, lambda_context) assert not result mock_create_secret.assert_called_once_with( mock_service_client, mock_secret_id, mock_client_request_token, lambda_context, ) @patch("src.app.boto3") @patch("src.app.set_secret") def test_handler_calls_calls_set_secret( mock_set_secret: MagicMock, mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler calls set_secret.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "setSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { mock_client_request_token: ["AWSPENDING"], }, } mock_boto3.client.return_value = mock_service_client result = index.handler(mock_event, lambda_context) assert not result mock_set_secret.assert_called_once_with( mock_service_client, mock_secret_id, mock_client_request_token, ) @patch("src.app.boto3") @patch("src.app.test_secret") def test_handler_calls_calls_test_secret( mock_test_secret: MagicMock, mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler calls test_secret.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "testSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { mock_client_request_token: ["AWSPENDING"], }, } mock_boto3.client.return_value = mock_service_client result = index.handler(mock_event, lambda_context) assert not result mock_test_secret.assert_called_once_with( mock_service_client, mock_secret_id, mock_client_request_token, ) @patch("src.app.boto3") @patch("src.app.finish_secret") def test_handler_calls_calls_finish_secret( mock_finish_secret: MagicMock, mock_boto3: MagicMock, mock_client_request_token: str, mock_secret_id: str, lambda_context: object, ) -> None: """Test handler calls finish_secret.""" mock_event = { "SecretId": mock_secret_id, "ClientRequestToken": mock_client_request_token, "Step": "finishSecret", } mock_service_client = MagicMock() mock_service_client.describe_secret.return_value = { "RotationEnabled": True, "VersionIdsToStages": { mock_client_request_token: ["AWSPENDING"], }, } mock_boto3.client.return_value = mock_service_client result = index.handler(mock_event, lambda_context) assert not result mock_finish_secret.assert_called_once_with( mock_service_client, mock_secret_id, mock_client_request_token, ) @freeze_time("2024-09-17 00:00:00.000000+0000") @patch("src.app.generate_auth_token") def test_create_secret_succeeds_by_exception_thrown( mock_generate_auth_token: MagicMock, mock_service_client: MagicMock, mock_valid_auth0_response: dict[str, Any], mock_client_request_token: str, mock_arn: str, lambda_context: object, ) -> None: """Test create_secret success by exception thrown.""" mock_m2m_token = M2MToken.model_validate( { "token": "some_token", "expires_at": "2024-09-17 10:00:00.000000+0000", "client_credentials_secret_arn": "some_client_id", } ).model_dump_json() mock_service_client.exceptions.InvalidRequestException = MockException mock_service_client.exceptions.ResourceNotFoundException = MockException mock_service_client.get_secret_value.side_effect = [ MockException("InvalidRequestException"), {"SecretString": mock_m2m_token}, { "SecretString": ClientCredentials.model_validate( { "audience": "some_audience", "client_id": "some_client_id", "client_secret": "some_client_secret", "grant_type": "some_grant_type", } ).model_dump_json() }, ] mock_generate_auth_token.return_value = OAuthToken.model_validate( mock_valid_auth0_response ) mock_service_client.put_secret_value.return_value = { "VersionId": "some_version_id", } index.create_secret( mock_service_client, mock_arn, mock_client_request_token, lambda_context, # type: ignore[arg-type] ) mock_service_client.get_secret_value.assert_has_calls( [ call( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ), call( SecretId=mock_arn, VersionStage="AWSCURRENT", ), call( SecretId="some_client_id", VersionStage="AWSCURRENT", ), ] ) mock_generate_auth_token.assert_called_once_with( ClientCredentials.model_validate( { "audience": "some_audience", "client_id": "some_client_id", "client_secret": "some_client_secret", "grant_type": "some_grant_type", } ) ) mock_service_client.put_secret_value.assert_called_once_with( SecretId=mock_arn, ClientRequestToken=mock_client_request_token, SecretString=mock_m2m_token, VersionStages=["AWSPENDING"], ) @patch("src.app.generate_auth_token") def test_create_secret_succeeds_for_existing_secret( mock_generate_auth_token: MagicMock, mock_service_client: MagicMock, mock_valid_auth0_response: dict[str, Any], mock_client_request_token: str, mock_arn: str, lambda_context: object, ) -> None: """Test create_secret succeeds for existing secret.""" mock_service_client.get_secret_value.side_effect = [ { "SecretString": M2MToken.model_validate( { "token": "some_token", "expires_at": "2024-09-17 10:00:00.000000+0000", "client_credentials_secret_arn": "some_client_id", } ).model_dump_json() }, ] mock_generate_auth_token.return_value = OAuthToken.model_validate( mock_valid_auth0_response ) mock_service_client.put_secret_value.return_value = { "VersionId": "some_version_id", } index.create_secret( mock_service_client, mock_arn, mock_client_request_token, lambda_context, # type: ignore[arg-type] ) mock_service_client.get_secret_value.assert_has_calls( [ call( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ), ] ) @freeze_time("2024-09-17 00:00:00.000000+0000") @patch("src.app.generate_auth_token") @patch("src.app.create_dummy_m2m_token") def test_create_secret_with_dummy_client_credentials( mock_create_dummy_m2m_token: MagicMock, mock_generate_auth_token: MagicMock, mock_service_client: MagicMock, mock_client_request_token: str, mock_arn: str, lambda_context: object, caplog: pytest.LogCaptureFixture, ) -> None: """Test create_secret uses dummy M2M token when client_credentials is 'dummy'.""" client_credentials_secret_arn = "some_client_credentials_arn" mock_dummy_token = M2MToken.model_validate( { "token": "dummy", "expires_at": "2024-09-17 10:00:00.000000+0000", "client_credentials_secret_arn": client_credentials_secret_arn, } ) mock_create_dummy_m2m_token.return_value = mock_dummy_token mock_service_client.exceptions.InvalidRequestException = MockException mock_service_client.exceptions.ResourceNotFoundException = MockException mock_service_client.get_secret_value.side_effect = [ MockException("InvalidRequestException"), { "SecretString": M2MToken.model_validate( { "token": "current_token", "expires_at": "2024-09-17 05:00:00.000000+0000", "client_credentials_secret_arn": client_credentials_secret_arn, } ).model_dump_json() }, {"SecretString": "dummy"}, ] mock_service_client.put_secret_value.return_value = { "VersionId": "some_version_id", } with caplog.at_level(logging.WARNING): index.create_secret( mock_service_client, mock_arn, mock_client_request_token, lambda_context, # type: ignore[arg-type] ) # Verify get_secret_value was called correctly mock_service_client.get_secret_value.assert_has_calls( [ call( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ), call( SecretId=mock_arn, VersionStage="AWSCURRENT", ), call( SecretId=client_credentials_secret_arn, VersionStage="AWSCURRENT", ), ] ) # Verify create_dummy_m2m_token was called mock_create_dummy_m2m_token.assert_called_once_with(client_credentials_secret_arn) # Verify generate_auth_token was NOT called mock_generate_auth_token.assert_not_called() # Verify put_secret_value was called with the dummy token mock_service_client.put_secret_value.assert_called_once_with( SecretId=mock_arn, ClientRequestToken=mock_client_request_token, SecretString=mock_dummy_token.model_dump_json(), VersionStages=["AWSPENDING"], ) # Verify logging assert ( f"Client credentials {client_credentials_secret_arn} have not been configured, setting dummy M2MToken." in caplog.text ) @patch("src.app.generate_auth_token") def test_create_secret_fails( mock_service_client: MagicMock, mock_client_request_token: str, mock_arn: str, lambda_context: object, ) -> None: """Test create_secret fails due invalid response.""" mock_service_client.exceptions.InvalidRequestException = MockException mock_service_client.exceptions.ResourceNotFoundException = MockException mock_service_client.get_secret_value.side_effect = [ MockException("InvalidRequestException"), ] with pytest.raises(index.Auth0M2MRotationError) as exc_info: index.create_secret( mock_service_client, mock_arn, mock_client_request_token, lambda_context, # type: ignore[arg-type] ) # NOTE: pytest.raises does not include the error message from exception chaining (raise from ex). assert ( "create_secret error. request_id:52fdfc07-2182-154f-163f-5f0f9a621d72. arn:some_arn" in str(exc_info.value) ) def test_set_secret_succeeds( mock_service_client: MagicMock, mock_client_request_token: str, mock_arn: str ) -> None: """Test set_secret succeeds.""" mock_service_client.get_secret_value.side_effect = [ { "SecretString": M2MToken.model_validate( { "token": "1", "expires_at": "2024-07-22 16:55:30.13455+0000", "client_credentials_secret_arn": "some_other_arn", } ).model_dump_json() }, { "SecretString": M2MToken.model_validate( { "token": "2", "expires_at": "2024-07-23 16:55:30.13455+0000", "client_credentials_secret_arn": "some_other_arn", } ).model_dump_json() }, ] index.set_secret(mock_service_client, mock_arn, mock_client_request_token) mock_service_client.get_secret_value.assert_has_calls( [ call( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ), call( SecretId=mock_arn, VersionStage="AWSCURRENT", ), ] ) def test_set_secret_assertion_error( mock_service_client: MagicMock, mock_client_request_token: str, mock_arn: str, ) -> None: """Test set_secret has an assertion error.""" mock_service_client.get_secret_value.side_effect = [ { "SecretString": M2MToken.model_validate( { "token": "1", "expires_at": "2024-07-22 16:55:30.13455+0000", "client_credentials_secret_arn": "some_other_arn", } ).model_dump_json() }, { "SecretString": M2MToken.model_validate( { "token": "bad-arn", "expires_at": "2024-07-23 16:55:30.13455+0000", "client_credentials_secret_arn": "some-non-matching-arn", } ).model_dump_json() }, ] with pytest.raises(AssertionError): index.set_secret(mock_service_client, mock_arn, mock_client_request_token) mock_service_client.get_secret_value.assert_has_calls( [ call( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ), call( SecretId=mock_arn, VersionStage="AWSCURRENT", ), ] ) @pytest.mark.parametrize( "get_token_effect, is_expired, expect_is_expired_call, expected_exception", [ pytest.param( "its a token", False, True, None, id="happy path", ), pytest.param( Exception("its not a token"), False, False, Exception("its not a token"), id="jwt token could not be decoded", ), pytest.param( "its a token", True, True, AssertionError( "testSecret: AWSPENDING secret expires_at 2024-07-22 16:55:30.134550+00:00 - should not be expired" ), id="expires_at is old", ), ], ) @patch("src.app.jwt_auth_from_environment") @patch.object(M2MToken, "is_expired") def test_test_secret( mock_is_expired: MagicMock, mock_jwt_auth_from_environment: MagicMock, get_token_effect: str | Exception, is_expired: bool, expect_is_expired_call: bool, expected_exception: Exception | None, expiration_string: str, mock_service_client: MagicMock, mock_client_request_token: str, mock_arn: str, ) -> None: """Test test_secret succeeds.""" mock_is_expired.return_value = is_expired mock_jwt_auth = MagicMock() mock_jwt_auth.get_token.side_effect = [(get_token_effect)] mock_jwt_auth_from_environment.return_value = mock_jwt_auth mock_service_client.get_secret_value.return_value = { "SecretString": f'{{"token": "the jwt bearer token", "expires_at": "{expiration_string}", "client_credentials_secret_arn": "some_other_arn"}}' } if expected_exception: with pytest.raises(Exception) as exc_info: index.test_secret( mock_service_client, mock_arn, mock_client_request_token, ) assert str(exc_info.value) == str(expected_exception) else: index.test_secret( mock_service_client, mock_arn, mock_client_request_token, ) mock_service_client.get_secret_value.assert_called_once_with( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ) mock_jwt_auth.get_token.assert_called_once_with("the jwt bearer token") if expect_is_expired_call: mock_is_expired.assert_called_once() else: mock_is_expired.assert_not_called() @patch("src.app.jwt_auth_from_environment") @patch.object(M2MToken, "is_expired") def test_test_secret_handles_dummy( mock_is_expired: MagicMock, mock_jwt_auth_from_environment: MagicMock, mock_service_client: MagicMock, mock_client_request_token: str, mock_arn: str, caplog: pytest.LogCaptureFixture, ) -> None: """Test test_secret succeeds when the pending JWT is dummy.""" mock_jwt_auth = MagicMock() mock_jwt_auth.get_token = MagicMock() mock_jwt_auth_from_environment.return_value = mock_jwt_auth mock_service_client.get_secret_value.return_value = { "SecretString": f"{create_dummy_m2m_token('some client arn').model_dump_json()}" } index.test_secret( mock_service_client, mock_arn, mock_client_request_token, ) mock_service_client.get_secret_value.assert_called_once_with( SecretId=mock_arn, VersionId=mock_client_request_token, VersionStage="AWSPENDING", ) mock_is_expired.assert_not_called() mock_jwt_auth.get_token.assert_not_called() assert f"testSecret: AWSPENDING secret is still dummy for {mock_arn}" in caplog.text def test_finish_secret( mock_service_client: MagicMock, mock_client_request_token: str, ) -> None: """Test finish_secret.""" mock_service_client.describe_secret.return_value = { "VersionIdsToStages": { mock_client_request_token: ["AWSCURRENT"], }, } index.finish_secret(mock_service_client, "arn", "token") mock_service_client.describe_secret.assert_called_once_with(SecretId="arn") mock_service_client.update_secret_version_stage.assert_called_once_with( SecretId="arn", VersionStage="AWSCURRENT", MoveToVersionId="token", RemoveFromVersionId="some token", ) def test_finish_secret_fails_already_marked_as_current( mock_service_client: MagicMock, ) -> None: """Test finish_secret fails.""" mock_service_client.describe_secret.return_value = { "VersionIdsToStages": { "awscurrent_token": ["AWSCURRENT"], }, } index.finish_secret(mock_service_client, "arn", "awscurrent_token") mock_service_client.describe_secret.assert_called_once_with(SecretId="arn") mock_service_client.update_secret_version_stage.assert_not_called()