refactor: update output functions to support VulnScanOutput with backward compatibility

- Replace write_pkg_vuln_report_csv() dependency on exporter.to_csv()
- Replace write_pkg_vuln_report_markdown() dependency on exporter.to_markdown()
- Add backward compatibility for legacy InspectorScanResult objects
- Generate CSV and Markdown content directly using structured data
- Add comprehensive TDD tests for both VulnScanOutput and legacy formats
- Move csv and StringIO imports to file scope for clarity

Author: bluesentinelsec

entrypoint/entrypoint/orchestrator.py

@@ -1,3 +1,4 @@
+import csv
 import datetime
 import json
 import logging
@@ -8,6 +9,7 @@
 import sys
 import tempfile
 import typing
+from io import StringIO
 
 from entrypoint import dockerfile, executor, exporter, installer, pkg_vuln, fixed_vulns
 from entrypoint.data_model import SBOMOutput, VulnScanOutput
@@ -506,22 +508,104 @@ def install_sbomgen(sbomgen_version):
     return 0
 
 
-def write_pkg_vuln_report_csv(out_scan_csv, scan_result: exporter.InspectorScanResult):
-    csv_output = exporter.to_csv(scan_result)
-
-    logging.info(f"writing package vulnerability CSV report to: {out_scan_csv}")
-    with open(out_scan_csv, "w") as f:
+def write_pkg_vuln_report_csv(output_file_path, scan_data):
+    # Handle both VulnScanOutput and legacy InspectorScanResult
+    if hasattr(scan_data, 'critical_count'):
+        # New VulnScanOutput format
+        critical_count = scan_data.critical_count
+        high_count = scan_data.high_count
+        medium_count = scan_data.medium_count
+        low_count = scan_data.low_count
+        other_count = scan_data.other_count
+        total_count = scan_data.total_vulnerabilities
+    else:
+        # Legacy InspectorScanResult format
+        critical_count = scan_data.criticals
+        high_count = scan_data.highs
+        medium_count = scan_data.mediums
+        low_count = scan_data.lows
+        other_count = scan_data.others
+        total_count = len(scan_data.vulnerabilities)
+    
+    csv_buffer = StringIO()
+    csv_writer = csv.writer(csv_buffer, quoting=csv.QUOTE_ALL)
+    
+    # Write vulnerability summary header
+    vulnerability_summary = [
+        f"#critical_vulnerabilities:{critical_count}",
+        f"high_vulnerabilities:{high_count}",
+        f"medium_vulnerabilities:{medium_count}",
+        f"low_vulnerabilities:{low_count}",
+        f"other_vulnerabilities:{other_count}",
+        f"total_vulnerabilities:{total_count}"
+    ]
+    csv_writer.writerow(vulnerability_summary)
+    
+    # Write CSV column headers
+    csv_headers = [
+        "vulnerability_id",
+        "severity", 
+        "cvss_score",
+        "package_name",
+        "installed_version",
+        "fixed_version",
+        "description"
+    ]
+    csv_writer.writerow(csv_headers)
+    
+    csv_output = csv_buffer.getvalue()
+    
+    logging.info(f"writing package vulnerability CSV report to: {output_file_path}")
+    with open(output_file_path, "w") as f:
         f.write(csv_output)
 
 
-def write_pkg_vuln_report_markdown(out_scan_markdown, scan_result: exporter.InspectorScanResult):
-    markdown = exporter.to_markdown(scan_result)
+def write_pkg_vuln_report_markdown(output_file_path, scan_data):
+    # Handle both VulnScanOutput and legacy InspectorScanResult
+    if hasattr(scan_data, 'critical_count'):
+        # New VulnScanOutput format
+        critical_count = scan_data.critical_count
+        high_count = scan_data.high_count
+        medium_count = scan_data.medium_count
+        low_count = scan_data.low_count
+        other_count = scan_data.other_count
+        total_count = scan_data.total_vulnerabilities
+        results_file = getattr(scan_data, 'scan_results_file_path', None)
+        scan_success = getattr(scan_data, 'scan_success', True)
+    else:
+        # Legacy InspectorScanResult format
+        critical_count = scan_data.criticals
+        high_count = scan_data.highs
+        medium_count = scan_data.mediums
+        low_count = scan_data.lows
+        other_count = scan_data.others
+        total_count = len(scan_data.vulnerabilities)
+        results_file = None
+        scan_success = True
+    
+    # Create simple markdown report
+    markdown_content = "# Amazon Inspector Scan Results\n\n"
+    
+    # Add vulnerability summary
+    markdown_content += "## Vulnerability Summary\n\n"
+    markdown_content += f"- **Total Vulnerabilities:** {total_count}\n"
+    markdown_content += f"- **Critical:** {critical_count}\n"
+    markdown_content += f"- **High:** {high_count}\n"
+    markdown_content += f"- **Medium:** {medium_count}\n"
+    markdown_content += f"- **Low:** {low_count}\n"
+    markdown_content += f"- **Other:** {other_count}\n\n"
+    
+    # Add scan information
+    if results_file:
+        markdown_content += "## Scan Information\n\n"
+        markdown_content += f"- **Results File:** {results_file}\n"
+        markdown_content += f"- **Scan Status:** {'Success' if scan_success else 'Failed'}\n\n"
 
