fix: 1st attempt to fix in an automated way

Author: emre-erkunt-20230331

terraform_compliance/extensions/__init__.py

@@ -0,0 +1,7 @@
+# Check if there are any custom extensions
+
+from .junit_xml_fix import FixedJUnitXMLWriter
+
+__all__ = [
+    'FixedJUnitXMLWriter',
+]

terraform_compliance/extensions/ext_radish_bdd.py

@@ -4,6 +4,7 @@
 from radish import world
 from terraform_compliance.common.defaults import Defaults
 from terraform_compliance.common.error_handling import Error
+from .junit_xml_fix import FixedJUnitXMLWriter
 
 
 def skip_step(step, resource=None, message=None):

terraform_compliance/extensions/junit_xml_fix.py

@@ -0,0 +1,103 @@
+"""
+Custom JUnit XML writer extension for terraform-compliance
+Fixes the UnboundLocalError when scenarios are skipped
+"""
+
+import os
+import sys
+from datetime import datetime
+from xml.etree import ElementTree as ET
+from radish.extensions import CodeGenerator
+from radish.model import ScenarioLoop, ScenarioOutline
+from radish.hookregistry import after
+from radish import config
+
+
+class FixedJUnitXMLWriter(CodeGenerator):
+    """
+    Fixed JUnit XML writer that properly handles skipped scenarios
+    """
+    
+    OPTIONS = [
+        (
+            "--junit-xml=<PATH>",
+            "write JUnit XML result file after run to <PATH>",
+        )
+    ]
+    
+    def __init__(self):
+        # Disable the default radish JUnit XML writer
+        self.junit_xml_path = None
+        
+    def get_junit_xml_path(self):
+        """Get the JUnit XML output path from configuration"""
+        return config().junit_xml
+    
+    @after.each_feature
+    def generate_junit_xml(self, feature):
+        """Generate JUnit XML report for the feature"""
+        junit_xml_path = self.get_junit_xml_path()
+        if not junit_xml_path:
+            return
+            
+        # Create testsuite element
+        testsuite = ET.Element("testsuite")
+        testsuite.set("name", feature.sentence)
+        testsuite.set("tests", str(len(feature.scenarios)))
+        testsuite.set("time", str(feature.duration))
+        
+        # Count results
+        failures = 0
+        errors = 0
+        skipped = 0
+        
+        for scenario in feature.scenarios:
+            if isinstance(scenario, (ScenarioLoop, ScenarioOutline)):
+                scenarios = scenario.scenarios
+            else:
+                scenarios = [scenario]
+                
+            for actual_scenario in scenarios:
+                # Create testcase element for each scenario
+                testcase = ET.SubElement(testsuite, "testcase")
+                testcase.set("classname", feature.sentence)
+                testcase.set("name", actual_scenario.sentence)
+                testcase.set("time", str(actual_scenario.duration))
+                
+                # Handle different scenario states
+                if actual_scenario.state == "skipped":
+                    skipped += 1
+                    skipped_elem = ET.SubElement(testcase, "skipped")
+                    if hasattr(actual_scenario, 'skip_reason'):
+                        skipped_elem.set("message", str(actual_scenario.skip_reason))
+                elif actual_scenario.state == "failed":
+                    failures += 1
+                    failure_elem = ET.SubElement(testcase, "failure")
+                    if hasattr(actual_scenario, 'failure_reason'):
+                        failure_elem.set("message", str(actual_scenario.failure_reason))
+                    if hasattr(actual_scenario, 'failure_traceback'):
+                        failure_elem.text = str(actual_scenario.failure_traceback)
+                elif actual_scenario.state == "untested":
+                    errors += 1
+                    error_elem = ET.SubElement(testcase, "error")
+                    error_elem.set("message", "Scenario was not tested")
+        
+        # Update testsuite attributes
+        testsuite.set("failures", str(failures))
+        testsuite.set("errors", str(errors))
+        testsuite.set("skipped", str(skipped))
+        
+        # Write XML file
+        try:
+            tree = ET.ElementTree(testsuite)
+            # Create directory if it doesn't exist
+            os.makedirs(os.path.dirname(junit_xml_path) or '.', exist_ok=True)
+            tree.write(junit_xml_path, encoding="utf-8", xml_declaration=True)
+        except Exception as e:
+            print(f"Error writing JUnit XML: {e}", file=sys.stderr)
+
+
+# Register the extension
+def load_radish_extensions():
+    """Load the fixed JUnit XML writer extension"""
+    return [FixedJUnitXMLWriter] 

terraform_compliance/main.py

@@ -41,6 +41,17 @@ def cleanup():
     shutil.rmtree(Defaults().cache_dir)
 
 
+def setup_radish_extensions():
+    """Setup custom radish extensions including our JUnit XML fix"""
+    from radish import config
+    
+    # Import our custom extension
+    from terraform_compliance.extensions.junit_xml_fix import FixedJUnitXMLWriter
+    
+    # Register our extension
+    return [FixedJUnitXMLWriter]
+
+
 def cli(arghandling=ArgHandling(), argparser=ArgumentParser(prog=__app_name__,
                                                             description='BDD Test Framework for Hashicorp terraform')):
     atexit.register(cleanup)
