From d9889776ade2be6c27c0c6230369fdacefa0716e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A1n=20Jes=C3=BAs=20Pe=C3=B1a=20Rodr=C3=ADguez?= Date: Thu, 13 Feb 2025 21:28:10 +0100 Subject: [PATCH] test: add export unittests --- api/src/backend/api/tests/test_views.py | 126 +++++++++++- api/src/backend/api/v1/views.py | 2 - tests/lib/outputs/csv/csv_test.py | 4 +- tests/lib/outputs/finding_test.py | 249 ++++++++++++++++++++++++ tests/providers/aws/lib/s3/s3_test.py | 1 - 5 files changed, 375 insertions(+), 7 deletions(-) diff --git a/api/src/backend/api/tests/test_views.py b/api/src/backend/api/tests/test_views.py index 73f2bab027..a6438c8359 100644 --- a/api/src/backend/api/tests/test_views.py +++ b/api/src/backend/api/tests/test_views.py @@ -1,11 +1,14 @@ import json from datetime import datetime, timedelta, timezone -from unittest.mock import ANY, Mock, patch +from io import BytesIO +from unittest.mock import ANY, Mock, mock_open, patch import jwt import pytest +from botocore.exceptions import ClientError from conftest import API_JSON_CONTENT_TYPE, TEST_PASSWORD, TEST_USER from django.conf import settings +from django.http import HttpResponse from django.urls import reverse from rest_framework import status @@ -2156,6 +2159,127 @@ class TestScanViewSet: response = authenticated_client.get(reverse("scan-list"), {"sort": "invalid"}) assert response.status_code == status.HTTP_400_BAD_REQUEST + def test_report_no_output(self, authenticated_client, scans_fixture): + """ + If the scan's output_path is empty, the view should return a 404 response. + """ + scan = scans_fixture[0] + url = reverse("scan-report", kwargs={"pk": scan.pk}) + response = authenticated_client.get(url) + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.data["detail"] == "No files found" + + def test_report_s3_get_client_fail(self, authenticated_client, scans_fixture): + """ + If output_path starts with "s3://", but get_s3_client() fails, the view should return a 403. + """ + scan = scans_fixture[0] + scan.output_path = "s3://bucket/path/to/file.zip" + scan.save() + url = reverse("scan-report", kwargs={"pk": scan.pk}) + # Patch get_s3_client to raise one of the expected exceptions + client_err = ClientError({"Error": {"Code": "AccessDenied"}}, "get_s3_client") + with patch("api.v1.views.get_s3_client", side_effect=client_err): + response = authenticated_client.get(url) + assert response.status_code == status.HTTP_403_FORBIDDEN + assert response.data["detail"] == "There is a problem with the AWS credentials." + + def test_report_s3_get_object_fail(self, authenticated_client, scans_fixture): + """ + If output_path starts with "s3://", and get_s3_client() succeeds but get_object() fails, + the view should return a 500 error. + """ + scan = scans_fixture[0] + scan.output_path = "s3://bucket/path/to/file.zip" + scan.save() + url = reverse("scan-report", kwargs={"pk": scan.pk}) + client_err = ClientError({"Error": {"Code": "NoSuchKey"}}, "get_object") + with patch("api.v1.views.get_s3_client") as mock_get_s3_client, \ + patch("api.v1.views.env.str", return_value="bucket"): + s3_client = mock_get_s3_client.return_value + s3_client.get_object.side_effect = client_err + response = authenticated_client.get(url) + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert response.data["detail"] == "Error accessing cloud storage" + + def test_report_s3_success(self, authenticated_client, scans_fixture): + """ + If output_path starts with "s3://", and S3 functions succeed, the view should return an + HttpResponse with the ZIP content. + """ + fake_file_content = b"fake s3 zip content" + scan = scans_fixture[0] + scan.output_path = "s3://bucket/path/to/file.zip" + scan.save() + url = reverse("scan-report", kwargs={"pk": scan.pk}) + # Create a dummy S3 object with a 'Body' that returns our fake file content + dummy_body = BytesIO(fake_file_content) + dummy_s3_object = {"Body": dummy_body} + with patch("api.v1.views.get_s3_client") as mock_get_s3_client, \ + patch("api.v1.views.env.str", return_value="bucket"): + s3_client = mock_get_s3_client.return_value + s3_client.get_object.return_value = dummy_s3_object + response = authenticated_client.get(url) + # The view returns an HttpResponse (not a DRF Response) for file downloads + assert isinstance(response, HttpResponse) + assert response.status_code == 200 + assert response.content == fake_file_content + # Check that the Content-Disposition header includes the filename "file.zip" + content_disp = response.get("Content-Disposition", "") + assert content_disp.startswith('attachment; filename="') + assert "file.zip" in content_disp + + def test_report_local_no_files(self, authenticated_client, scans_fixture): + """ + If output_path does not start with "s3://" and glob.glob finds no matching files, + the view should return a 404 response. + """ + scan = scans_fixture[0] + scan.output_path = "/non/existent/path/*.zip" + scan.save() + url = reverse("scan-report", kwargs={"pk": scan.pk}) + with patch("api.v1.views.glob.glob", return_value=[]): + response = authenticated_client.get(url) + assert response.status_code == status.HTTP_404_NOT_FOUND + assert response.data["detail"] == "No local files found" + + def test_report_local_ioerror(self, authenticated_client, scans_fixture): + """ + If output_path does not start with "s3://", glob.glob finds a file but reading it raises an IOError, + the view should return a 500 response. + """ + scan = scans_fixture[0] + scan.output_path = "/path/to/file.zip" + scan.save() + url = reverse("scan-report", kwargs={"pk": scan.pk}) + with patch("api.v1.views.glob.glob", return_value=["/path/to/file.zip"]), \ + patch("api.v1.views.open", side_effect=IOError): + response = authenticated_client.get(url) + assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR + assert response.data["detail"] == "Error reading local file" + + def test_report_local_success(self, authenticated_client, scans_fixture): + """ + If output_path does not start with "s3://", and a local file is found and read successfully, + the view should return an HttpResponse with the file contents. + """ + fake_file_content = b"local zip file content" + scan = scans_fixture[0] + scan.output_path = "/path/to/file.zip" + scan.save() + url = reverse("scan-report", kwargs={"pk": scan.pk}) + m_open = mock_open(read_data=fake_file_content) + with patch("api.v1.views.glob.glob", return_value=["/path/to/file.zip"]), \ + patch("api.v1.views.open", m_open): + response = authenticated_client.get(url) + assert isinstance(response, HttpResponse) + assert response.status_code == 200 + assert response.content == fake_file_content + content_disp = response.get("Content-Disposition", "") + assert content_disp.startswith('attachment; filename="') + # The filename should be the basename of the file path + assert "file.zip" in content_disp + @pytest.mark.django_db class TestTaskViewSet: diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py index 47bb46fbc1..973c800be5 100644 --- a/api/src/backend/api/v1/views.py +++ b/api/src/backend/api/v1/views.py @@ -1132,7 +1132,6 @@ class ScanViewSet(BaseRLSViewSet): ) bucket_name = env.str("DJANGO_ARTIFACTS_AWS_S3_OUTPUT_BUCKET") - try: key = output_path[len(f"s3://{bucket_name}/") :] s3_object = s3_client.get_object(Bucket=bucket_name, Key=key) @@ -1143,7 +1142,6 @@ class ScanViewSet(BaseRLSViewSet): {"detail": "Error accessing cloud storage"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) - else: zip_files = glob.glob(output_path) if not zip_files: diff --git a/tests/lib/outputs/csv/csv_test.py b/tests/lib/outputs/csv/csv_test.py index 75b8e9cc70..8a96ca3017 100644 --- a/tests/lib/outputs/csv/csv_test.py +++ b/tests/lib/outputs/csv/csv_test.py @@ -144,9 +144,7 @@ class TestCSV: file_path = file.name # Instantiate the mock class - output_instance = mock_output_class( - findings, create_file_descriptor=True, file_path=file_path - ) + output_instance = mock_output_class(findings, file_path=file_path) # Check that transform was called once output_instance.transform.assert_called_once_with(findings) diff --git a/tests/lib/outputs/finding_test.py b/tests/lib/outputs/finding_test.py index 6ae17416d0..917dfa1062 100644 --- a/tests/lib/outputs/finding_test.py +++ b/tests/lib/outputs/finding_test.py @@ -1,4 +1,5 @@ from datetime import datetime +from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest @@ -55,6 +56,49 @@ def mock_get_check_compliance(*_): return {"mock_compliance_key": "mock_compliance_value"} +class DummyTag: + def __init__(self, key, value): + self.key = key + self.value = value + +class DummyTags: + def __init__(self, tags): + self._tags = tags + + def all(self): + return self._tags + +class DummyResource: + def __init__(self, uid, name, region, tags): + self.uid = uid + self.name = name + self.region = region + self.tags = DummyTags(tags) + +class DummyResources: + """Simulate a collection with a first() method.""" + def __init__(self, resource): + self._resource = resource + + def first(self): + return self._resource + +class DummyProvider: + def __init__(self, uid): + self.uid = uid + +class DummyScan: + def __init__(self, provider): + self.provider = provider + +class DummyAPIFinding: + """ + A dummy API finding model to simulate the database model. + Attributes will be added dynamically. + """ + pass + + class TestFinding: @patch( "prowler.lib.outputs.finding.get_check_compliance", @@ -461,3 +505,208 @@ class TestFinding: # Generate the finding with pytest.raises(ValidationError): Finding.generate_output(provider, check_output, output_options) + + def test_transform_api_finding(self): + """ + Test that a dummy API Finding is correctly + transformed into a Finding instance. + """ + # Set up the dummy API finding attributes + inserted_at = 1234567890 + provider = DummyProvider(uid="account123") + scan = DummyScan(provider=provider) + + # Create a dummy resource with one tag + tag = DummyTag("env", "prod") + resource = DummyResource(uid="res-uid-1", name="ResourceName1", region="us-east-1", tags=[tag]) + resources = DummyResources(resource) + + # Create a dummy check_metadata dict with all required fields + check_metadata = { + "provider": "test_provider", + "checkid": "check-001", + "checktitle": "Test Check", + "checktype": ["type1"], + "servicename": "TestService", + "subservicename": "SubService", + "severity": "high", + "resourcetype": "TestResource", + "description": "A test check", + "risk": "High risk", + "relatedurl": "http://example.com", + "remediation": { + "recommendation": { + "text": "Fix it", + "url": "http://fix.com" + }, + "code": { + "nativeiac": "iac_code", + "terraform": "terraform_code", + "cli": "cli_code", + "other": "other_code", + }, + }, + "resourceidtemplate": "template", + "categories": ["cat-one", "cat-two"], + "dependson": ["dep1"], + "relatedto": ["rel1"], + "notes": "Some notes", + } + + # Create the dummy API finding and assign required attributes + dummy_finding = DummyAPIFinding() + dummy_finding.inserted_at = inserted_at + dummy_finding.scan = scan + dummy_finding.uid = "finding-uid-1" + dummy_finding.status = "FAIL" # will be converted to Status("FAIL") + dummy_finding.status_extended = "extended" + dummy_finding.check_metadata = check_metadata + dummy_finding.resources = resources + + # Call the transform_api_finding classmethod + finding_obj = Finding.transform_api_finding(dummy_finding) + + # Fields directly set in transform_api_finding + assert finding_obj.auth_method == "" + assert finding_obj.timestamp == inserted_at + assert finding_obj.account_uid == "account123" + assert finding_obj.account_name == "" + + # Check that metadata was built correctly + meta = finding_obj.metadata + assert meta.Provider == "test_provider" + assert meta.CheckID == "check-001" + assert meta.CheckTitle == "Test Check" + assert meta.CheckType == ["type1"] + assert meta.ServiceName == "TestService" + assert meta.SubServiceName == "SubService" + assert meta.Severity == "high" + assert meta.ResourceType == "TestResource" + assert meta.Description == "A test check" + assert meta.Risk == "High risk" + assert meta.RelatedUrl == "http://example.com" + assert meta.Remediation.Recommendation.Text == "Fix it" + assert meta.Remediation.Recommendation.Url == "http://fix.com" + assert meta.Remediation.Code.NativeIaC == "iac_code" + assert meta.Remediation.Code.Terraform == "terraform_code" + assert meta.Remediation.Code.CLI == "cli_code" + assert meta.Remediation.Code.Other == "other_code" + assert meta.ResourceIdTemplate == "template" + assert meta.Categories == ["cat-one", "cat-two"] + assert meta.DependsOn == ["dep1"] + assert meta.RelatedTo == ["rel1"] + assert meta.Notes == "Some notes" + + # Check other Finding fields + assert finding_obj.uid == "finding-uid-1" + assert finding_obj.status == Status("FAIL") + assert finding_obj.status_extended == "extended" + # From the dummy resource + assert finding_obj.resource_uid == "res-uid-1" + assert finding_obj.resource_name == "ResourceName1" + assert finding_obj.resource_details == "" + # unroll_tags is called on a list with one tag -> expect {"env": "prod"} + assert finding_obj.resource_tags == {"env": "prod"} + assert finding_obj.region == "us-east-1" + # compliance is hardcoded to an empty dict + assert finding_obj.compliance == {} + + def test_transform_findings_stats_all_fails_muted(self): + """ + Test _transform_findings_stats when every failing finding is muted. + """ + # Create a dummy scan object with a unique_resource_count + dummy_scan = SimpleNamespace(unique_resource_count=10) + # Build summaries covering each severity branch. + ss1 = SimpleNamespace(_pass=1, fail=2, total=3, muted=2, severity="critical", scan=dummy_scan) + ss2 = SimpleNamespace(_pass=2, fail=0, total=2, muted=0, severity="high", scan=dummy_scan) + ss3 = SimpleNamespace(_pass=2, fail=3, total=5, muted=3, severity="medium", scan=dummy_scan) + ss4 = SimpleNamespace(_pass=3, fail=0, total=3, muted=0, severity="low", scan=dummy_scan) + + summaries = [ss1, ss2, ss3, ss4] + stats = Finding._transform_findings_stats(summaries) + + # Expected calculations: + # total_pass = 1+2+2+3 = 8 + # total_fail = 2+0+3+0 = 5 + # findings_count = 3+2+5+3 = 13 + # muted_pass = (ss1: 1) + (ss3: 2) = 3 + # muted_fail = (ss1: 2) + (ss3: 3) = 5 + expected = { + "total_pass": 8, + "total_muted_pass": 3, + "total_fail": 5, + "total_muted_fail": 5, + "resources_count": 10, + "findings_count": 13, + "total_critical_severity_fail": 2, + "total_critical_severity_pass": 1, + "total_high_severity_fail": 0, + "total_high_severity_pass": 2, + "total_medium_severity_fail": 3, + "total_medium_severity_pass": 2, + "total_low_severity_fail": 0, + "total_low_severity_pass": 3, + "all_fails_are_muted": True, # total_fail equals muted_fail and total_fail > 0 + } + assert stats == expected + + def test_transform_findings_stats_not_all_fails_muted(self): + """ + Test _transform_findings_stats when at least one failing finding is not muted. + """ + dummy_scan = SimpleNamespace(unique_resource_count=5) + # Build summaries: one summary has fail > 0 but muted == 0 + ss1 = SimpleNamespace(_pass=1, fail=2, total=3, muted=0, severity="critical", scan=dummy_scan) + ss2 = SimpleNamespace(_pass=2, fail=1, total=3, muted=1, severity="high", scan=dummy_scan) + summaries = [ss1, ss2] + stats = Finding._transform_findings_stats(summaries) + + # Expected calculations: + # total_pass = 1+2 = 3 + # total_fail = 2+1 = 3 + # findings_count = 3+3 = 6 + # muted_pass = (ss2: 2) since ss1 muted is 0 + # muted_fail = (ss2: 1) + # Severity breakdown: critical: pass 1, fail 2; high: pass 2, fail 1 + expected = { + "total_pass": 3, + "total_muted_pass": 2, + "total_fail": 3, + "total_muted_fail": 1, + "resources_count": 5, + "findings_count": 6, + "total_critical_severity_fail": 2, + "total_critical_severity_pass": 1, + "total_high_severity_fail": 1, + "total_high_severity_pass": 2, + "total_medium_severity_fail": 0, + "total_medium_severity_pass": 0, + "total_low_severity_fail": 0, + "total_low_severity_pass": 0, + "all_fails_are_muted": False, # 3 (total_fail) != 1 (muted_fail) + } + assert stats == expected + + def test_transform_api_finding_validation_error(self): + """ + Test that if required data is missing (causing a ValidationError) + the function logs the error and re-raises the exception. + For example, if the metadata dict is missing required keys. + """ + # Create a dummy API finding that is missing some required metadata + dummy_finding = DummyAPIFinding() + dummy_finding.inserted_at = 1234567890 + dummy_finding.scan = DummyScan(provider=DummyProvider(uid="account123")) + dummy_finding.uid = "finding-uid-invalid" + dummy_finding.status = "PASS" + dummy_finding.status_extended = "extended" + # Missing required metadata keys – using an empty dict + dummy_finding.check_metadata = {} + # Provide a dummy resources with a minimal resource + tag = DummyTag("env", "prod") + resource = DummyResource(uid="res-uid-1", name="ResourceName1", region="us-east-1", tags=[tag]) + dummy_finding.resources = DummyResources(resource) + + with pytest.raises(KeyError): + Finding.transform_api_finding(dummy_finding) diff --git a/tests/providers/aws/lib/s3/s3_test.py b/tests/providers/aws/lib/s3/s3_test.py index 2724e9749c..2cede7364d 100644 --- a/tests/providers/aws/lib/s3/s3_test.py +++ b/tests/providers/aws/lib/s3/s3_test.py @@ -113,7 +113,6 @@ class TestS3: csv_file = f"test{extension}" csv = CSV( findings=[FINDING], - create_file_descriptor=True, file_path=f"{CURRENT_DIRECTORY}/{csv_file}", )