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
8 changes: 8 additions & 0 deletions .github/workflows/cowork-auto-pr.yml
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,14 @@ jobs:
ensure-pr:
runs-on: ubuntu-latest
steps:
# gh pr create requires a local git checkout to diff head against base;
# without this step every run failed with "not a git repository" and no
# PR was ever opened (fleet-wide defect: 11/11 seeded copies lacked it).
- name: Check out the pushed branch
uses: actions/checkout@v4
with:
ref: ${{ github.ref_name }}
fetch-depth: 0
- name: Open PR for this branch if none exists
env:
GH_TOKEN: ${{ github.token }}
Expand Down
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ All notable changes to DeployDiff CLI will be documented in this file.
### Fixed

- GitHub Actions versions downgraded to stable v4/v5 (v6 caused workflow parse failures)
- All three parsers (Terraform, CloudFormation, Pulumi) now raise a clear `FileNotFoundError` when input is neither valid JSON nor an existing file path, instead of a cryptic `FileNotFoundError`/`PermissionError` from `open()`
- Removed dead `.get()` calls in terraform_parser and cloudformation_parser (orphaned `data.get("planned_values")`, `data.get("output_changes")`, `resource_change_data.get("Scope")`, `data.get("StackName")`) that looked like they were processing data but silently discarded results — silent-failure traps removed
- YAML indentation in CI workflows
- Git merge conflicts resolved in dependabot.yml, publish.yml, and pyproject.toml
- UTF-8 encoding (mojibake) in file output
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ classifiers = [
# tiers: pip install deploydiff[license]
license = [
"revenueholdings_license @ git+https://github.com/Coding-Dev-Tools/revenueholdings_license.git",
"revenueholdings-license>=0.1.0",
]
dev = [
"pytest>=7.0",
Expand Down
11 changes: 7 additions & 4 deletions src/deploydiff/cloudformation_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from .models import ChangeAction, ChangeSource, DeployPlan, ResourceChange
Expand Down Expand Up @@ -41,7 +42,12 @@ def parse_cloudformation_changeset(changeset_json: str | dict[str, Any]) -> Depl
try:
data = json.loads(changeset_json)
except json.JSONDecodeError:
with open(changeset_json) as f:
path = Path(changeset_json)
if not path.is_file():
raise FileNotFoundError(
f"Input is neither valid JSON nor an existing file: {changeset_json!r}"
) from None
with open(path) as f:
data = json.load(f)
else:
data = changeset_json
Expand Down Expand Up @@ -76,7 +82,6 @@ def parse_cloudformation_changeset(changeset_json: str | dict[str, Any]) -> Depl
)

# Scope details for update changes
resource_change_data.get("Scope", [])
details = resource_change_data.get("Details", [])

before = {}
Expand All @@ -100,8 +105,6 @@ def parse_cloudformation_changeset(changeset_json: str | dict[str, Any]) -> Depl
)
changes.append(resource_change)

data.get("StackName", data.get("ChangeSetName", "unknown"))

