fix(check_report): WIP

This commit is contained in:
Pepe Fagoaga
2025-01-30 08:07:04 +01:00
parent bf4a8c6815
commit 210020021d
2 changed files with 186 additions and 11 deletions
+39 -9
View File
@@ -441,7 +441,16 @@ class Check_Report:
@dataclass
class Check_Report_AWS(Check_Report):
"""Contains the AWS Check's finding information."""
"""
Contains the AWS Check's finding information.
Attributes:
resource_id (str): The resource ID.
resource_arn (str): The resource ARN.
region (str): The region of the resource.
Raises:
AttributeError: If the any of the required attributes are not present.
"""
resource_id: str
resource_arn: str
@@ -449,7 +458,13 @@ class Check_Report_AWS(Check_Report):
def __init__(self, metadata: Dict, resource: Any) -> None:
super().__init__(metadata, resource)
self.resource_id = getattr(resource, "id", getattr(resource, "name"))
# Use resource.name as fallback if no resource.id
# If no name or id, an exception is raised
try:
self.resource_id = getattr(resource, "id")
except AttributeError:
self.resource_id = getattr(resource, "name")
# ARN and Region must be present
self.resource_arn = getattr(resource, "arn")
self.region = getattr(resource, "region")
@@ -471,10 +486,17 @@ class Check_Report_Azure(Check_Report):
resource: Basic information about the resource. Defaults to None.
"""
super().__init__(metadata, resource)
self.resource_name = getattr(
resource, "name", getattr(resource, "resource_name")
)
self.resource_id = getattr(resource, "id", getattr(resource, "resource_id"))
try:
self.resource_id = getattr(resource, "id")
except AttributeError:
self.resource_id = getattr(resource, "resource_id")
try:
self.resource_name = getattr(resource, "name")
except AttributeError:
self.resource_name = getattr(resource, "resource_name")
# TODO: review self.subscription
self.subscription = ""
self.location = getattr(resource, "location", "global")
@@ -498,6 +520,12 @@ class Check_Report_GCP(Check_Report):
project_id=None,
) -> None:
super().__init__(metadata, resource)
try:
self.resource_id = getattr(resource, "id")
except AttributeError:
self.resource_id = getattr(resource, "name")
self.resource_id = resource_id or getattr(
resource, "id", getattr(resource, "name")
)
@@ -519,11 +547,13 @@ class Check_Report_Kubernetes(Check_Report):
def __init__(self, metadata: Dict, resource: Any) -> None:
super().__init__(metadata, resource)
self.resource_id = getattr(resource, "uid", getattr(resource, "name"))
try:
self.resource_id = getattr(resource, "uid")
except AttributeError:
self.resource_id = getattr(resource, "name")
self.resource_name = getattr(resource, "name")
self.namespace = getattr(resource, "namespace", "cluster-wide")
if not self.namespace:
self.namespace = "cluster-wide"
@dataclass
+147 -2
View File
@@ -1,6 +1,13 @@
from unittest import mock
from prowler.lib.check.models import CheckMetadata
import pytest
from prowler.lib.check.models import (
Check_Report,
Check_Report_AWS,
Check_Report_Azure,
CheckMetadata,
)
from tests.lib.check.compliance_check_test import custom_compliance_metadata
mock_metadata = CheckMetadata(
@@ -63,7 +70,6 @@ mock_metadata_lambda = CheckMetadata(
class TestCheckMetada:
@mock.patch("prowler.lib.check.models.load_check_metadata")
@mock.patch("prowler.lib.check.models.recover_checks_from_provider")
def test_get_bulk(self, mock_recover_checks, mock_load_metadata):
@@ -324,3 +330,142 @@ class TestCheckMetada:
result = CheckMetadata.list(bulk_checks_metadata=bulk_metadata)
assert result == set()
class TestCheckReport:
def test_check_report_resource_dict(self):
resource = {"id": "test_id"}
check_report = Check_Report(metadata=mock_metadata.json(), resource=resource)
assert check_report.status == ""
assert check_report.check_metadata == mock_metadata
assert check_report.resource == resource
assert check_report.status_extended == ""
assert check_report.resource_details == ""
assert check_report.resource_tags == []
assert check_report.muted is False
# def test_check_report_resource_dict_method(self):
# resource = mock.Mock()
# resource.dict = lambda: {"id": "test_id"}
# check_report = Check_Report(metadata=mock_metadata.json(), resource=resource)
# assert check_report.status == ""
# assert check_report.check_metadata == mock_metadata
# assert check_report.resource == {"id": "test_id"}
# assert check_report.status_extended == ""
# assert check_report.resource_details == ""
# assert check_report.resource_tags == []
# assert check_report.muted is False
class TestCheckReportAWS:
def test_check_report_aws(self):
resource = mock.Mock
resource.id = "test_id"
resource.arn = "test_arn"
resource.region = "test_region"
check_report_aws = Check_Report_AWS(
metadata=mock_metadata.json(), resource=resource
)
assert check_report_aws.resource_id == "test_id"
assert check_report_aws.resource_arn == "test_arn"
assert check_report_aws.region == "test_region"
def test_check_report_aws_no_id_but_name(self):
resource = mock.Mock
resource.name = "test_id"
resource.arn = "test_arn"
resource.region = "test_region"
report = Check_Report_AWS(metadata=mock_metadata.json(), resource=resource)
assert report.resource_id == "test_id"
assert report.resource_arn == "test_arn"
assert report.region == "test_region"
def test_check_report_aws_no_id_or_name(self):
resource = mock.Mock
resource.arn = "test_arn"
resource.region = "test_region"
with pytest.raises(AttributeError):
Check_Report_AWS(metadata=mock_metadata.json(), resource=resource)
def test_check_report_aws_no_arn(self):
resource = mock.Mock
resource.id = "test_id"
resource.region = "test_region"
with pytest.raises(AttributeError):
Check_Report_AWS(metadata=mock_metadata.json(), resource=resource)
def test_check_report_aws_no_region(self):
resource = mock.Mock
resource.id = "test_id"
resource.region = "test_region"
with pytest.raises(AttributeError):
Check_Report_AWS(metadata=mock_metadata.json(), resource=resource)
# check finding without resource_id
# raise log error
# continue execution
class TestCheckReportAzure:
def test_check_report_azure(self):
resource = mock.Mock
resource.id = "test_id"
resource.name = "test_name"
resource.location = "test_location"
report = Check_Report_Azure(metadata=mock_metadata.json(), resource=resource)
assert report.resource_id == "test_id"
assert report.resource_name == "test_name"
assert report.location == "test_location"
def test_check_report_azure_no_id(self):
resource = mock.Mock
resource.name = "test_name"
resource.location = "global"
with pytest.raises(AttributeError):
Check_Report_Azure(metadata=mock_metadata.json(), resource=resource)
def test_check_report_azure_resource_id(self):
resource = mock.Mock
resource.resource_id = "resource_id"
resource.name = "test_name"
resource.location = "global"
report = Check_Report_Azure(metadata=mock_metadata.json(), resource=resource)
assert report.resource_id == "test_id"
assert report.resource_name == "test_name"
assert report.location == "global"
def test_check_report_azure_no_name(self):
resource = mock.Mock
resource.id = "test_id"
resource.location = "global"
with pytest.raises(AttributeError):
Check_Report_Azure(metadata=mock_metadata.json(), resource=resource)
def test_check_report_azure_resource_name(self):
resource = mock.Mock
resource.id = "test_id"
resource.resource_name = "test_name"
resource.location = "global"
report = Check_Report_Azure(metadata=mock_metadata.json(), resource=resource)
assert report.resource_id == "test_id"
assert report.resource_name == "test_name"
assert report.location == "global"
def test_check_report_azure_no_location(self):
resource = mock.Mock
resource.id = "test_id"
resource.name = "test_name"
report = Check_Report_Azure(metadata=mock_metadata.json(), resource=resource)
assert report.resource_id == "test_id"
assert report.resource_name == "test_name"
assert report.location == "global"