From 7ec9f6572f8f5c8d703a21bbc4ab662874e7ee1d Mon Sep 17 00:00:00 2001 From: Andoni Alonso <14891798+andoniaf@users.noreply.github.com> Date: Thu, 29 May 2025 12:39:58 +0200 Subject: [PATCH] chore(iac): add iac provider tests (#7874) --- prowler/providers/iac/iac_provider.py | 1 - tests/providers/iac/iac_fixtures.py | 67 ++++++++++++ tests/providers/iac/iac_provider_test.py | 129 +++++++++++++++++++++++ 3 files changed, 196 insertions(+), 1 deletion(-) create mode 100644 tests/providers/iac/iac_fixtures.py create mode 100644 tests/providers/iac/iac_provider_test.py diff --git a/prowler/providers/iac/iac_provider.py b/prowler/providers/iac/iac_provider.py index 26b54a5b3d..f84f9250fb 100644 --- a/prowler/providers/iac/iac_provider.py +++ b/prowler/providers/iac/iac_provider.py @@ -98,7 +98,6 @@ class IacProvider(Provider): Returns: CheckReportIAC: The processed check report """ - metadata_dict = { "Provider": "iac", "CheckID": check.get("check_id", ""), diff --git a/tests/providers/iac/iac_fixtures.py b/tests/providers/iac/iac_fixtures.py new file mode 100644 index 0000000000..70f0698bdf --- /dev/null +++ b/tests/providers/iac/iac_fixtures.py @@ -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" diff --git a/tests/providers/iac/iac_provider_test.py b/tests/providers/iac/iac_provider_test.py new file mode 100644 index 0000000000..7d27938abd --- /dev/null +++ b/tests/providers/iac/iac_provider_test.py @@ -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