return DeployPlan(
source=ChangeSource.CLOUDFORMATION,
changes=changes,
Expand Down
8 changes: 3 additions & 5 deletions src/deploydiff/cost_estimator.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import copy
import json
from pathlib import Path

Expand Down Expand Up @@ -199,10 +200,7 @@ def _estimate_resource_cost(
# If deleting, after cost is 0; if creating, before cost is 0
if before and change.action == ChangeAction.CREATE:
return 0.0
if not before and change.action in (
ChangeAction.DELETE,
ChangeAction.DELETE_BEFORE_CREATE,
):
if not before and change.action == ChangeAction.DELETE:
return 0.0

resource_type = change.resource_type
Expand Down Expand Up @@ -251,7 +249,7 @@ def _load_pricing(
custom = json.load(f)

# Merge with defaults (custom overrides)
merged = DEFAULT_PRICING.copy()
merged = copy.deepcopy(DEFAULT_PRICING)
for resource_type, prices in custom.items():
if resource_type in merged:
merged[resource_type].update(prices)
Expand Down
7 changes: 6 additions & 1 deletion src/deploydiff/pulumi_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,12 @@ def parse_pulumi_preview(preview_json: str | dict[str, Any]) -> DeployPlan:
try:
data = json.loads(preview_json)
except json.JSONDecodeError:
with Path(preview_json).open() as f:
path = Path(preview_json)
if not path.is_file():
raise FileNotFoundError(
f"Input is neither valid JSON nor an existing file: {preview_json!r}"
) from None
with path.open() as f:
data = json.load(f)
else:
data = preview_json
Expand Down
12 changes: 6 additions & 6 deletions src/deploydiff/terraform_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,12 @@ def parse_terraform_plan(plan_json: str | dict[str, Any]) -> DeployPlan:
data = json.loads(plan_json)
except json.JSONDecodeError:
# Try as file path
with Path(plan_json).open() as f:
path = Path(plan_json)
if not path.is_file():
raise FileNotFoundError(
f"Input is neither valid JSON nor an existing file: {plan_json!r}"
) from None
with path.open() as f:
data = json.load(f)
else:
data = plan_json
Expand All @@ -43,7 +48,6 @@ def parse_terraform_plan(plan_json: str | dict[str, Any]) -> DeployPlan:
changes: list[ResourceChange] = []

# Parse planned changes
data.get("planned_values", {})
resource_changes = data.get("resource_changes", [])

for rc in resource_changes:
Expand Down Expand Up @@ -91,10 +95,6 @@ def parse_terraform_plan(plan_json: str | dict[str, Any]) -> DeployPlan:
)
changes.append(resource_change)

# Parse output changes
data.get("output_changes", {})
# We track these as metadata but don't create ResourceChange entries

return DeployPlan(
source=ChangeSource.TERRAFORM,
changes=changes,
Expand Down
21 changes: 20 additions & 1 deletion tests/test_edge_cases.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from rich.console import Console

from deploydiff.cli import _load_plan, _render_costs
from deploydiff.cost_estimator import DEFAULT_PRICING, _load_pricing
from deploydiff.cost_estimator import DEFAULT_PRICING, _load_pricing, estimate_costs
from deploydiff.models import (
ChangeAction,
ChangeSource,
Expand Down Expand Up @@ -94,6 +94,25 @@ def test_load_pricing_custom_type_not_in_defaults(self, tmp_path):
# Defaults should still be present
assert "t3.micro" in pricing["aws_instance"]

def test_delete_before_create_has_nonzero_after_cost(self):
"""DELETE_BEFORE_CREATE should report the new resource cost as after_cost."""
pricing = _load_pricing()
change = ResourceChange(
address="aws_instance.replaced",
action=ChangeAction.DELETE_BEFORE_CREATE,
resource_type="aws_instance",
resource_name="replaced",
source=ChangeSource.TERRAFORM,
before={"instance_type": "t3.micro"},
after={"instance_type": "t3.large"},
)
plan = DeployPlan(source=ChangeSource.TERRAFORM, changes=[change])
estimates = estimate_costs(plan)
assert len(estimates) == 1
est = estimates[0]
assert est.monthly_cost_before == pricing["aws_instance"]["t3.micro"]
assert est.monthly_cost_after == pricing["aws_instance"]["t3.large"]


class TestRollbackEdgeCases:
"""Tests for uncovered rollback paths."""
Expand Down
61 changes: 61 additions & 0 deletions tests/test_parse_errors.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
"""Tests for parser error handling: invalid input gives clear FileNotFoundError."""

from __future__ import annotations

import pytest

from deploydiff.cloudformation_parser import parse_cloudformation_changeset
from deploydiff.pulumi_parser import parse_pulumi_preview
from deploydiff.terraform_parser import parse_terraform_plan


class TestParserErrorHandling:
"""Each parser should raise a clear FileNotFoundError when input is neither
valid JSON nor an existing file path, instead of a cryptic exception."""

def test_terraform_invalid_string_raises_filenotfound(self):
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_terraform_plan("not-json-and-not-a-file")

def test_terraform_empty_string_raises_filenotfound(self):
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_terraform_plan("")

def test_cloudformation_invalid_string_raises_filenotfound(self):
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_cloudformation_changeset("not-json-and-not-a-file")

def test_pulumi_invalid_string_raises_filenotfound(self):
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_pulumi_preview("not-json-and-not-a-file")

def test_terraform_truncated_json_raises_filenotfound(self):
"""Truncated JSON (not a file, not parseable) should give clear error."""
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_terraform_plan('{"format_version":')

def test_cloudformation_empty_string_raises_filenotfound(self):
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_cloudformation_changeset("")

def test_pulumi_empty_string_raises_filenotfound(self):
with pytest.raises(FileNotFoundError, match="neither valid JSON nor an existing file"):
parse_pulumi_preview("")

def test_terraform_valid_dict_still_works(self):
"""Passing a dict directly should work as before."""
data = {"format_version": "1.2", "resource_changes": []}
plan = parse_terraform_plan(data)
assert len(plan.changes) == 0

def test_pulumi_valid_dict_still_works(self):
"""Passing a dict directly should work as before."""
data = {"steps": []}
plan = parse_pulumi_preview(data)
assert len(plan.changes) == 0

def test_cloudformation_valid_dict_still_works(self):
"""Passing a dict directly should work as before."""
data = {"Changes": []}
plan = parse_cloudformation_changeset(data)
assert len(plan.changes) == 0
Loading