632ac9e328
- Python CLI tool for rolling updates of CL-AppPipe-* and CL-SvcPipe-* stacks - Async update engine with configurable concurrency (asyncio.Semaphore) - Exponential backoff retry for API throttling - Dry-run mode for safe preview - IAM permission pre-validation - Comprehensive test suite (80 tests: 11 property-based + 69 unit) - Full spec documentation (requirements, design, tasks)
217 lines
8.1 KiB
Python
217 lines
8.1 KiB
Python
"""Unit tests for CLI argument parsing and pipeline orchestration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from cfn_updater.cli import main, parse_args
|
|
from cfn_updater.config import DEFAULT_CONCURRENCY, DEFAULT_PREFIX
|
|
from cfn_updater.models import DiscoveredStack, StackUpdateResult
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Argument parsing tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestParseArgs:
|
|
"""Tests for parse_args() default and custom values."""
|
|
|
|
def test_default_values(self):
|
|
args = parse_args([])
|
|
assert args.prefix is None # None means "all configured prefixes"
|
|
assert args.concurrency == DEFAULT_CONCURRENCY
|
|
assert args.dry_run is False
|
|
assert args.region is None
|
|
assert args.profile is None
|
|
|
|
def test_custom_prefix(self):
|
|
args = parse_args(["--prefix", "MyStack-"])
|
|
assert args.prefix == "MyStack-"
|
|
|
|
def test_custom_concurrency(self):
|
|
args = parse_args(["--concurrency", "10"])
|
|
assert args.concurrency == 10
|
|
|
|
def test_dry_run_flag(self):
|
|
args = parse_args(["--dry-run"])
|
|
assert args.dry_run is True
|
|
|
|
def test_region_flag(self):
|
|
args = parse_args(["--region", "eu-west-1"])
|
|
assert args.region == "eu-west-1"
|
|
|
|
def test_profile_flag(self):
|
|
args = parse_args(["--profile", "audit"])
|
|
assert args.profile == "audit"
|
|
|
|
def test_all_custom_values(self):
|
|
args = parse_args([
|
|
"--prefix", "Test-",
|
|
"--concurrency", "3",
|
|
"--dry-run",
|
|
"--region", "us-east-1",
|
|
"--profile", "prod",
|
|
])
|
|
assert args.prefix == "Test-"
|
|
assert args.concurrency == 3
|
|
assert args.dry_run is True
|
|
assert args.region == "us-east-1"
|
|
assert args.profile == "prod"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pipeline orchestration tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_PATCH_BASE = "cfn_updater.cli"
|
|
|
|
|
|
def _make_stacks(count: int, updatable: bool = True) -> list[DiscoveredStack]:
|
|
return [
|
|
DiscoveredStack(name=f"CL-AppPipe-{i}", status="CREATE_COMPLETE", updatable=updatable)
|
|
for i in range(count)
|
|
]
|
|
|
|
|
|
def _make_results(stacks: list[DiscoveredStack], status: str = "succeeded") -> list[StackUpdateResult]:
|
|
return [
|
|
StackUpdateResult(stack_name=s.name, status=status, duration_seconds=0.5)
|
|
for s in stacks
|
|
]
|
|
|
|
|
|
class TestMainPipeline:
|
|
"""Tests for the main() pipeline orchestration."""
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=[])
|
|
@patch(f"{_PATCH_BASE}.discover_stacks")
|
|
@patch(f"{_PATCH_BASE}.asyncio.run")
|
|
@patch(f"{_PATCH_BASE}.generate_report")
|
|
@patch(f"{_PATCH_BASE}.format_report", return_value="report output")
|
|
@patch(f"{_PATCH_BASE}.get_exit_code", return_value=0)
|
|
def test_successful_end_to_end(
|
|
self, mock_exit, mock_fmt, mock_gen, mock_arun, mock_disc, mock_perm, mock_session
|
|
):
|
|
stacks = _make_stacks(3)
|
|
results = _make_results(stacks)
|
|
mock_disc.return_value = stacks
|
|
mock_arun.return_value = results
|
|
mock_gen.return_value = MagicMock()
|
|
|
|
exit_code = main(["--prefix", "CL-AppPipe-"])
|
|
|
|
assert exit_code == 0
|
|
mock_perm.assert_called_once()
|
|
mock_disc.assert_called_once()
|
|
mock_arun.assert_called_once()
|
|
mock_gen.assert_called_once()
|
|
mock_exit.assert_called_once()
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=["cloudformation:UpdateStack"])
|
|
def test_permission_failure_returns_exit_code_2(self, mock_perm, mock_session, capsys):
|
|
exit_code = main([])
|
|
|
|
assert exit_code == 2
|
|
captured = capsys.readouterr()
|
|
assert "Missing required permissions" in captured.out
|
|
assert "cloudformation:UpdateStack" in captured.out
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=[])
|
|
@patch(f"{_PATCH_BASE}.discover_stacks", return_value=[])
|
|
def test_no_stacks_found_returns_exit_code_0(self, mock_disc, mock_perm, mock_session, capsys):
|
|
exit_code = main([])
|
|
|
|
assert exit_code == 0
|
|
captured = capsys.readouterr()
|
|
assert "No stacks found" in captured.out
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=[])
|
|
@patch(f"{_PATCH_BASE}.discover_stacks")
|
|
@patch(f"{_PATCH_BASE}.format_dry_run_output", return_value="Dry-run: 2 stack(s) discovered:")
|
|
def test_dry_run_returns_exit_code_0(self, mock_dry, mock_disc, mock_perm, mock_session, capsys):
|
|
stacks = _make_stacks(2)
|
|
mock_disc.return_value = stacks
|
|
|
|
exit_code = main(["--prefix", "CL-AppPipe-", "--dry-run"])
|
|
|
|
assert exit_code == 0
|
|
mock_dry.assert_called_once_with(stacks)
|
|
captured = capsys.readouterr()
|
|
assert "Dry-run" in captured.out
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=[])
|
|
@patch(f"{_PATCH_BASE}.discover_stacks")
|
|
@patch(f"{_PATCH_BASE}.asyncio.run")
|
|
@patch(f"{_PATCH_BASE}.generate_report")
|
|
@patch(f"{_PATCH_BASE}.format_report", return_value="report")
|
|
@patch(f"{_PATCH_BASE}.get_exit_code", return_value=1)
|
|
def test_failed_stacks_return_exit_code_1(
|
|
self, mock_exit, mock_fmt, mock_gen, mock_arun, mock_disc, mock_perm, mock_session
|
|
):
|
|
stacks = _make_stacks(2)
|
|
results = _make_results(stacks, status="failed")
|
|
mock_disc.return_value = stacks
|
|
mock_arun.return_value = results
|
|
mock_gen.return_value = MagicMock()
|
|
|
|
exit_code = main(["--prefix", "CL-AppPipe-"])
|
|
|
|
assert exit_code == 1
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=[])
|
|
@patch(f"{_PATCH_BASE}.discover_stacks")
|
|
@patch(f"{_PATCH_BASE}.format_dry_run_output", return_value="dry-run output")
|
|
def test_dry_run_does_not_call_update(self, mock_dry, mock_disc, mock_perm, mock_session):
|
|
"""Dry-run should never invoke update_all_stacks."""
|
|
mock_disc.return_value = _make_stacks(1)
|
|
|
|
with patch(f"{_PATCH_BASE}.asyncio.run") as mock_arun:
|
|
exit_code = main(["--dry-run"])
|
|
|
|
assert exit_code == 0
|
|
mock_arun.assert_not_called()
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
def test_profile_passed_to_session(self, mock_session_cls):
|
|
"""--profile should be forwarded to boto3.Session."""
|
|
mock_session = MagicMock()
|
|
mock_session.client.return_value = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
with patch(f"{_PATCH_BASE}.validate_permissions", return_value=["cloudformation:ListStacks"]):
|
|
main(["--profile", "audit"])
|
|
|
|
mock_session_cls.assert_called_once_with(profile_name="audit")
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
def test_region_passed_to_session(self, mock_session_cls):
|
|
"""--region should be forwarded to boto3.Session."""
|
|
mock_session = MagicMock()
|
|
mock_session.client.return_value = MagicMock()
|
|
mock_session_cls.return_value = mock_session
|
|
|
|
with patch(f"{_PATCH_BASE}.validate_permissions", return_value=["cloudformation:ListStacks"]):
|
|
main(["--region", "ap-southeast-1"])
|
|
|
|
mock_session_cls.assert_called_once_with(region_name="ap-southeast-1")
|
|
|
|
@patch(f"{_PATCH_BASE}.boto3.Session")
|
|
@patch(f"{_PATCH_BASE}.validate_permissions", return_value=[])
|
|
@patch(f"{_PATCH_BASE}.discover_stacks")
|
|
def test_discovered_count_printed(self, mock_disc, mock_perm, mock_session, capsys):
|
|
mock_disc.return_value = _make_stacks(5)
|
|
|
|
with patch(f"{_PATCH_BASE}.format_dry_run_output", return_value="dry"):
|
|
main(["--dry-run"])
|
|
|
|
captured = capsys.readouterr()
|
|
assert "Discovered 5 stack(s)" in captured.out
|