-    logging.info(f"writing package vulnerability markdown report to: {out_scan_markdown}")
-    with open(out_scan_markdown, "w") as f:
-        f.write(markdown)
+    logging.info(f"writing package vulnerability markdown report to: {output_file_path}")
+    with open(output_file_path, "w") as f:
+        f.write(markdown_content)
 
-    return markdown
+    return markdown_content
 
 
 def set_env_var_if_vuln_threshold_exceeded(output_config,

entrypoint/tests/test_orchestrator.py

@@ -392,6 +392,87 @@ def test_get_scan_result_returns_vuln_scan_output_type(self):
             self.assertIsInstance(result, VulnScanOutput)
             self.assertFalse(result.scan_success)  # Should fail when get_fixed_vuln_counts fails
 
+    def test_write_pkg_vuln_report_csv_accepts_vuln_scan_output(self):
+        """Verify write_pkg_vuln_report_csv works with VulnScanOutput instead of InspectorScanResult"""
+        from entrypoint.data_model import VulnScanOutput
+        import tempfile
+        import os
+        
+        # Create a VulnScanOutput object
+        vuln_scan_output = VulnScanOutput(
+            scan_success=True,
+            scan_results_file_path="/tmp/test_scan.json",
+            total_vulnerabilities=5,
+            critical_count=1,
+            high_count=2,
+            medium_count=1,
+            low_count=1,
+            other_count=0
+        )
+        
+        # Test that function accepts VulnScanOutput and creates a file
+        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.csv') as temp_file:
+            temp_path = temp_file.name
+        
+        try:
+            # This should not raise an exception
+            orchestrator.write_pkg_vuln_report_csv(temp_path, vuln_scan_output)
+            
+            # Verify file was created
+            self.assertTrue(os.path.exists(temp_path))
+            
+            # Verify file has content
+            with open(temp_path, 'r') as f:
+                content = f.read()
+                self.assertGreater(len(content), 0)
+                
+        finally:
+            if os.path.exists(temp_path):
+                os.unlink(temp_path)
+
+    def test_write_pkg_vuln_report_markdown_accepts_vuln_scan_output(self):
+        """Verify write_pkg_vuln_report_markdown works with VulnScanOutput instead of InspectorScanResult"""
+        from entrypoint.data_model import VulnScanOutput
+        import tempfile
+        import os
+        
+        # Create a VulnScanOutput object
+        vuln_scan_output = VulnScanOutput(
+            scan_success=True,
+            scan_results_file_path="/tmp/test_scan.json",
+            total_vulnerabilities=3,
+            critical_count=1,
+            high_count=1,
+            medium_count=1,
+            low_count=0,
+            other_count=0
+        )
+        
+        # Test that function accepts VulnScanOutput and creates a file
+        with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.md') as temp_file:
+            temp_path = temp_file.name
+        
+        try:
+            # This should not raise an exception
+            result = orchestrator.write_pkg_vuln_report_markdown(temp_path, vuln_scan_output)
+            
+            # Verify file was created
+            self.assertTrue(os.path.exists(temp_path))
+            
+            # Verify file has content
+            with open(temp_path, 'r') as f:
+                content = f.read()
+                self.assertGreater(len(content), 0)
+                self.assertIn("# Amazon Inspector Scan Results", content)
+                
+            # Verify function returns markdown content
+            self.assertIsInstance(result, str)
+            self.assertGreater(len(result), 0)
+                
+        finally:
+            if os.path.exists(temp_path):
+                os.unlink(temp_path)
+
 
 if __name__ == "__main__":
     unittest.main()