diff --git a/tests/providers/aws/services/neptune/neptune_cluster_uses_public_subnet/neptune_cluster_uses_public_subnet_test.py b/tests/providers/aws/services/neptune/neptune_cluster_uses_public_subnet/neptune_cluster_uses_public_subnet_test.py index c7ccbf7c..01cf053d 100644 --- a/tests/providers/aws/services/neptune/neptune_cluster_uses_public_subnet/neptune_cluster_uses_public_subnet_test.py +++ b/tests/providers/aws/services/neptune/neptune_cluster_uses_public_subnet/neptune_cluster_uses_public_subnet_test.py @@ -1,16 +1,18 @@ from unittest import mock -from boto3 import client, session +from boto3 import client from mock import MagicMock, patch from moto import mock_ec2, mock_neptune -from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info from prowler.providers.aws.services.neptune.neptune_service import Neptune from prowler.providers.aws.services.vpc.vpc_service import VpcSubnet -from prowler.providers.common.models import Audit_Metadata +from tests.providers.aws.audit_info_utils import ( + AWS_REGION_US_EAST_1, + AWS_REGION_US_EAST_1_AZA, + AWS_REGION_US_EAST_1_AZB, + set_mocked_aws_audit_info, +) from tests.providers.aws.services.neptune.neptune_service_test import ( - AWS_REGION_AZ1, - AWS_REGION_AZ2, NEPTUNE_CLUSTER_NAME, NEPTUNE_CLUSTER_TAGS, NEPTUNE_ENGINE, @@ -19,46 +21,12 @@ from tests.providers.aws.services.neptune.neptune_service_test import ( mock_make_api_call, ) -AWS_ACCOUNT_NUMBER = "123456789012" -AWS_ACCOUNT_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" -AWS_REGION = "us-east-1" - VPC_ID = "vpc-12345678901234567" # Patch every AWS call using Boto3 @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) class Test_neptune_cluster_uses_public_subnet: - def set_mocked_audit_info(self): - audit_info = AWS_Audit_Info( - session_config=None, - original_session=None, - audit_session=session.Session( - profile_name=None, - botocore_session=None, - ), - audited_account=AWS_ACCOUNT_NUMBER, - audited_account_arn=AWS_ACCOUNT_ARN, - audited_user_id=None, - audited_partition="aws", - audited_identity_arn=None, - profile=None, - profile_region=None, - credentials=None, - assumed_role_info=None, - audited_regions=[AWS_REGION], - organizations_metadata=None, - audit_resources=None, - mfa_enabled=False, - audit_metadata=Audit_Metadata( - services_scanned=0, - expected_checks=[], - completed_checks=0, - audit_progress=0, - ), - ) - return audit_info - @mock_neptune @mock_ec2 def test_neptune_no_clusters(self): @@ -66,7 +34,7 @@ class Test_neptune_cluster_uses_public_subnet: vpc_client = MagicMock vpc_client.vpc_subnets = {} - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) with mock.patch( "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info", @@ -98,10 +66,10 @@ class Test_neptune_cluster_uses_public_subnet: default=False, vpc_id=VPC_ID, cidr_block="192.168.0.0/24", - availability_zone=AWS_REGION_AZ1, + availability_zone=AWS_REGION_US_EAST_1_AZA, public=False, nat_gateway=False, - region=AWS_REGION, + region=AWS_REGION_US_EAST_1, tags=[], mapPublicIpOnLaunch=False, ) @@ -112,19 +80,19 @@ class Test_neptune_cluster_uses_public_subnet: default=False, vpc_id=VPC_ID, cidr_block="192.168.0.1/24", - availability_zone=AWS_REGION_AZ2, + availability_zone=AWS_REGION_US_EAST_1_AZB, public=False, nat_gateway=False, - region=AWS_REGION, + region=AWS_REGION_US_EAST_1, tags=[], mapPublicIpOnLaunch=False, ) # Neptune client - neptune_client = client("neptune", region_name=AWS_REGION) + neptune_client = client("neptune", region_name=AWS_REGION_US_EAST_1) # Create Neptune Cluster cluster = neptune_client.create_db_cluster( - AvailabilityZones=[AWS_REGION_AZ1, AWS_REGION_AZ2], + AvailabilityZones=[AWS_REGION_US_EAST_1_AZA, AWS_REGION_US_EAST_1_AZB], BackupRetentionPeriod=1, CopyTagsToSnapshot=True, Engine=NEPTUNE_ENGINE, @@ -139,7 +107,7 @@ class Test_neptune_cluster_uses_public_subnet: cluster_arn = cluster["DBClusterArn"] cluster_id = cluster["DbClusterResourceId"] - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) with mock.patch( "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info", new=audit_info, @@ -162,7 +130,7 @@ class Test_neptune_cluster_uses_public_subnet: result[0].status_extended == f"Cluster {cluster_id} is not using public subnets." ) - assert result[0].region == AWS_REGION + assert result[0].region == AWS_REGION_US_EAST_1 assert result[0].resource_id == cluster_id assert result[0].resource_arn == cluster_arn assert result[0].resource_tags == NEPTUNE_CLUSTER_TAGS @@ -179,10 +147,10 @@ class Test_neptune_cluster_uses_public_subnet: default=False, vpc_id=VPC_ID, cidr_block="192.168.0.0/24", - availability_zone=AWS_REGION_AZ1, + availability_zone=AWS_REGION_US_EAST_1_AZA, public=True, nat_gateway=False, - region=AWS_REGION, + region=AWS_REGION_US_EAST_1, tags=[], mapPublicIpOnLaunch=False, ) @@ -193,19 +161,19 @@ class Test_neptune_cluster_uses_public_subnet: default=False, vpc_id=VPC_ID, cidr_block="192.168.0.1/24", - availability_zone=AWS_REGION_AZ2, + availability_zone=AWS_REGION_US_EAST_1_AZB, public=True, nat_gateway=False, - region=AWS_REGION, + region=AWS_REGION_US_EAST_1, tags=[], mapPublicIpOnLaunch=False, ) # Neptune client - neptune_client = client("neptune", region_name=AWS_REGION) + neptune_client = client("neptune", region_name=AWS_REGION_US_EAST_1) # Create Neptune Cluster cluster = neptune_client.create_db_cluster( - AvailabilityZones=[AWS_REGION_AZ1, AWS_REGION_AZ2], + AvailabilityZones=[AWS_REGION_US_EAST_1_AZA, AWS_REGION_US_EAST_1_AZB], BackupRetentionPeriod=1, CopyTagsToSnapshot=True, Engine=NEPTUNE_ENGINE, @@ -220,7 +188,7 @@ class Test_neptune_cluster_uses_public_subnet: cluster_arn = cluster["DBClusterArn"] cluster_id = cluster["DbClusterResourceId"] - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) with mock.patch( "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info", new=audit_info, @@ -243,7 +211,7 @@ class Test_neptune_cluster_uses_public_subnet: result[0].status_extended == f"Cluster {cluster_id} is using subnet-1, subnet-2 public subnets." ) - assert result[0].region == AWS_REGION + assert result[0].region == AWS_REGION_US_EAST_1 assert result[0].resource_id == cluster_id assert result[0].resource_arn == cluster_arn assert result[0].resource_tags == NEPTUNE_CLUSTER_TAGS diff --git a/tests/providers/aws/services/neptune/neptune_service_test.py b/tests/providers/aws/services/neptune/neptune_service_test.py index 33a9e92e..4069e778 100644 --- a/tests/providers/aws/services/neptune/neptune_service_test.py +++ b/tests/providers/aws/services/neptune/neptune_service_test.py @@ -1,18 +1,16 @@ import botocore -from boto3 import client, session +from boto3 import client from mock import patch from moto import mock_neptune -from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info from prowler.providers.aws.services.neptune.neptune_service import Cluster, Neptune -from prowler.providers.common.models import Audit_Metadata - -AWS_ACCOUNT_NUMBER = "123456789012" -AWS_ACCOUNT_ARN = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" - -AWS_REGION = "us-east-1" -AWS_REGION_AZ1 = "us-east-1a" -AWS_REGION_AZ2 = "us-east-b" +from tests.providers.aws.audit_info_utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + AWS_REGION_US_EAST_1_AZA, + AWS_REGION_US_EAST_1_AZB, + set_mocked_aws_audit_info, +) SUBNET_GROUP_NAME = "default" SUBNET_1 = "subnet-1" @@ -48,16 +46,20 @@ def mock_make_api_call(self, operation_name, kwargs): "Subnets": [ { "SubnetIdentifier": "subnet-1", - "SubnetAvailabilityZone": {"Name": AWS_REGION_AZ1}, + "SubnetAvailabilityZone": { + "Name": AWS_REGION_US_EAST_1_AZA + }, "SubnetStatus": "Active", }, { "SubnetIdentifier": "subnet-2", - "SubnetAvailabilityZone": {"Name": AWS_REGION_AZ2}, + "SubnetAvailabilityZone": { + "Name": AWS_REGION_US_EAST_1_AZB + }, "SubnetStatus": "Active", }, ], - "DBSubnetGroupArn": f"arn:aws:rds:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:subgrp:{SUBNET_GROUP_NAME}", + "DBSubnetGroupArn": f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:subgrp:{SUBNET_GROUP_NAME}", } ] } @@ -68,9 +70,11 @@ def mock_make_api_call(self, operation_name, kwargs): def mock_generate_regional_clients(service, audit_info, _): - regional_client = audit_info.audit_session.client(service, region_name=AWS_REGION) - regional_client.region = AWS_REGION - return {AWS_REGION: regional_client} + regional_client = audit_info.audit_session.client( + service, region_name=AWS_REGION_US_EAST_1 + ) + regional_client.region = AWS_REGION_US_EAST_1 + return {AWS_REGION_US_EAST_1: regional_client} @patch( @@ -80,62 +84,32 @@ def mock_generate_regional_clients(service, audit_info, _): # Patch every AWS call using Boto3 @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) class Test_Neptune_Service: - # Mocked Audit Info - def set_mocked_audit_info(self): - audit_info = AWS_Audit_Info( - session_config=None, - original_session=None, - audit_session=session.Session( - profile_name=None, - botocore_session=None, - ), - audited_account=AWS_ACCOUNT_NUMBER, - audited_account_arn=AWS_ACCOUNT_ARN, - audited_user_id=None, - audited_partition="aws", - audited_identity_arn=None, - profile=None, - profile_region=None, - credentials=None, - assumed_role_info=None, - audited_regions=None, - organizations_metadata=None, - audit_resources=None, - mfa_enabled=False, - audit_metadata=Audit_Metadata( - services_scanned=0, - expected_checks=[], - completed_checks=0, - audit_progress=0, - ), - ) - return audit_info # Test Neptune Service @mock_neptune def test_service(self): - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) neptune = Neptune(audit_info) assert neptune.service == "neptune" # Test Neptune Client] @mock_neptune def test_client(self): - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) neptune = Neptune(audit_info) assert neptune.client.__class__.__name__ == "Neptune" # Test Neptune Session @mock_neptune def test__get_session__(self): - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) neptune = Neptune(audit_info) assert neptune.session.__class__.__name__ == "Session" # Test Neptune Session @mock_neptune def test_audited_account(self): - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) neptune = Neptune(audit_info) assert neptune.audited_account == AWS_ACCOUNT_NUMBER @@ -143,10 +117,10 @@ class Test_Neptune_Service: @mock_neptune def test_describe_db_clusters(self): # Neptune client - neptune_client = client("neptune", region_name=AWS_REGION) + neptune_client = client("neptune", region_name=AWS_REGION_US_EAST_1) # Create Neptune Cluster cluster = neptune_client.create_db_cluster( - AvailabilityZones=[AWS_REGION_AZ1, AWS_REGION_AZ2], + AvailabilityZones=[AWS_REGION_US_EAST_1_AZA, AWS_REGION_US_EAST_1_AZB], BackupRetentionPeriod=1, CopyTagsToSnapshot=True, Engine=NEPTUNE_ENGINE, @@ -161,7 +135,7 @@ class Test_Neptune_Service: cluster_arn = cluster["DBClusterArn"] cluster_id = cluster["DbClusterResourceId"] - audit_info = self.set_mocked_audit_info() + audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1]) neptune = Neptune(audit_info) assert len(neptune.clusters) == 1 @@ -170,7 +144,7 @@ class Test_Neptune_Service: arn=cluster_arn, name=NEPTUNE_CLUSTER_NAME, id=cluster_id, - region=AWS_REGION, + region=AWS_REGION_US_EAST_1, db_subnet_group_id=SUBNET_GROUP_NAME, subnets=[SUBNET_1, SUBNET_2], tags=NEPTUNE_CLUSTER_TAGS,