diff --git a/tests/providers/gcp/lib/fix/gcpfixer_test.py b/tests/providers/gcp/lib/fix/gcpfixer_test.py new file mode 100644 index 0000000000..19319e855a --- /dev/null +++ b/tests/providers/gcp/lib/fix/gcpfixer_test.py @@ -0,0 +1,98 @@ +from unittest.mock import patch + +from prowler.lib.check.models import ( + Check_Report_GCP, + CheckMetadata, + Code, + Recommendation, + Remediation, +) +from prowler.providers.gcp.lib.fix.fixer import GCPFixer + + +def get_mock_gcp_finding(): + metadata = CheckMetadata( + Provider="gcp", + CheckID="test_check", + CheckTitle="Test Check", + CheckType=["type1"], + CheckAliases=[], + ServiceName="testservice", + SubServiceName="", + ResourceIdTemplate="", + Severity="low", + ResourceType="resource", + Description="desc", + Risk="risk", + RelatedUrl="url", + Remediation=Remediation( + Code=Code(NativeIaC="", Terraform="", CLI="", Other=""), + Recommendation=Recommendation(Text="", Url=""), + ), + Categories=["cat1"], + DependsOn=[], + RelatedTo=[], + Notes="", + Compliance=[], + ) + return Check_Report_GCP( + metadata.dict(), + project_id="project_id", + resource_id="resource_id", + resource_name="resource_name", + location="location", + ) + + +class TestGCPFixer: + def test_fix_success(self): + finding = get_mock_gcp_finding() + finding.status = "FAIL" + fixer = GCPFixer(description="desc", service="compute") + assert fixer.fix(finding=finding) + + def test_fix_failure(self, caplog): + finding = get_mock_gcp_finding() + finding.status = "FAIL" + fixer = GCPFixer(description="desc", service="compute") + with patch("prowler.providers.gcp.lib.fix.fixer.logger") as mock_logger: + with caplog.at_level("ERROR"): + result = fixer.fix(finding=None) + assert result is False + assert mock_logger.error.called + + def test_get_fixer_info(self): + fixer = GCPFixer( + description="desc", + service="compute", + cost_impact=True, + cost_description="cost", + iam_policy_required={"roles": ["roles/owner"]}, + ) + info = fixer._get_fixer_info() + assert info["description"] == "desc" + assert info["cost_impact"] is True + assert info["cost_description"] == "cost" + assert info["service"] == "compute" + assert info["iam_policy_required"] == {"roles": ["roles/owner"]} + assert info["provider"] == "gcp" + + def test_fix_prints(self): + fixer = GCPFixer(description="desc", service="compute") + finding = get_mock_gcp_finding() + with ( + patch("builtins.print") as mock_print, + patch("prowler.providers.gcp.lib.fix.fixer.logger"), + ): + result = fixer.fix(finding=finding) + assert result is True + mock_print.assert_called_once_with( + f"\tFIXING {finding.resource_id} in project {finding.project_id}..." + ) + + def test_fix_exception(self): + fixer = GCPFixer(description="desc", service="compute") + with patch("prowler.providers.gcp.lib.fix.fixer.logger") as mock_logger: + result = fixer.fix(finding=None) + assert result is False + assert mock_logger.error.called diff --git a/tests/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer_test.py b/tests/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer_test.py new file mode 100644 index 0000000000..abfd4ea361 --- /dev/null +++ b/tests/providers/gcp/services/compute/compute_project_os_login_enabled/compute_project_os_login_enabled_fixer_test.py @@ -0,0 +1,130 @@ +from re import search +from unittest import mock + +from prowler.providers.gcp.models import GCPProject +from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider + + +class Test_compute_project_os_login_enabled: + def test_compute_no_project(self): + compute_client = mock.MagicMock() + compute_client.project_ids = [GCP_PROJECT_ID] + compute_client.projects = [] + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), + mock.patch( + "prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled.compute_client", + new=compute_client, + ), + ): + from prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled import ( + compute_project_os_login_enabled, + ) + + check = compute_project_os_login_enabled() + result = check.execute() + assert len(result) == 0 + + def test_one_compliant_project(self): + from prowler.providers.gcp.services.compute.compute_service import Project + + project = Project( + id=GCP_PROJECT_ID, + enable_oslogin=True, + ) + + compute_client = mock.MagicMock() + compute_client.project_ids = [GCP_PROJECT_ID] + compute_client.compute_projects = [project] + compute_client.projects = { + GCP_PROJECT_ID: GCPProject( + id=GCP_PROJECT_ID, + number="123456789012", + name="test", + labels={}, + lifecycle_state="ACTIVE", + ) + } + compute_client.region = "global" + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), + mock.patch( + "prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled.compute_client", + new=compute_client, + ), + ): + from prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled import ( + compute_project_os_login_enabled, + ) + + check = compute_project_os_login_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert search( + f"Project {project.id} has OS Login enabled", + result[0].status_extended, + ) + assert result[0].resource_id == project.id + assert result[0].resource_name == "test" + assert result[0].location == "global" + assert result[0].project_id == GCP_PROJECT_ID + + def test_one_non_compliant_project(self): + from prowler.providers.gcp.services.compute.compute_service import Project + + project = Project( + id=GCP_PROJECT_ID, + enable_oslogin=False, + ) + + compute_client = mock.MagicMock() + compute_client.project_ids = [GCP_PROJECT_ID] + compute_client.compute_projects = [project] + compute_client.projects = { + GCP_PROJECT_ID: GCPProject( + id=GCP_PROJECT_ID, + number="123456789012", + name="test", + labels={}, + lifecycle_state="ACTIVE", + ) + } + compute_client.region = "global" + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_gcp_provider(), + ), + mock.patch( + "prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled.compute_client", + new=compute_client, + ), + ): + from prowler.providers.gcp.services.compute.compute_project_os_login_enabled.compute_project_os_login_enabled import ( + compute_project_os_login_enabled, + ) + + check = compute_project_os_login_enabled() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert search( + f"Project {project.id} does not have OS Login enabled", + result[0].status_extended, + ) + assert result[0].resource_id == project.id + assert result[0].resource_name == "test" + assert result[0].location == "global" + assert result[0].project_id == GCP_PROJECT_ID