Ensure the analyzer and reporter can handle with action as a unique replace action.

Author: rafaelherik

Taskfile.yml

@@ -17,7 +17,7 @@ tasks:
   test:
     desc: Run all tests with pytest
     cmds:
-      - . .venv/bin/activate && pytest
+      - . .venv/bin/activate && poetry run pytest --cov=tfsumpy --cov-report=term && poetry run coverage report --fail-under=75
     deps: [install]
 
   lint:

tests/plan/test_plan_analyzer.py

@@ -171,4 +171,97 @@ def test_debug_logging(self, mock_debug, analyzer, sample_plan_json):
 
         with patch("builtins.open", mock_open(read_data=plan_content)):
             analyzer.analyze(Mock(), plan_path="test.tfplan")
-            assert mock_debug.called 
+            assert mock_debug.called 
+
+    def test_replacement_detection(self, analyzer):
+        """Test detection of replacement (delete+create) and extraction of triggers."""
+        plan_json = {
+            "resource_changes": [
+                {
+                    "address": "aws_instance.server",
+                    "type": "aws_instance",
+                    "change": {
+                        "actions": ["delete", "create"],
+                        "before": {"instance_type": "t2.micro"},
+                        "after": {"instance_type": "t2.small"},
+                        "replacement_triggered_by": ["instance_type"],
+                        "before_sensitive": {}
+                    }
+                }
+            ]
+        }
+        changes = analyzer._parse_plan(json.dumps(plan_json))
+        assert len(changes) == 1
+        change = changes[0]
+        assert change.action == "replace"
+        assert change.replacement is True
+        assert change.replacement_triggers == ["instance_type"]
+        assert change.before == {"instance_type": "t2.micro"}
+        assert change.after == {"instance_type": "t2.small"}
+
+    def test_replacement_no_triggers(self, analyzer):
+        """Test replacement detection when no replacement_triggered_by is present."""
+        plan_json = {
+            "resource_changes": [
+                {
+                    "address": "aws_instance.server",
+                    "type": "aws_instance",
+                    "change": {
+                        "actions": ["delete", "create"],
+                        "before": {"instance_type": "t2.micro"},
+                        "after": {"instance_type": "t2.small"},
+                        "before_sensitive": {}
+                    }
+                }
+            ]
+        }
+        changes = analyzer._parse_plan(json.dumps(plan_json))
+        assert len(changes) == 1
+        change = changes[0]
+        assert change.action == "replace"
+        assert change.replacement is True
+        assert change.replacement_triggers == []
+
+    def test_regular_create_update_delete(self, analyzer):
+        """Test that create, update, and delete actions are not marked as replacement."""
+        plan_json = {
+            "resource_changes": [
+                {
+                    "address": "aws_s3_bucket.data",
+                    "type": "aws_s3_bucket",
+                    "change": {
+                        "actions": ["create"],
+                        "before": None,
+                        "after": {"bucket": "test-bucket"},
+                        "before_sensitive": {}
+                    }
+                },
+                {
+                    "address": "aws_instance.server",
+                    "type": "aws_instance",
+                    "change": {
+                        "actions": ["delete"],
+                        "before": {"instance_type": "t2.micro"},
+                        "after": None,
+                        "before_sensitive": {}
+                    }
+                },
+                {
+                    "address": "aws_instance.web",
+                    "type": "aws_instance",
+                    "change": {
+                        "actions": ["update"],
+                        "before": {"instance_type": "t2.micro"},
+                        "after": {"instance_type": "t2.small"},
+                        "before_sensitive": {}
+                    }
+                }
+            ]
+        }
+        changes = analyzer._parse_plan(json.dumps(plan_json))
+        assert len(changes) == 3
+        for c in changes:
+            if c.action == "replace":
+                assert False, "Should not detect replace for create, update, or delete"
+            assert c.replacement is False
+            assert c.replacement_triggers == [] 

tfsumpy/plan/analyzer.py

@@ -55,7 +55,14 @@ def analyze(self, context: 'Context', **kwargs) -> AnalyzerResult:
             # Generate summary statistics
             change_counts = {'create': 0, 'update': 0, 'delete': 0}
             for change in changes:
-                change_counts[change.action] += 1
+                # For reporting, treat 'replace' as both a delete and a create
+                if change.action == 'replace':
+                    change_counts['delete'] += 1
+                    change_counts['create'] += 1
+                elif change.action in change_counts:
+                    change_counts[change.action] += 1
+                else:
+                    change_counts[change.action] = 1  # fallback for unexpected actions
 
             self.logger.info(f"Found {len(changes)} resource changes")
             self.logger.debug(f"Change breakdown: {change_counts}")
@@ -97,9 +104,10 @@ def _parse_plan(self, plan_content: str) -> List[ResourceChange]:
 
         for change in resource_changes:
             # Extract change action
-            action = change.get('change', {}).get('actions', ['no-op'])[0]
+            actions = change.get('change', {}).get('actions', ['no-op'])
+            action = actions[0] if actions else 'no-op'
             if action != 'no-op':
-                self.logger.debug(f"Processing {action} change for {change.get('address', '')}")
+                self.logger.debug(f"Processing {actions} change for {change.get('address', '')}")
 
                 # Extract module information
                 address = change.get('address', '')
@@ -108,14 +116,22 @@ def _parse_plan(self, plan_content: str) -> List[ResourceChange]:
                 # Get change details
                 change_details = change.get('change', {})
 
