chore(aws): cleanup aws test cases (#4049)

prowler/providers/aws/services/sns/sns_topics_kms_encryption_at_rest_enabled/sns_topics_kms_encryption_at_rest_enabled.py

@@ -12,10 +12,10 @@ class sns_topics_kms_encryption_at_rest_enabled(Check):
             report.resource_arn = topic.arn
             report.resource_tags = topic.tags
             report.status = "PASS"
-            report.status_extended = f"SNS topic {topic.arn} is encrypted."
+            report.status_extended = f"SNS topic {topic.name} is encrypted."
             if not topic.kms_master_key_id:
                 report.status = "FAIL"
-                report.status_extended = f"SNS topic {topic.arn} is not encrypted."
+                report.status_extended = f"SNS topic {topic.name} is not encrypted."
 
             findings.append(report)
 

prowler/providers/aws/services/workspaces/workspaces_volume_encryption_enabled/workspaces_volume_encryption_enabled.py

@@ -14,7 +14,7 @@ class workspaces_volume_encryption_enabled(Check):
             report.resource_arn = workspace.arn
             report.resource_tags = workspace.tags
             report.status = "PASS"
-            report.status_extended = f"WorkSpaces workspace {workspace.id} without root or user unencrypted volumes."
+            report.status_extended = f"WorkSpaces workspace {workspace.id} root and user volumes are encrypted."
             if not workspace.user_volume_encryption_enabled:
                 report.status = "FAIL"
                 report.status_extended = f"WorkSpaces workspace {workspace.id} with user unencrypted volumes."

tests/providers/aws/services/sagemaker/sagemaker_models_network_isolation_enabled/sagemaker_models_network_isolation_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,10 @@ class Test_sagemaker_models_network_isolation_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has network isolation enabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has network isolation enabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn
 
@@ -76,6 +78,9 @@ class Test_sagemaker_models_network_isolation_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has network isolation disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has network isolation disabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn

tests/providers/aws/services/sagemaker/sagemaker_models_vpc_settings_configured/sagemaker_models_vpc_settings_configured_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,10 @@ class Test_sagemaker_models_vpc_settings_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has VPC settings enabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has VPC settings enabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn
 
@@ -75,6 +77,9 @@ class Test_sagemaker_models_vpc_settings_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has VPC settings disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has VPC settings disabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn

tests/providers/aws/services/sagemaker/sagemaker_notebook_instance_encryption_enabled/sagemaker_notebook_instance_encryption_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,10 @@ class Test_sagemaker_notebook_instance_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has data encryption enabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has data encryption enabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn
 
@@ -75,6 +77,9 @@ class Test_sagemaker_notebook_instance_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has data encryption disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has data encryption disabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn

tests/providers/aws/services/sagemaker/sagemaker_notebook_instance_root_access_disabled/sagemaker_notebook_instance_root_access_disabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 
 from prowler.providers.aws.services.sagemaker.sagemaker_service import NotebookInstance
@@ -47,7 +46,10 @@ class Test_sagemaker_notebook_instance_root_access_disabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has root access disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has root access disabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn
 
@@ -74,6 +76,9 @@ class Test_sagemaker_notebook_instance_root_access_disabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has root access enabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has root access enabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn

tests/providers/aws/services/sagemaker/sagemaker_notebook_instance_vpc_settings_configured/sagemaker_notebook_instance_vpc_settings_configured_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,10 @@ class Test_sagemaker_notebook_instance_vpc_settings_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("is in a VPC", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} is in a VPC."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn
 
@@ -76,6 +78,9 @@ class Test_sagemaker_notebook_instance_vpc_settings_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has VPC settings disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has VPC settings disabled."
+            )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn

tests/providers/aws/services/sagemaker/sagemaker_notebook_instance_without_direct_internet_access_configured/sagemaker_notebook_instance_without_direct_internet_access_configured_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 
 from prowler.providers.aws.services.sagemaker.sagemaker_service import NotebookInstance
@@ -51,8 +50,9 @@ class Test_sagemaker_notebook_instance_without_direct_internet_access_configured
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "has direct internet access disabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has direct internet access disabled."
             )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn
