mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-09 21:14:22 +00:00
feat(DMS): Add Database Migration Service (DMS) (#4249)
This commit is contained in:
sansns-aws
+100
@@ -0,0 +1,100 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.aws.services.dms.dms_service import RepInstance
|
||||
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
|
||||
|
||||
DMS_INSTANCE_NAME = "rep-instance"
|
||||
DMS_INSTANCE_ARN = (
|
||||
f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:rep:{DMS_INSTANCE_NAME}"
|
||||
)
|
||||
KMS_KEY_ID = f"arn:aws:kms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:key/abcdabcd-1234-abcd-1234-abcdabcdabcd"
|
||||
|
||||
|
||||
class Test_dms_instance_minor_version_upgrade_enabled:
|
||||
def test_dms_no_instances(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_minor_version_upgrade_enabled.dms_instance_minor_version_upgrade_enabled import (
|
||||
dms_instance_minor_version_upgrade_enabled,
|
||||
)
|
||||
|
||||
check = dms_instance_minor_version_upgrade_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_dms_minor_version_upgrade_not_enabled(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=False,
|
||||
multi_az=True,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_minor_version_upgrade_enabled.dms_instance_minor_version_upgrade_enabled import (
|
||||
dms_instance_minor_version_upgrade_enabled,
|
||||
)
|
||||
|
||||
check = dms_instance_minor_version_upgrade_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} does not have auto minor version upgrade enabled."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
|
||||
def test_dms_instance_minor_version_upgrade_enabled(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=True,
|
||||
multi_az=True,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_minor_version_upgrade_enabled.dms_instance_minor_version_upgrade_enabled import (
|
||||
dms_instance_minor_version_upgrade_enabled,
|
||||
)
|
||||
|
||||
check = dms_instance_minor_version_upgrade_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} has auto minor version upgrade enabled."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
@@ -0,0 +1,100 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.aws.services.dms.dms_service import RepInstance
|
||||
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1
|
||||
|
||||
DMS_INSTANCE_NAME = "rep-instance"
|
||||
DMS_INSTANCE_ARN = (
|
||||
f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:rep:{DMS_INSTANCE_NAME}"
|
||||
)
|
||||
KMS_KEY_ID = f"arn:aws:kms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:key/abcdabcd-1234-abcd-1234-abcdabcdabcd"
|
||||
|
||||
|
||||
class Test_dms_instance_multi_az:
|
||||
def test_dms_no_instances(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_multi_az.dms_instance_multi_az import (
|
||||
dms_instance_multi_az,
|
||||
)
|
||||
|
||||
check = dms_instance_multi_az()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_dms_instance_multi_az_not_enabled(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=False,
|
||||
multi_az=False,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_multi_az.dms_instance_multi_az import (
|
||||
dms_instance_multi_az,
|
||||
)
|
||||
|
||||
check = dms_instance_multi_az()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} does not have multi az enabled."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
|
||||
def test_dms_instance_multi_az_enabled(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=True,
|
||||
multi_az=True,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_multi_az.dms_instance_multi_az import (
|
||||
dms_instance_multi_az,
|
||||
)
|
||||
|
||||
check = dms_instance_multi_az()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} has multi az enabled."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
+272
@@ -0,0 +1,272 @@
|
||||
from unittest import mock
|
||||
|
||||
from boto3 import client
|
||||
from moto import mock_aws
|
||||
|
||||
from prowler.providers.aws.services.dms.dms_service import RepInstance
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
DMS_INSTANCE_NAME = "rep-instance"
|
||||
DMS_INSTANCE_ARN = (
|
||||
f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:rep:{DMS_INSTANCE_NAME}"
|
||||
)
|
||||
KMS_KEY_ID = f"arn:aws:kms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:key/abcdabcd-1234-abcd-1234-abcdabcdabcd"
|
||||
|
||||
|
||||
class Test_dms_instance_no_public_access:
|
||||
def test_dms_no_instances(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.rds.rds_instance_no_public_access.rds_instance_no_public_access.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
|
||||
dms_instance_no_public_access,
|
||||
)
|
||||
|
||||
check = dms_instance_no_public_access()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_dms_private(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=False,
|
||||
security_groups=[],
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=False,
|
||||
multi_az=False,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
|
||||
dms_instance_no_public_access,
|
||||
)
|
||||
|
||||
check = dms_instance_no_public_access()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} is not publicly accessible."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
|
||||
@mock_aws
|
||||
def test_dms_public(self):
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
security_groups=[],
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=False,
|
||||
multi_az=False,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
):
|
||||
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
|
||||
dms_instance_no_public_access,
|
||||
)
|
||||
|
||||
check = dms_instance_no_public_access()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} is set as publicly accessible, but is not publicly exposed."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
|
||||
@mock_aws
|
||||
def test_dms_public_with_public_sg(self):
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
|
||||
"SecurityGroups"
|
||||
][0]
|
||||
default_sg_id = default_sg["GroupId"]
|
||||
ec2_client.authorize_security_group_ingress(
|
||||
GroupId=default_sg_id,
|
||||
IpPermissions=[
|
||||
{
|
||||
"IpProtocol": "-1",
|
||||
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
|
||||
}
|
||||
],
|
||||
)
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
security_groups=[default_sg_id],
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=False,
|
||||
multi_az=False,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
aws_provider.audit_metadata.expected_checks = [
|
||||
"ec2_securitygroup_allow_ingress_from_internet_to_any_port"
|
||||
]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
|
||||
dms_instance_no_public_access,
|
||||
)
|
||||
|
||||
check = dms_instance_no_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} is set as publicly accessible and security group default ({default_sg_id}) is open to the Internet."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
|
||||
@mock_aws
|
||||
def test_dms_public_with_filtered_sg(self):
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
|
||||
"SecurityGroups"
|
||||
][0]
|
||||
default_sg_id = default_sg["GroupId"]
|
||||
ec2_client.authorize_security_group_ingress(
|
||||
GroupId=default_sg_id,
|
||||
IpPermissions=[
|
||||
{
|
||||
"IpProtocol": "-1",
|
||||
"IpRanges": [{"CidrIp": "123.123.123.123/32"}],
|
||||
}
|
||||
],
|
||||
)
|
||||
dms_client = mock.MagicMock
|
||||
dms_client.instances = []
|
||||
dms_client.instances.append(
|
||||
RepInstance(
|
||||
id=DMS_INSTANCE_NAME,
|
||||
arn=DMS_INSTANCE_ARN,
|
||||
status="available",
|
||||
public=True,
|
||||
security_groups=[default_sg_id],
|
||||
kms_key=KMS_KEY_ID,
|
||||
auto_minor_version_upgrade=False,
|
||||
multi_az=False,
|
||||
region=AWS_REGION_US_EAST_1,
|
||||
)
|
||||
)
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
aws_provider.audit_metadata.expected_checks = [
|
||||
"ec2_securitygroup_allow_ingress_from_internet_to_any_port"
|
||||
]
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_service.DMS",
|
||||
new=dms_client,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dms.dms_instance_no_public_access.dms_instance_no_public_access import (
|
||||
dms_instance_no_public_access,
|
||||
)
|
||||
|
||||
check = dms_instance_no_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"DMS Replication Instance {DMS_INSTANCE_NAME} is set as publicly accessible but filtered with security groups."
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_id == DMS_INSTANCE_NAME
|
||||
assert result[0].resource_arn == DMS_INSTANCE_ARN
|
||||
@@ -0,0 +1,86 @@
|
||||
import botocore
|
||||
from mock import patch
|
||||
|
||||
from prowler.providers.aws.services.dms.dms_service import DMS
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
DMS_INSTANCE_NAME = "rep-instance"
|
||||
DMS_INSTANCE_ARN = (
|
||||
f"arn:aws:dms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:rep:{DMS_INSTANCE_NAME}"
|
||||
)
|
||||
KMS_KEY_ID = f"arn:aws:kms:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:key/abcdabcd-1234-abcd-1234-abcdabcdabcd"
|
||||
|
||||
# Mocking Access Analyzer Calls
|
||||
make_api_call = botocore.client.BaseClient._make_api_call
|
||||
|
||||
|
||||
def mock_make_api_call(self, operation_name, kwargs):
|
||||
if operation_name == "DescribeReplicationInstances":
|
||||
return {
|
||||
"ReplicationInstances": [
|
||||
{
|
||||
"ReplicationInstanceIdentifier": DMS_INSTANCE_NAME,
|
||||
"ReplicationInstanceStatus": "available",
|
||||
"AutoMinorVersionUpgrade": True,
|
||||
"PubliclyAccessible": True,
|
||||
"ReplicationInstanceArn": DMS_INSTANCE_ARN,
|
||||
"MultiAZ": True,
|
||||
"VpcSecurityGroups": [],
|
||||
"KmsKeyId": KMS_KEY_ID,
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
return make_api_call(self, operation_name, kwargs)
|
||||
|
||||
|
||||
def mock_generate_regional_clients(provider, service):
|
||||
regional_client = provider._session.current_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(
|
||||
"prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
|
||||
new=mock_generate_regional_clients,
|
||||
)
|
||||
# Patch every AWS call using Boto3
|
||||
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
|
||||
class Test_DMS_Service:
|
||||
# Test DMS Service
|
||||
def test_service(self):
|
||||
aws_provider = set_mocked_aws_provider()
|
||||
DMS(aws_provider)
|
||||
|
||||
# Test DMS Client
|
||||
def test_client(self):
|
||||
aws_provider = set_mocked_aws_provider()
|
||||
dms = DMS(aws_provider)
|
||||
assert dms.client.__class__.__name__ == "DatabaseMigrationService"
|
||||
|
||||
# Test DMS Account
|
||||
def test_audited_account(self):
|
||||
aws_provider = set_mocked_aws_provider()
|
||||
dms = DMS(aws_provider)
|
||||
assert dms.audited_account == AWS_ACCOUNT_NUMBER
|
||||
|
||||
# Test DMS Replication Instances
|
||||
def test_describe_rep_instances(self):
|
||||
aws_provider = set_mocked_aws_provider()
|
||||
dms = DMS(aws_provider)
|
||||
|
||||
assert len(dms.instances) == 1
|
||||
assert dms.instances[0].id == DMS_INSTANCE_NAME
|
||||
assert dms.instances[0].region == AWS_REGION_US_EAST_1
|
||||
assert dms.instances[0].status == "available"
|
||||
assert dms.instances[0].public
|
||||
assert dms.instances[0].kms_key == KMS_KEY_ID
|
||||
assert dms.instances[0].auto_minor_version_upgrade
|
||||
assert dms.instances[0].multi_az
|
||||
assert dms.instances[0].security_groups == []
|
||||
Reference in New Issue
Block a user