Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,6 @@ def get_list_finding(
)
list_open_findings.append(finding_open)

if "'error'" in str(result):
if isinstance(result, dict) and "error" in result:
raise Exception(result.get("error"))
return list_open_findings
Original file line number Diff line number Diff line change
Expand Up @@ -413,7 +413,11 @@ def _async_scan(self, queue, checkov_config: CheckovConfig, command_prefix):
result = []
try:
output = self._execute(checkov_config, command_prefix)
result.append(json.loads(output))
parsed_output = json.loads(output)
if isinstance(parsed_output, list):
result.extend(parsed_output)
else:
result.append(parsed_output)
except json.JSONDecodeError as e:
error_msg = f"Failed to parse Checkov output as JSON: {e}"
logger.error(error_msg)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
Category,
)
from datetime import datetime
import pytest

def test_get_list_finding():
results_scan_list = [
Expand Down Expand Up @@ -130,3 +131,38 @@ def test_get_list_finding():
)

assert list_findings == list_findings_compare


def test_get_list_finding_does_not_raise_when_check_text_contains_error_word():
results_scan_list = [
{
"check_type": "terraform",
"results": {
"failed_checks": [
{
"check_id": "CKV_AWS_1",
"check_name": "Ensure proper error handling is configured",
"resource": "aws_lambda_function.this",
"repo_file_path": "/main.tf",
"guideline": "Add error handling",
}
]
},
}
]

list_findings = CheckovDeserealizator.get_list_finding(
results_scan_list, {}, "high", "vulnerability"
)

assert len(list_findings) == 1
assert list_findings[0].id == "CKV_AWS_1"


def test_get_list_finding_raises_on_real_error_result():
results_scan_list = [
{"error": "Checkov execution failed", "checkov_config": "terraform"}
]

with pytest.raises(Exception, match="Checkov execution failed"):
CheckovDeserealizator.get_list_finding(results_scan_list, {}, "high", "vulnerability")
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,31 @@ def test_async_scan(mock_checkov_tool, checkov_tool):

assert output_queue.get() == [{"key": "value"}]

@patch(
"devsecops_engine_tools.engine_sast.engine_iac.src.infrastructure.driven_adapters.checkov.checkov_tool.CheckovTool._execute",
autospec=True,
)
def test_async_scan_with_multiple_frameworks_list_output(mock_checkov_tool, checkov_tool):
checkov_config = MagicMock()
checkov_config.path_config_file = "/path/to/config/"
checkov_config.config_file_name = "checkov_config"

output_queue = Queue()

mock_checkov_tool.return_value = (
'[{"check_type": "terraform", "results": {"failed_checks": []}}, '
'{"check_type": "terraform_plan", "results": {"failed_checks": []}}]'
)

checkov_tool._async_scan(output_queue, checkov_config, "checkov")

result = output_queue.get()
assert result == [
{"check_type": "terraform", "results": {"failed_checks": []}},
{"check_type": "terraform_plan", "results": {"failed_checks": []}},
]
assert all(isinstance(entry, dict) for entry in result)

@patch(
"devsecops_engine_tools.engine_sast.engine_iac.src.infrastructure.driven_adapters.checkov.checkov_tool.CheckovTool._execute",
autospec=True,
Expand Down
Loading