@@ -82,8 +82,9 @@ class Test_sagemaker_notebook_instance_without_direct_internet_access_configured
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "has direct internet access enabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"Sagemaker notebook instance {test_notebook_instance} has direct internet access enabled."
             )
             assert result[0].resource_id == test_notebook_instance
             assert result[0].resource_arn == notebook_instance_arn

tests/providers/aws/services/sagemaker/sagemaker_training_jobs_intercontainer_encryption_enabled/sagemaker_training_jobs_intercontainer_encryption_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 
 from prowler.providers.aws.services.sagemaker.sagemaker_service import TrainingJob
@@ -47,8 +46,9 @@ class Test_sagemaker_training_jobs_intercontainer_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "has intercontainer encryption enabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has intercontainer encryption enabled."
             )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn
@@ -75,8 +75,9 @@ class Test_sagemaker_training_jobs_intercontainer_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "has intercontainer encryption disabled", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has intercontainer encryption disabled."
             )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn

tests/providers/aws/services/sagemaker/sagemaker_training_jobs_network_isolation_enabled/sagemaker_training_jobs_network_isolation_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,10 @@ class Test_sagemaker_training_jobs_network_isolation_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has network isolation enabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has network isolation enabled."
+            )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn
 
@@ -75,6 +77,9 @@ class Test_sagemaker_training_jobs_network_isolation_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has network isolation disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has network isolation disabled."
+            )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn

tests/providers/aws/services/sagemaker/sagemaker_training_jobs_volume_and_output_encryption_enabled/sagemaker_training_jobs_volume_and_output_encryption_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,10 @@ class Test_sagemaker_training_jobs_volume_and_output_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("has KMS encryption enabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has KMS encryption enabled."
+            )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn
 
@@ -75,6 +77,9 @@ class Test_sagemaker_training_jobs_volume_and_output_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("has KMS encryption disabled", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has KMS encryption disabled."
+            )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn

tests/providers/aws/services/sagemaker/sagemaker_training_jobs_vpc_settings_configured/sagemaker_training_jobs_vpc_settings_configured_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,9 +48,9 @@ class Test_sagemaker_training_jobs_vpc_settings_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "has VPC settings for the training job volume and output enabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has VPC settings for the training job volume and output enabled."
             )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn
@@ -78,9 +77,9 @@ class Test_sagemaker_training_jobs_vpc_settings_configured:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "has VPC settings for the training job volume and output disabled",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == f"Sagemaker training job {test_training_job} has VPC settings for the training job volume and output disabled."
             )
             assert result[0].resource_id == test_training_job
             assert result[0].resource_arn == training_job_arn

tests/providers/aws/services/sns/sns_topics_kms_encryption_at_rest_enabled/sns_topics_kms_encryption_at_rest_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -49,7 +48,7 @@ class Test_sns_topics_kms_encryption_at_rest_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("is encrypted", result[0].status_extended)
+            assert result[0].status_extended == f"SNS topic {topic_name} is encrypted."
             assert result[0].resource_id == topic_name
             assert result[0].resource_arn == topic_arn
 
@@ -71,6 +70,8 @@ class Test_sns_topics_kms_encryption_at_rest_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("is not encrypted", result[0].status_extended)
+            assert (
+                result[0].status_extended == f"SNS topic {topic_name} is not encrypted."
+            )
             assert result[0].resource_id == topic_name
             assert result[0].resource_arn == topic_arn

tests/providers/aws/services/sqs/sqs_queues_not_publicly_accessible/sqs_queues_not_publicly_accessible_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -145,7 +144,10 @@ class Test_sqs_queues_not_publicly_accessible:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("is not public", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"SQS queue {test_queue_url} is not public."
+            )
             assert result[0].resource_id == test_queue_url
             assert result[0].resource_arn == test_queue_arn
             assert result[0].resource_tags == []
@@ -175,9 +177,9 @@ class Test_sqs_queues_not_publicly_accessible:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "is public because its policy allows public access",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == f"SQS queue {test_queue_url} is public because its policy allows public access."
             )
             assert result[0].resource_id == test_queue_url
             assert result[0].resource_arn == test_queue_arn
