"""Lambda test module.""" import boto3 from moto import mock_aws import pytest import config from src import app as index @pytest.fixture() def mock_account_id(): """Return an AWS account id.""" return '123456789000' @pytest.fixture() @mock_aws def mock_rds_client(): """Return a mock RDS client.""" client = boto3.client('rds', region_name=config.AWS_DEFAULT_REGION) return client @mock_aws def test_assume_source_account_role(mock_account_id, monkeypatch): """Test assume_source_account_role function.""" aws_credentials = ['AccessKeyId', 'SecretAccessKey', 'SessionToken'] monkeypatch.setattr(index.config, 'EXTERNAL_ID', 'abc1234') result = index.assume_source_account_role(mock_account_id, 'role') for credential in aws_credentials: assert result[credential] @mock_aws def test_get_snapshots_to_delete_cluster(mock_rds_client): """Test get_snapshots_to_delete function for a cluster.""" mock_rds_client.create_db_cluster( DBClusterIdentifier='my-db', Engine='aurora-mysql', MasterUsername='dummy', MasterUserPassword='dummy123') mock_rds_client.create_db_cluster_snapshot( DBClusterIdentifier='my-db', DBClusterSnapshotIdentifier='my-snapshot-1', Tags=[ {'Key': 'execution_id', 'Value': '123'} ]) mock_rds_client.create_db_cluster_snapshot( DBClusterIdentifier='my-db', DBClusterSnapshotIdentifier='my-snapshot-2', Tags=[ {'Key': 'execution_id', 'Value': '456'} ]) event = { 'db_name': 'my-db', 'db_type': 'cluster', 'execution_id': '123' } result = index.get_snapshots_to_delete(event, mock_rds_client) assert result == {'my-snapshot-1'} @mock_aws def test_get_snapshots_to_delete_instance(mock_rds_client): """Test get_snapshots_to_delete function for an instance.""" mock_rds_client.create_db_instance( DBInstanceIdentifier='my-db', DBInstanceClass='db.t2.micro', Engine='mysql', MasterUsername='dummy', MasterUserPassword='dummy123') mock_rds_client.create_db_snapshot( DBInstanceIdentifier='my-db', DBSnapshotIdentifier='my-snapshot-1', Tags=[ {'Key': 'execution_id', 'Value': '123'} ]) mock_rds_client.create_db_snapshot( DBInstanceIdentifier='my-db', DBSnapshotIdentifier='my-snapshot-2', Tags=[ {'Key': 'execution_id', 'Value': '456'} ]) event = { 'db_name': 'my-db', 'db_type': 'standalone', 'execution_id': '123' } result = index.get_snapshots_to_delete(event, mock_rds_client) assert result == {'my-snapshot-1'}