test: add export unittests

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:

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:

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)

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)

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}",
         )