test(gcp): Add bigquery and half of cloudsql check tests (#4462)

tests/providers/gcp/services/bigquery/bigquery_dataset_cmk_encryption/bigquery_dataset_cmk_encryption_test.py

@@ -0,0 +1,105 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider
+
+
+class Test_bigquery_dataset_cmk_encryption:
+    def test_bigquery_no_datasets(self):
+        bigquery_client = mock.MagicMock
+        bigquery_client.datasets = []
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption.bigquery_client",
+            new=bigquery_client,
+        ):
+            from prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption import (
+                bigquery_dataset_cmk_encryption,
+            )
+
+            check = bigquery_dataset_cmk_encryption()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_one_compliant_dataset(self):
+        bigquery_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption.bigquery_client",
+            new=bigquery_client,
+        ):
+            from prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption import (
+                bigquery_dataset_cmk_encryption,
+            )
+            from prowler.providers.gcp.services.bigquery.bigquery_service import Dataset
+
+            dataset = Dataset(
+                name="test",
+                id="1234567890",
+                region="us-central1",
+                cmk_encryption=True,
+                public=False,
+                project_id=GCP_PROJECT_ID,
+            )
+            bigquery_client.project_ids = [GCP_PROJECT_ID]
+            bigquery_client.datasets = [dataset]
+
+            check = bigquery_dataset_cmk_encryption()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Dataset {dataset.name} is encrypted with Customer-Managed Keys (CMKs)."
+            )
+            assert result[0].resource_id == dataset.id
+            assert result[0].resource_name == dataset.name
+            assert result[0].project_id == dataset.project_id
+            assert result[0].location == dataset.region
+
+    def test_one_non_compliant_dataset(self):
+        bigquery_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption.bigquery_client",
+            new=bigquery_client,
+        ):
+            from prowler.providers.gcp.services.bigquery.bigquery_dataset_cmk_encryption.bigquery_dataset_cmk_encryption import (
+                bigquery_dataset_cmk_encryption,
+            )
+            from prowler.providers.gcp.services.bigquery.bigquery_service import Dataset
+
+            dataset = Dataset(
+                name="test",
+                id="1234567890",
+                region="us-central1",
+                cmk_encryption=False,
+                public=False,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            bigquery_client.project_ids = [GCP_PROJECT_ID]
+            bigquery_client.datasets = [dataset]
+
+            check = bigquery_dataset_cmk_encryption()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Dataset {dataset.name} is not encrypted with Customer-Managed Keys (CMKs)."
+            )
+            assert result[0].resource_id == dataset.id
+            assert result[0].resource_name == dataset.name
+            assert result[0].project_id == dataset.project_id
+            assert result[0].location == dataset.region

tests/providers/gcp/services/bigquery/bigquery_table_cmk_encryption/bigquery_table_cmk_encryption_test.py

@@ -0,0 +1,104 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider
+
+
+class Test_bigquery_table_cmk_encryption:
+    def test_bigquery_no_tables(self):
+        bigquery_client = mock.MagicMock
+        bigquery_client.tables = []
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption.bigquery_client",
+            new=bigquery_client,
+        ):
+            from prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption import (
+                bigquery_table_cmk_encryption,
+            )
+
+            check = bigquery_table_cmk_encryption()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_one_compliant_table(self):
+        bigquery_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption.bigquery_client",
+            new=bigquery_client,
+        ):
+            from prowler.providers.gcp.services.bigquery.bigquery_service import Table
+            from prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption import (
+                bigquery_table_cmk_encryption,
+            )
+
+            table = Table(
+                name="test",
+                id="1234567890",
+                region="us-central1",
+                cmk_encryption=True,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            bigquery_client.project_ids = [GCP_PROJECT_ID]
+            bigquery_client.tables = [table]
+
+            check = bigquery_table_cmk_encryption()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Table {table.name} is encrypted with Customer-Managed Keys (CMKs)."
+            )
+            assert result[0].resource_id == table.id
+            assert result[0].resource_name == table.name
+            assert result[0].project_id == table.project_id
+            assert result[0].location == table.region
+
+    def test_one_non_compliant_table(self):
+        bigquery_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption.bigquery_client",
+            new=bigquery_client,
+        ):
+            from prowler.providers.gcp.services.bigquery.bigquery_service import Table
+            from prowler.providers.gcp.services.bigquery.bigquery_table_cmk_encryption.bigquery_table_cmk_encryption import (
+                bigquery_table_cmk_encryption,
+            )
+
+            table = Table(
+                name="test",
+                id="1234567890",
+                region="us-central1",
+                cmk_encryption=False,
+                project_id=GCP_PROJECT_ID,
+            )
+
+            bigquery_client.project_ids = [GCP_PROJECT_ID]
+            bigquery_client.tables = [table]
+
+            check = bigquery_table_cmk_encryption()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Table {table.name} is not encrypted with Customer-Managed Keys (CMKs)."
+            )
+            assert result[0].resource_id == table.id
+            assert result[0].resource_name == table.name
+            assert result[0].project_id == table.project_id
+            assert result[0].location == table.region

