mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 12:31:54 +00:00
feat(gcp): add first version of gcp tests
This commit is contained in:
@@ -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
|
||||
+130
@@ -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
|
||||
Reference in New Issue
Block a user