@@ -209,9 +211,9 @@ class Test_sqs_queues_not_publicly_accessible:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "is public because its policy allows public access",
-                result[0].status_extended,
+            assert (
+                result[0].status_extended
+                == f"SQS queue {test_queue_url} is public because its policy allows public access, and the condition does not limit access to resources within the same account."
             )
             assert result[0].resource_id == test_queue_url
             assert result[0].resource_arn == test_queue_arn

tests/providers/aws/services/sqs/sqs_queues_server_side_encryption_enabled/sqs_queues_server_side_encryption_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -53,7 +52,10 @@ class Test_sqs_queues_server_side_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search("is using Server Side Encryption", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"SQS queue {test_queue_url} is using Server Side Encryption."
+            )
             assert result[0].resource_id == test_queue_url
             assert result[0].resource_arn == test_queue_arn
 
@@ -80,8 +82,9 @@ class Test_sqs_queues_server_side_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "is not using Server Side Encryption", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"SQS queue {test_queue_url} is not using Server Side Encryption."
             )
             assert result[0].resource_id == test_queue_url
             assert result[0].resource_arn == test_queue_arn

tests/providers/aws/services/wellarchitected/wellarchitected_workload_no_high_or_medium_risks/wellarchitected_workload_no_high_or_medium_risks_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -58,8 +57,9 @@ class Test_wellarchitected_workload_no_high_or_medium_risks:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "does not contain high or medium risks", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == "Well Architected workload test does not contain high or medium risks."
             )
             assert result[0].resource_id == workload_id
             assert (
@@ -99,8 +99,9 @@ class Test_wellarchitected_workload_no_high_or_medium_risks:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "does not contain high or medium risks", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == "Well Architected workload test does not contain high or medium risks."
             )
             assert result[0].resource_id == workload_id
             assert (
@@ -142,8 +143,9 @@ class Test_wellarchitected_workload_no_high_or_medium_risks:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "contains 10 high and 20 medium risks", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == "Well Architected workload test contains 10 high and 20 medium risks."
             )
             assert result[0].resource_id == workload_id
             assert (

tests/providers/aws/services/workspaces/workspaces_volume_encryption_enabled/workspaces_volume_encryption_enabled_test.py

@@ -1,4 +1,3 @@
-from re import search
 from unittest import mock
 from uuid import uuid4
 
@@ -56,8 +55,9 @@ class Test_workspaces_volume_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "PASS"
-            assert search(
-                "without root or user unencrypted volumes", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"WorkSpaces workspace {WORKSPACE_ID} root and user volumes are encrypted."
             )
             assert result[0].resource_id == WORKSPACE_ID
             assert result[0].resource_arn == WORKSPACE_ARN
@@ -91,7 +91,10 @@ class Test_workspaces_volume_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("user unencrypted volumes", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"WorkSpaces workspace {WORKSPACE_ID} with user unencrypted volumes."
+            )
             assert result[0].resource_id == WORKSPACE_ID
             assert result[0].resource_arn == WORKSPACE_ARN
             assert result[0].region == AWS_REGION_EU_WEST_1
@@ -124,7 +127,10 @@ class Test_workspaces_volume_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search("root unencrypted volumes", result[0].status_extended)
+            assert (
+                result[0].status_extended
+                == f"WorkSpaces workspace {WORKSPACE_ID} with root unencrypted volumes."
+            )
             assert result[0].resource_id == WORKSPACE_ID
             assert result[0].resource_arn == WORKSPACE_ARN
             assert result[0].region == AWS_REGION_EU_WEST_1
@@ -157,8 +163,9 @@ class Test_workspaces_volume_encryption_enabled:
             result = check.execute()
             assert len(result) == 1
             assert result[0].status == "FAIL"
-            assert search(
-                "with root and user unencrypted volumes", result[0].status_extended
+            assert (
+                result[0].status_extended
+                == f"WorkSpaces workspace {WORKSPACE_ID} with root and user unencrypted volumes."
             )
             assert result[0].resource_id == WORKSPACE_ID
             assert result[0].resource_arn == WORKSPACE_ARN