Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,8 @@ Compare deployment configurations across environments. Detect drift between stag

## Build & Test Commands
- Install (editable, from this repo): `pip install -e .`
- Install (prebuilt wheel from the self-hosted index): `pip install --index-url https://coding-dev-tools.github.io/pypi-index/simple/ deploydiff`
- Install (from source): `pip install git+https://github.com/Coding-Dev-Tools/deploydiff.git`
- NOTE: `deploydiff` is NOT on public PyPI — use the self-hosted index or a `git+` URL above.
- NOTE: `deploydiff` is NOT on public PyPI and the self-hosted pypi-index is unavailable (returns 404, verified dead 2026-08-21 Run 217). Use the `git+` form above — it is the only verified-working pip install.
- Test: `pytest tests/` (or `python -m pytest tests/ -v --tb=short`)
- Lint: `ruff check .`
- Build: `pip install build twine && python -m build && twine check dist/*`
Expand Down
55 changes: 40 additions & 15 deletions src/deploydiff/rollback.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

from __future__ import annotations

from .models import ChangeSource, DeployPlan
from .models import ChangeAction, ChangeSource, DeployPlan


def generate_rollback_commands(plan: DeployPlan) -> list[str]:
Expand All @@ -14,6 +14,11 @@ def generate_rollback_commands(plan: DeployPlan) -> list[str]:
Returns:
List of rollback commands as strings.
"""
if all(
change.action in (ChangeAction.NO_OP, ChangeAction.READ)
for change in plan.changes
):
return ["# No changes to roll back"]
if plan.source == ChangeSource.TERRAFORM:
return _terraform_rollback(plan)
elif plan.source == ChangeSource.CLOUDFORMATION:
Expand All @@ -28,17 +33,47 @@ def _terraform_rollback(plan: DeployPlan) -> list[str]:

Strategy: target the reverse of each destructive/create change.
"""
if not plan.changes:
return ["# No changes to roll back"]

commands: list[str] = []
commands.append("# Terraform Rollback Commands")
commands.append("# Run these in reverse order to undo the deployment")
commands.append("")

# For each create, we need to destroy it
for change in plan.creates:
# Replacements (create-before-delete / delete-before-create): revert by
# re-applying the PREVIOUS config. Do NOT also emit destroy + apply for
# these -- they used to appear in both the creates and destructive lists,
# producing contradictory commands for the same resource.
replacements = [
c
for c in plan.destructive_changes
if c.action
in (
ChangeAction.CREATE_BEFORE_DELETE,
ChangeAction.DELETE_BEFORE_CREATE,
ChangeAction.REPLACE,
)
]

# For each pure create, we need to destroy it
pure_creates = [c for c in plan.creates if c.action == ChangeAction.CREATE]
for change in pure_creates:
commands.append(f"terraform destroy -target={change.address} -auto-approve")

# For each destructive change (delete/replace), we need to re-apply it
for change in plan.destructive_changes:
# For each pure delete, we need to re-create it from the previous config
pure_deletes = [c for c in plan.destructive_changes if c.action == ChangeAction.DELETE]
for change in pure_deletes:
commands.append(
f"# To restore {change.address}, restore previous config and run:"
)
commands.append(f"terraform apply -target={change.address} -auto-approve")

# For each replacement, revert with the previous config
for change in replacements:
commands.append(
f"# To revert replaced {change.address}, restore previous config and run:"
)
commands.append(f"terraform apply -target={change.address} -auto-approve")

# For updates, we can try to revert with the previous state
Expand All @@ -48,16 +83,6 @@ def _terraform_rollback(plan: DeployPlan) -> list[str]:
)
commands.append(f"terraform apply -target={change.address} -auto-approve")

if not plan.changes:
commands.append("# No changes to roll back")

# Add a full rollback option
commands.append("")
commands.append("# Or rollback the entire stack:")
commands.append("terraform apply -auto-approve # with previous .tf files")
commands.append("# OR destroy everything and re-apply from a known good state:")
commands.append("terraform destroy -auto-approve && terraform apply -auto-approve")

return commands


Expand Down
5 changes: 3 additions & 2 deletions tests/test_edge_cases.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,8 +143,9 @@ def test_pulumi_rollback_unsupported_source_fallback(self):
# all produce meaningful output.
plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[])
cmds = generate_rollback_commands(plan)
assert len(cmds) > 1
assert "Terraform" in cmds[0]
# Empty plans short-circuit: no header, no blanket destroy-everything
# suggestion for a plan with nothing to roll back.
assert cmds == ["# No changes to roll back"]

def test_cloudformation_rollback_no_raw_data(self):
"""_cloudformation_rollback with no raw_data uses STACK_NAME."""
Expand Down
70 changes: 70 additions & 0 deletions tests/test_rollback_safety.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
"""Regression tests for rollback command generation safety/correctness."""

from deploydiff.models import (
ChangeAction,
ChangeSource,
DeployPlan,
ResourceChange,
)
from deploydiff.rollback import generate_rollback_commands


def _tf_change(address, action):
return ResourceChange(
address=address,
action=action,
resource_type="aws_instance",
resource_name=address.split(".")[-1],
source=ChangeSource.TERRAFORM,
)


def test_empty_plan_returns_only_noop_message():
plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[])
commands = generate_rollback_commands(plan)
assert commands == ["# No changes to roll back"]
# The dangerous blanket destroy-everything suggestion must not appear.
assert not any("destroy -auto-approve &&" in c for c in commands)


def test_read_and_noop_only_plans_return_noop_message():
for action in (ChangeAction.READ, ChangeAction.NO_OP):
plan = DeployPlan(
source=ChangeSource.TERRAFORM,
changes=[_tf_change("data.aws_ami.current", action)],
)
assert generate_rollback_commands(plan) == ["# No changes to roll back"]


def test_create_before_delete_not_double_commanded():
"""A create-first replacement must not produce both destroy and apply
for the same resource (contradictory rollback commands)."""
change = _tf_change("aws_instance.web", ChangeAction.CREATE_BEFORE_DELETE)
plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[change])
commands = generate_rollback_commands(plan)
destroys = [c for c in commands if c.startswith("terraform destroy -target=aws_instance.web")]
applies = [c for c in commands if c.startswith("terraform apply -target=aws_instance.web")]
assert destroys == [], "replacement should not be destroyed on rollback"
assert len(applies) == 1


def test_delete_before_create_and_replace_use_one_restore_command():
for action in (ChangeAction.DELETE_BEFORE_CREATE, ChangeAction.REPLACE):
change = _tf_change("aws_instance.web", action)
commands = generate_rollback_commands(
DeployPlan(source=ChangeSource.TERRAFORM, changes=[change])
)
restores = [c for c in commands if "-target=aws_instance.web" in c]
assert len(restores) == 1
assert not any("destroy -target=aws_instance.web" in c for c in commands)


def test_pure_create_gets_destroy_pure_delete_gets_apply():
created = _tf_change("aws_instance.new", ChangeAction.CREATE)
deleted = _tf_change("aws_instance.old", ChangeAction.DELETE)
plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[created, deleted])
commands = generate_rollback_commands(plan)
assert "terraform destroy -target=aws_instance.new -auto-approve" in commands
assert any(
c.startswith("terraform apply -target=aws_instance.old") for c in commands
)
Loading