+                # Detect replacement (Terraform: actions == ["delete", "create"])
+                is_replacement = actions == ["delete", "create"]
+                replacement_triggers = change_details.get('replacement_triggered_by', []) if is_replacement else []
+                # For reporting, treat as 'replace' action
+                action_display = 'replace' if is_replacement else action
+                
                 changes.append(ResourceChange(
-                    action=action,
+                    action=action_display,
                     resource_type=change.get('type', ''),
                     identifier=self._sanitize_text(address),
                     changes=change_details.get('before_sensitive', {}),
                     module=module_name,
                     before=change_details.get('before', {}),
-                    after=change_details.get('after', {})
+                    after=change_details.get('after', {}),
+                    replacement=is_replacement,
+                    replacement_triggers=replacement_triggers
                 ))
 
         return changes

tfsumpy/plan/reporter.py

@@ -67,7 +67,10 @@ def _process_resources(self, resources: List[Dict], show_changes: bool = False,
                 'identifier': resource['identifier'],
                 'action': resource['action'],
                 'provider': resource.get('provider', 'unknown'),
-                'module': resource.get('module', 'root')
+                'module': resource.get('module', 'root'),
+                # Add replacement and triggers for reporting
+                'replacement': resource.get('replacement', False),
+                'replacement_triggers': resource.get('replacement_triggers', []),
             }
 
             # Process changes if requested
@@ -102,7 +105,9 @@ def _process_resources(self, resources: List[Dict], show_changes: bool = False,
                     'raw': {
                         'before': resource.get('before', {}),
                         'after': resource.get('after', {})
-                    }
+                    },
+                    # Add replacement triggers to details if present
+                    'replacement_triggers': resource.get('replacement_triggers', [])
                 }
 
             processed_resources.append(resource_data)
@@ -140,9 +145,9 @@ def _print_summary(self, report: Dict) -> None:
         """Format the change summary section."""
         self._write(f"\n{self._colorize('Total Changes: ' + str(report['total_changes']), 'bold')}\n")
 
-        # Add change counts by type
+        # Add change counts by type (do not show 'replace' in summary)
         for action in ['create', 'update', 'delete']:
-            count = report['change_breakdown'][action]
+            count = report['change_breakdown'].get(action, 0)
             self._write(f"{action.title()}: {count}\n")
 
     def _print_resource_details(self, resources: list, show_changes: bool = False) -> None:
@@ -153,7 +158,8 @@ def _print_resource_details(self, resources: list, show_changes: bool = False) -
         action_colors = {
             'CREATE': 'green',
             'UPDATE': 'blue',
-            'DELETE': 'red'
+            'DELETE': 'red',
+            'REPLACE': 'yellow',
         }
 
         for resource in resources:
@@ -164,9 +170,14 @@ def _print_resource_details(self, resources: list, show_changes: bool = False) -
                 f"\n{colored_action} {resource['resource_type']}: "
                 f"{resource['identifier']}\n"
             )
-            
+            # Show replacement triggers if this is a replacement
+            if resource.get('replacement', False) and resource.get('replacement_triggers'):
+                triggers = ', '.join(resource['replacement_triggers'])
+                self._write(f"  Replacement triggered by: {triggers}\n")
             if show_changes:
                 self._print_attribute_changes(resource)
+        # Ensure a newline at the end to avoid shell prompt artifacts
+        self._write("\n")
 
     def _print_attribute_changes(self, resource: Dict) -> None:
         """Format attribute changes for a resource."""
@@ -182,7 +193,8 @@ def _print_attribute_changes(self, resource: Dict) -> None:
         symbol_colors = {
             '+': 'green',   # create
             '~': 'blue',    # update
-            '-': 'red'      # delete
+            '-': 'red',     # delete
+            '-/+': 'yellow' # replace
         }
 
         for attr in sorted(all_attrs - skip_attrs):
@@ -196,9 +208,12 @@ def _print_attribute_changes(self, resource: Dict) -> None:
                 elif resource['action'] == 'delete':
                     symbol = self._colorize('-', symbol_colors['-'])
                     lines.append(f"  {symbol} {attr} = {before_val}")
-                else:  # update
+                elif resource['action'] == 'update':
                     symbol = self._colorize('~', symbol_colors['~'])
                     lines.append(f"  {symbol} {attr} = {before_val} -> {after_val}")
+                elif resource['action'] == 'replace':
+                    symbol = self._colorize('-/+', symbol_colors['-/+'])
+                    lines.append(f"  {symbol} {attr} = {before_val} -> {after_val}")
 
         self._write('\n'.join(lines))
 

tfsumpy/reporters/base_reporter.py

@@ -14,6 +14,7 @@ class BaseReporter:
         'green': Fore.GREEN,
         'blue': Fore.BLUE,
         'red': Fore.RED,
+        'yellow': Fore.YELLOW,
         'reset': Style.RESET_ALL,
         'bold': Style.BRIGHT
     }

tfsumpy/resource.py

@@ -11,4 +11,6 @@ class ResourceChange:
     changes: Dict[str, Any] = field(default_factory=dict)
     module: str = "root"
     before: Dict[str, Any] = field(default_factory=dict)
-    after: Dict[str, Any] = field(default_factory=dict)
+    after: Dict[str, Any] = field(default_factory=dict)
+    replacement: bool = False
+    replacement_triggers: list = field(default_factory=list)

tfsumpy/templates/plan_report.md

@@ -9,6 +9,10 @@
 ## Resource Changes
 {% for resource in resources %}
 ### {{ resource.resource_type }}.{{ resource.identifier }}
+**Action:** {{ resource.action | upper }}
+{% if resource.replacement and resource.replacement_triggers %}
+- **Replacement triggered by:** {{ resource.replacement_triggers | join(', ') }}
+{% endif %}
 {% if resource.changes %}
 #### Changes:
 {% for change in resource.changes %}