tests/providers/gcp/services/cloudsql/cloudsql_instance_automated_backups/cloudsql_instance_automated_backups_test.py

@@ -0,0 +1,119 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_automated_backups:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups import (
+                cloudsql_instance_automated_backups,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_automated_backups()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_with_automated_backups(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups import (
+                cloudsql_instance_automated_backups,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_automated_backups()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Database Instance instance1 has automated backups configured."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_without_automated_backups(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_automated_backups.cloudsql_instance_automated_backups import (
+                cloudsql_instance_automated_backups,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=False,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_automated_backups()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Database Instance instance1 does not have automated backups configured."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_local_infile_flag/cloudsql_instance_mysql_local_infile_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_mysql_local_infile_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import (
+                cloudsql_instance_mysql_local_infile_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_mysql_local_infile_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_postgres_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import (
+                cloudsql_instance_mysql_local_infile_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_local_infile_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_with_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import (
+                cloudsql_instance_mysql_local_infile_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_local_infile_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "MySQL Instance instance1 does not have 'local_infile' flag set to 'off'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_with_local_infile_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import (
+                cloudsql_instance_mysql_local_infile_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "local_infile", "value": "off"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_local_infile_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "MySQL Instance instance1 has 'local_infile' flag set to 'off'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_with_local_infile_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_local_infile_flag.cloudsql_instance_mysql_local_infile_flag import (
+                cloudsql_instance_mysql_local_infile_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "local_infile", "value": "on"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_local_infile_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "MySQL Instance instance1 does not have 'local_infile' flag set to 'off'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_mysql_skip_show_database_flag/cloudsql_instance_mysql_skip_show_database_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_mysql_skip_show_database_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import (
+                cloudsql_instance_mysql_skip_show_database_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_mysql_skip_show_database_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_postgres_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import (
+                cloudsql_instance_mysql_skip_show_database_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_skip_show_database_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import (
+                cloudsql_instance_mysql_skip_show_database_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_skip_show_database_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "MySQL Instance instance1 does not have 'skip_show_database' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_with_skip_show_databases_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import (
+                cloudsql_instance_mysql_skip_show_database_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "skip_show_database", "value": "off"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_skip_show_database_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "MySQL Instance instance1 does not have 'skip_show_database' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_with_skip_show_databases_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_mysql_skip_show_database_flag.cloudsql_instance_mysql_skip_show_database_flag import (
+                cloudsql_instance_mysql_skip_show_database_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "skip_show_database", "value": "on"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_mysql_skip_show_database_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "MySQL Instance instance1 has 'skip_show_database' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_enable_pgaudit_flag/cloudsql_instance_postgres_enable_pgaudit_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_enable_pgaudit_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import (
+                cloudsql_instance_postgres_enable_pgaudit_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_enable_pgaudit_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import (
+                cloudsql_instance_postgres_enable_pgaudit_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_enable_pgaudit_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import (
+                cloudsql_instance_postgres_enable_pgaudit_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_enable_pgaudit_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'cloudsql.enable_pgaudit' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_pgaudit_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import (
+                cloudsql_instance_postgres_enable_pgaudit_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "cloudsql.enable_pgaudit", "value": "off"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_enable_pgaudit_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'cloudsql.enable_pgaudit' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_pgaudit_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_enable_pgaudit_flag.cloudsql_instance_postgres_enable_pgaudit_flag import (
+                cloudsql_instance_postgres_enable_pgaudit_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "cloudsql.enable_pgaudit", "value": "on"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_enable_pgaudit_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'cloudsql.enable_pgaudit' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_connections_flag/cloudsql_instance_postgres_log_connections_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_connections_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import (
+                cloudsql_instance_postgres_log_connections_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_connections_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import (
+                cloudsql_instance_postgres_log_connections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_connections_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import (
+                cloudsql_instance_postgres_log_connections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_connections_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_connections' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_connections_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import (
+                cloudsql_instance_postgres_log_connections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_connections", "value": "off"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_connections_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_connections' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_connections_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_connections_flag.cloudsql_instance_postgres_log_connections_flag import (
+                cloudsql_instance_postgres_log_connections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_connections", "value": "on"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_connections_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_connections' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_disconnections_flag/cloudsql_instance_postgres_log_disconnections_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_disconnections_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import (
+                cloudsql_instance_postgres_log_disconnections_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_disconnections_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import (
+                cloudsql_instance_postgres_log_disconnections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_disconnections_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import (
+                cloudsql_instance_postgres_log_disconnections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_disconnections_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_disconnections' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_disconnections_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import (
+                cloudsql_instance_postgres_log_disconnections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_disconnections", "value": "off"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_disconnections_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_disconnections' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_disconnections_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_disconnections_flag.cloudsql_instance_postgres_log_disconnections_flag import (
+                cloudsql_instance_postgres_log_disconnections_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_disconnections", "value": "on"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_disconnections_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_disconnections' flag set to 'on'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_error_verbosity_flag/cloudsql_instance_postgres_log_error_verbosity_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_error_verbosity_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import (
+                cloudsql_instance_postgres_log_error_verbosity_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_error_verbosity_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import (
+                cloudsql_instance_postgres_log_error_verbosity_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_error_verbosity_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import (
+                cloudsql_instance_postgres_log_error_verbosity_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_error_verbosity_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_error_verbosity' flag set to 'default'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_error_verbosity_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import (
+                cloudsql_instance_postgres_log_error_verbosity_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_error_verbosity", "value": "off"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_error_verbosity_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_error_verbosity' flag set to 'default'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_error_verbosity_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_error_verbosity_flag.cloudsql_instance_postgres_log_error_verbosity_flag import (
+                cloudsql_instance_postgres_log_error_verbosity_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_error_verbosity", "value": "default"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_error_verbosity_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_error_verbosity' flag set to 'default'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_duration_statement_flag/cloudsql_instance_postgres_log_min_duration_statement_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_min_duration_statement_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import (
+                cloudsql_instance_postgres_log_min_duration_statement_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_min_duration_statement_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import (
+                cloudsql_instance_postgres_log_min_duration_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_duration_statement_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import (
+                cloudsql_instance_postgres_log_min_duration_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_duration_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_min_duration_statement' flag set to '-1'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_min_duration_statement_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import (
+                cloudsql_instance_postgres_log_min_duration_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_min_duration_statement", "value": "0"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_duration_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_min_duration_statement' flag set to '-1'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_min_duration_statement_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_duration_statement_flag.cloudsql_instance_postgres_log_min_duration_statement_flag import (
+                cloudsql_instance_postgres_log_min_duration_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_min_duration_statement", "value": "-1"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_duration_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_min_duration_statement' flag set to '-1'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_error_statement_flag/cloudsql_instance_postgres_log_min_error_statement_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_min_error_statement_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import (
+                cloudsql_instance_postgres_log_min_error_statement_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_min_error_statement_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import (
+                cloudsql_instance_postgres_log_min_error_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_error_statement_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import (
+                cloudsql_instance_postgres_log_min_error_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_error_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_min_error_statement' flag set minimum to 'error'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_min_error_statement_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import (
+                cloudsql_instance_postgres_log_min_error_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_min_error_statement", "value": "warning"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_error_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_min_error_statement' flag set minimum to 'error'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_min_error_statement_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_error_statement_flag.cloudsql_instance_postgres_log_min_error_statement_flag import (
+                cloudsql_instance_postgres_log_min_error_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_min_error_statement", "value": "error"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_error_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_min_error_statement' flag set minimum to 'error'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_min_messages_flag/cloudsql_instance_postgres_log_min_messages_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_min_messages_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import (
+                cloudsql_instance_postgres_log_min_messages_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_min_messages_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import (
+                cloudsql_instance_postgres_log_min_messages_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_messages_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import (
+                cloudsql_instance_postgres_log_min_messages_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_messages_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_min_messages' flag set minimum to 'error'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_min_messages_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import (
+                cloudsql_instance_postgres_log_min_messages_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_min_messages", "value": "debug"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_messages_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_min_messages' flag set minimum to 'error'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_min_messages_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_min_messages_flag.cloudsql_instance_postgres_log_min_messages_flag import (
+                cloudsql_instance_postgres_log_min_messages_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_min_messages", "value": "error"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_min_messages_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_min_messages' flag set minimum to 'error'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID

tests/providers/gcp/services/cloudsql/cloudsql_instance_postgres_log_statement_flag/cloudsql_instance_postgres_log_statement_flag_test.py

@@ -0,0 +1,200 @@
+from unittest import mock
+
+from tests.providers.gcp.gcp_fixtures import (
+    GCP_EU1_LOCATION,
+    GCP_PROJECT_ID,
+    set_mocked_gcp_provider,
+)
+
+
+class Test_cloudsql_instance_postgres_log_statement_flag:
+    def test_no_cloudsql_instances(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import (
+                cloudsql_instance_postgres_log_statement_flag,
+            )
+
+            cloudsql_client.instances = []
+
+            check = cloudsql_instance_postgres_log_statement_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_mysql_instance(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import (
+                cloudsql_instance_postgres_log_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="MYSQL_5_7",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_statement_flag()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_cloudsql_instance_no_flags(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import (
+                cloudsql_instance_postgres_log_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_statement' flag set to 'ddl'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_statement_flag_off(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import (
+                cloudsql_instance_postgres_log_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_statement", "value": "all"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 does not have 'log_statement' flag set to 'ddl'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID
+
+    def test_cloudsql_instance_log_statement_flag_on(self):
+        cloudsql_client = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_gcp_provider(),
+        ), mock.patch(
+            "prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag.cloudsql_client",
+            new=cloudsql_client,
+        ):
+            from prowler.providers.gcp.services.cloudsql.cloudsql_instance_postgres_log_statement_flag.cloudsql_instance_postgres_log_statement_flag import (
+                cloudsql_instance_postgres_log_statement_flag,
+            )
+            from prowler.providers.gcp.services.cloudsql.cloudsql_service import (
+                Instance,
+            )
+
+            cloudsql_client.instances = [
+                Instance(
+                    name="instance1",
+                    version="POSTGRES_15",
+                    ip_addresses=[],
+                    region=GCP_EU1_LOCATION,
+                    public_ip=False,
+                    ssl=False,
+                    automated_backups=True,
+                    authorized_networks=[],
+                    flags=[{"name": "log_statement", "value": "ddl"}],
+                    project_id=GCP_PROJECT_ID,
+                )
+            ]
+
+            check = cloudsql_instance_postgres_log_statement_flag()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "PostgreSQL Instance instance1 has 'log_statement' flag set to 'ddl'."
+            )
+            assert result[0].resource_id == "instance1"
+            assert result[0].resource_name == "instance1"
+            assert result[0].location == GCP_EU1_LOCATION
+            assert result[0].project_id == GCP_PROJECT_ID