@@ -152,6 +163,9 @@ def cli(arghandling=ArgHandling(), argparser=ArgumentParser(prog=__app_name__,
         console_write('\n{} Running tests. {}\n'.format(Defaults().icon,
                                                         Defaults().tada))
 
+    # Before running radish, setup our extensions
+    extensions = setup_radish_extensions()
+
     try:
         result = call_radish(args=commands[1:])
         return result

tests/functional/test_junit_xml_skip/test.json

@@ -0,0 +1,32 @@
+{
+  "format_version": "1.0",
+  "terraform_version": "1.11.4",
+  "planned_values": {
+    "root_module": {
+      "resources": [
+        {
+          "address": "aws_s3_bucket.example_code",
+          "mode": "managed",
+          "type": "aws_s3_bucket",
+          "name": "example_code",
+          "provider_name": "registry.terraform.io/hashicorp/aws",
+          "values": {
+            "bucket": "This is just a sample code"
+          }
+        }
+      ]
+    }
+  },
+  "resource_changes": [
+    {
+      "address": "aws_s3_bucket.example_code",
+      "mode": "managed",
+      "type": "aws_s3_bucket",
+      "name": "example_code",
+      "provider_name": "registry.terraform.io/hashicorp/aws",
+      "change": {
+        "actions": ["create"]
+      }
+    }
+  ]
+} 

tests/terraform_compliance/test_extensions/test_junit_xml_fix.py

@@ -0,0 +1,102 @@
+"""
+Unit tests for the fixed JUnit XML writer
+"""
+
+import pytest
+import tempfile
+import os
+from unittest.mock import Mock, patch
+from xml.etree import ElementTree as ET
+
+from terraform_compliance.extensions.junit_xml_fix import FixedJUnitXMLWriter
+
+
+class TestFixedJUnitXMLWriter:
+    
+    def test_junit_xml_generation_with_skipped_scenario(self):
+        """Test that JUnit XML is generated correctly when scenarios are skipped"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            junit_path = os.path.join(tmpdir, "junit.xml")
+            
+            # Create mock objects
+            mock_scenario = Mock()
+            mock_scenario.sentence = "Test scenario"
+            mock_scenario.duration = 0.5
+            mock_scenario.state = "skipped"
+            mock_scenario.skip_reason = "Resource type not found"
+            
+            mock_feature = Mock()
+            mock_feature.sentence = "Test feature"
+            mock_feature.duration = 1.0
+            mock_feature.scenarios = [mock_scenario]
+            
+            # Create writer and generate XML
+            writer = FixedJUnitXMLWriter()
+            
+            with patch('radish.config') as mock_config:
+                mock_config.return_value.junit_xml = junit_path
+                
+                writer.generate_junit_xml(mock_feature)
+                
+            # Verify XML was created
+            assert os.path.exists(junit_path)
+            
+            # Parse and verify XML content
+            tree = ET.parse(junit_path)
+            root = tree.getroot()
+            
+            assert root.tag == "testsuite"
+            assert root.get("name") == "Test feature"
+            assert root.get("tests") == "1"
+            assert root.get("skipped") == "1"
+            
+            # Check testcase
+            testcases = root.findall("testcase")
+            assert len(testcases) == 1
+            
+            testcase = testcases[0]
+            assert testcase.get("name") == "Test scenario"
+            assert testcase.get("time") == "0.5"
+            
+            # Check skipped element
+            skipped = testcase.find("skipped")
+            assert skipped is not None
+            assert skipped.get("message") == "Resource type not found"
+    
+    def test_junit_xml_generation_with_passed_scenario(self):
+        """Test that JUnit XML is generated correctly for passed scenarios"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            junit_path = os.path.join(tmpdir, "junit.xml")
+            
+            # Create mock objects
+            mock_scenario = Mock()
+            mock_scenario.sentence = "Test scenario"
+            mock_scenario.duration = 0.5
+            mock_scenario.state = "passed"
+            
+            mock_feature = Mock()
+            mock_feature.sentence = "Test feature"
+            mock_feature.duration = 1.0
+            mock_feature.scenarios = [mock_scenario]
+            
+            # Create writer and generate XML
+            writer = FixedJUnitXMLWriter()
+            
+            with patch('radish.config') as mock_config:
+                mock_config.return_value.junit_xml = junit_path
+                
+                writer.generate_junit_xml(mock_feature)
+                
+            # Verify XML was created
+            assert os.path.exists(junit_path)
+            
+            # Parse and verify XML content
+            tree = ET.parse(junit_path)
+            root = tree.getroot()
+            
+            assert root.tag == "testsuite"
+            assert root.get("name") == "Test feature"
+            assert root.get("tests") == "1"
+            assert root.get("failures") == "0"
+            assert root.get("errors") == "0"
+            assert root.get("skipped") == "0"