mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
chore(iac): add iac provider tests (#7874)
This commit is contained in:
@@ -98,7 +98,6 @@ class IacProvider(Provider):
|
||||
Returns:
|
||||
CheckReportIAC: The processed check report
|
||||
"""
|
||||
|
||||
metadata_dict = {
|
||||
"Provider": "iac",
|
||||
"CheckID": check.get("check_id", ""),
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
# IAC Provider Constants
|
||||
DEFAULT_SCAN_PATH = "."
|
||||
|
||||
# Sample Checkov Output
|
||||
SAMPLE_CHECKOV_OUTPUT = [
|
||||
{
|
||||
"check_type": "terraform",
|
||||
"results": {
|
||||
"failed_checks": [
|
||||
{
|
||||
"check_id": "CKV_AWS_1",
|
||||
"check_name": "Ensure S3 bucket has encryption enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_1-s3-bucket-has-encryption-enabled",
|
||||
"severity": "low",
|
||||
},
|
||||
{
|
||||
"check_id": "CKV_AWS_2",
|
||||
"check_name": "Ensure S3 bucket has public access blocked",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_2-s3-bucket-has-public-access-blocked",
|
||||
"severity": "low",
|
||||
},
|
||||
],
|
||||
"passed_checks": [
|
||||
{
|
||||
"check_id": "CKV_AWS_3",
|
||||
"check_name": "Ensure S3 bucket has versioning enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_3-s3-bucket-has-versioning-enabled",
|
||||
"severity": "low",
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
]
|
||||
|
||||
# Sample Finding Data
|
||||
SAMPLE_FINDING = SAMPLE_CHECKOV_OUTPUT[0]
|
||||
|
||||
SAMPLE_FAILED_CHECK = {
|
||||
"check_id": "CKV_AWS_1",
|
||||
"check_name": "Ensure S3 bucket has encryption enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_1-s3-bucket-has-encryption-enabled",
|
||||
"severity": "low",
|
||||
}
|
||||
|
||||
SAMPLE_PASSED_CHECK = {
|
||||
"check_id": "CKV_AWS_3",
|
||||
"check_name": "Ensure S3 bucket has versioning enabled",
|
||||
"guideline": "https://docs.bridgecrew.io/docs/s3_3-s3-bucket-has-versioning-enabled",
|
||||
"severity": "low",
|
||||
}
|
||||
|
||||
|
||||
def get_sample_checkov_json_output():
|
||||
"""Return sample Checkov JSON output as string"""
|
||||
import json
|
||||
|
||||
return json.dumps(SAMPLE_CHECKOV_OUTPUT)
|
||||
|
||||
|
||||
def get_empty_checkov_output():
|
||||
"""Return empty Checkov output as string"""
|
||||
return "[]"
|
||||
|
||||
|
||||
def get_invalid_checkov_output():
|
||||
"""Return invalid JSON output as string"""
|
||||
return "invalid json output"
|
||||
@@ -0,0 +1,129 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from prowler.lib.check.models import CheckReportIAC
|
||||
from prowler.providers.iac.iac_provider import IacProvider
|
||||
from tests.providers.iac.iac_fixtures import (
|
||||
DEFAULT_SCAN_PATH,
|
||||
SAMPLE_FAILED_CHECK,
|
||||
SAMPLE_FINDING,
|
||||
SAMPLE_PASSED_CHECK,
|
||||
get_empty_checkov_output,
|
||||
get_invalid_checkov_output,
|
||||
get_sample_checkov_json_output,
|
||||
)
|
||||
|
||||
|
||||
class TestIacProvider:
|
||||
def test_iac_provider(self):
|
||||
"""Test IAC provider with default parameters"""
|
||||
|
||||
provider = IacProvider()
|
||||
|
||||
assert provider._type == "iac"
|
||||
assert provider.type == "iac"
|
||||
assert provider.scan_path == DEFAULT_SCAN_PATH
|
||||
assert provider._audit_config == {}
|
||||
assert provider._mutelist is None
|
||||
|
||||
def test_iac_provider_custom_scan_path(self):
|
||||
"""Test IAC provider with custom scan path"""
|
||||
custom_path = "/custom/path"
|
||||
|
||||
provider = IacProvider(scan_path=custom_path)
|
||||
|
||||
assert provider._type == "iac"
|
||||
assert provider.scan_path == custom_path
|
||||
|
||||
def test_iac_provider_process_check_failed(self):
|
||||
"""Test processing a failed check"""
|
||||
provider = IacProvider()
|
||||
|
||||
report = provider._process_check(SAMPLE_FINDING, SAMPLE_FAILED_CHECK, "FAIL")
|
||||
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "FAIL"
|
||||
|
||||
assert report.check_metadata.Provider == "iac"
|
||||
assert report.check_metadata.CheckID == SAMPLE_FAILED_CHECK["check_id"]
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_FAILED_CHECK["check_name"]
|
||||
assert report.check_metadata.Severity == "low"
|
||||
assert report.check_metadata.RelatedUrl == SAMPLE_FAILED_CHECK["guideline"]
|
||||
|
||||
def test_iac_provider_process_check_passed(self):
|
||||
"""Test processing a passed check"""
|
||||
provider = IacProvider()
|
||||
|
||||
report = provider._process_check(SAMPLE_FINDING, SAMPLE_PASSED_CHECK, "PASS")
|
||||
|
||||
assert isinstance(report, CheckReportIAC)
|
||||
assert report.status == "PASS"
|
||||
|
||||
assert report.check_metadata.Provider == "iac"
|
||||
assert report.check_metadata.CheckID == SAMPLE_PASSED_CHECK["check_id"]
|
||||
assert report.check_metadata.CheckTitle == SAMPLE_PASSED_CHECK["check_name"]
|
||||
assert report.check_metadata.Severity == "low"
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_success(self, mock_subprocess):
|
||||
"""Test successful IAC scan with Checkov"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(
|
||||
stdout=get_sample_checkov_json_output(), stderr=""
|
||||
)
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
|
||||
# Should have 2 failed checks + 1 passed check = 3 total reports
|
||||
assert len(reports) == 3
|
||||
|
||||
# Check that we have both failed and passed reports
|
||||
failed_reports = [r for r in reports if r.status == "FAIL"]
|
||||
passed_reports = [r for r in reports if r.status == "PASS"]
|
||||
|
||||
assert len(failed_reports) == 2
|
||||
assert len(passed_reports) == 1
|
||||
|
||||
# Verify subprocess was called correctly
|
||||
mock_subprocess.assert_called_once_with(
|
||||
["checkov", "-d", "/test/directory", "-o", "json"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_empty_output(self, mock_subprocess):
|
||||
"""Test IAC scan with empty Checkov output"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(
|
||||
stdout=get_empty_checkov_output(), stderr=""
|
||||
)
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
|
||||
assert len(reports) == 0
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_invalid_json(self, mock_subprocess):
|
||||
"""Test IAC scan with invalid JSON output"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(
|
||||
stdout=get_invalid_checkov_output(), stderr=""
|
||||
)
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
|
||||
assert len(reports) == 0
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_iac_provider_run_scan_null_output(self, mock_subprocess):
|
||||
"""Test IAC scan with null Checkov output"""
|
||||
provider = IacProvider()
|
||||
|
||||
mock_subprocess.return_value = MagicMock(stdout="null", stderr="")
|
||||
|
||||
reports = provider.run_scan("/test/directory")
|
||||
|
||||
assert len(reports) == 0
|
||||
Reference in New Issue
Block a user