"""Lambda test module.""" import boto3 import pytest from moto import mock_rds from moto import mock_sts from src import app as index import config @pytest.fixture() def mock_account_id(): """Return an AWS account id.""" return '123456789000' @pytest.fixture() def mock_rds_cluster_name(): """Return an RDS cluster name.""" return 'prod-ows-cluster' @pytest.fixture() def mock_rds_standalone_name(): """Return an RDS standalone name.""" return 'prod-ows-standalone' @pytest.fixture() @mock_rds @mock_sts def mock_rds_client(mock_account_id, monkeypatch): """Return a mock RDS client.""" monkeypatch.setattr(index.config, 'EXTERNAL_ID', 'abc1234') credentials = index.assume_source_account_role(mock_account_id) client = boto3.client( 'rds', region_name=config.AWS_REGION, aws_access_key_id=credentials['AccessKeyId'], aws_secret_access_key=credentials['SecretAccessKey'], aws_session_token=credentials['SessionToken'] ) return client @mock_sts 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) for credential in aws_credentials: assert result[credential] @mock_rds def test_find_production_databases( mock_rds_client, mock_rds_cluster_name, mock_rds_standalone_name): """Test find_production_databases function.""" mock_rds_client.create_db_instance( AllocatedStorage=5, DBInstanceClass='db.t3.micro', DBInstanceIdentifier=mock_rds_standalone_name, Engine='aurora-mysql', MasterUserPassword='MyPassword', MasterUsername='MyUser', Tags=[ { 'Key': 'environment', 'Value': 'prod' }, ] ) mock_rds_client.create_db_cluster( AvailabilityZones=[ 'us-east-1a', ], BackupRetentionPeriod=1, DBClusterIdentifier=mock_rds_cluster_name, DatabaseName='dummy', Engine='aurora-mysql', EngineVersion='5.7.mysql_aurora.2.07.0', MasterUserPassword='dummypassword1234', MasterUsername='dummyuser', Port=3306, StorageEncrypted=True, Tags=[ { 'Key': 'environment', 'Value': 'prod' }, ] ) prod_dbs = index.find_production_databases(mock_rds_client) assert prod_dbs == { mock_rds_cluster_name: 'cluster', mock_rds_standalone_name: 'standalone' } @mock_rds def test_find_shared_databases( mock_rds_client, mock_rds_cluster_name, mock_rds_standalone_name): """Test find_production_databases function for shared environment.""" mock_rds_client.create_db_instance( AllocatedStorage=5, DBInstanceClass='db.t3.micro', DBInstanceIdentifier=mock_rds_standalone_name, Engine='aurora-mysql', MasterUserPassword='MyPassword', MasterUsername='MyUser', Tags=[ { 'Key': 'environment', 'Value': 'shared' }, ] ) mock_rds_client.create_db_cluster( AvailabilityZones=[ 'us-east-1a', ], BackupRetentionPeriod=1, DBClusterIdentifier=mock_rds_cluster_name, DatabaseName='dummy', Engine='aurora-mysql', EngineVersion='5.7.mysql_aurora.2.07.0', MasterUserPassword='dummypassword1234', MasterUsername='dummyuser', Port=3306, StorageEncrypted=True, Tags=[ { 'Key': 'environment', 'Value': 'shared' }, ] ) shared_dbs = index.find_production_databases(mock_rds_client) assert shared_dbs == { mock_rds_cluster_name: 'cluster', mock_rds_standalone_name: 'standalone' }