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)
325 lines
11 KiB
Python
325 lines
11 KiB
Python
# Feature: one-click-cfn-stack-updater, Property 1: Discovery returns exactly prefix-matched stacks with correct count
|
|
"""Property-based tests for stack discovery prefix filtering.
|
|
|
|
**Validates: Requirements 1.1, 1.3**
|
|
|
|
Property 1: For any list of CloudFormation stacks with arbitrary names,
|
|
the discovery function returns exactly those stacks whose names start with
|
|
the configured prefix, and the reported count equals the length of that
|
|
filtered list.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
from hypothesis import given, settings, strategies as st
|
|
|
|
from cfn_updater.discovery import discover_stacks
|
|
|
|
PREFIX = "CL-AppPipe-"
|
|
|
|
# Strategy: generate a suffix that doesn't accidentally start with the prefix
|
|
_safe_suffix = st.text(
|
|
alphabet=st.characters(whitelist_categories=("L", "N", "Pd"), whitelist_characters="-_"),
|
|
min_size=1,
|
|
max_size=20,
|
|
)
|
|
|
|
# Stack names that DO match the prefix
|
|
_prefixed_name = _safe_suffix.map(lambda s: PREFIX + s)
|
|
|
|
# Stack names that do NOT match the prefix — use a different leading string
|
|
_non_prefixed_name = _safe_suffix.map(lambda s: "OTHER-" + s).filter(
|
|
lambda n: not n.startswith(PREFIX)
|
|
)
|
|
|
|
# A mixed list of stack names (some prefixed, some not)
|
|
_stack_names = st.lists(
|
|
st.one_of(_prefixed_name, _non_prefixed_name),
|
|
min_size=0,
|
|
max_size=30,
|
|
)
|
|
|
|
|
|
def _mock_cfn_client(stack_names: list[str]) -> MagicMock:
|
|
"""Build a mock cfn_client whose list_stacks returns the given names."""
|
|
client = MagicMock()
|
|
summaries = [
|
|
{"StackName": name, "StackStatus": "CREATE_COMPLETE"}
|
|
for name in stack_names
|
|
]
|
|
client.list_stacks.return_value = {
|
|
"StackSummaries": summaries,
|
|
# No NextToken → single page
|
|
}
|
|
return client
|
|
|
|
|
|
@settings(max_examples=100)
|
|
@given(names=_stack_names)
|
|
def test_discovery_returns_exactly_prefix_matched_stacks(
|
|
names: list[str],
|
|
) -> None:
|
|
"""Property 1: Discovery returns exactly prefix-matched stacks with correct count.
|
|
|
|
**Validates: Requirements 1.1, 1.3**
|
|
"""
|
|
cfn = _mock_cfn_client(names)
|
|
|
|
result = discover_stacks(cfn, PREFIX)
|
|
|
|
expected_names = [n for n in names if n.startswith(PREFIX)]
|
|
|
|
# The returned stacks are exactly those whose names start with the prefix
|
|
assert [s.name for s in result] == expected_names
|
|
|
|
# The count equals the number of prefix-matched stacks
|
|
assert len(result) == len(expected_names)
|
|
|
|
|
|
# Feature: one-click-cfn-stack-updater, Property 6: Non-updatable stacks are skipped
|
|
"""Property-based test for non-updatable stack classification.
|
|
|
|
**Validates: Requirements 4.2**
|
|
|
|
Property 6: For any stack with a status in NON_UPDATABLE_STATUSES, updatable
|
|
is False. For any stack with a status not in NON_UPDATABLE_STATUSES, updatable
|
|
is True.
|
|
"""
|
|
|
|
from cfn_updater.config import NON_UPDATABLE_STATUSES
|
|
|
|
# Statuses that appear in _ACTIVE_STACK_STATUSES (excludes DELETE_COMPLETE)
|
|
_UPDATABLE_STATUSES = [
|
|
"CREATE_IN_PROGRESS",
|
|
"CREATE_FAILED",
|
|
"CREATE_COMPLETE",
|
|
"ROLLBACK_FAILED",
|
|
"DELETE_FAILED",
|
|
"UPDATE_IN_PROGRESS",
|
|
"UPDATE_COMPLETE_CLEANUP_IN_PROGRESS",
|
|
"UPDATE_COMPLETE",
|
|
"UPDATE_FAILED",
|
|
"UPDATE_ROLLBACK_FAILED",
|
|
"UPDATE_ROLLBACK_COMPLETE",
|
|
"REVIEW_IN_PROGRESS",
|
|
"IMPORT_IN_PROGRESS",
|
|
"IMPORT_COMPLETE",
|
|
"IMPORT_ROLLBACK_IN_PROGRESS",
|
|
"IMPORT_ROLLBACK_FAILED",
|
|
"IMPORT_ROLLBACK_COMPLETE",
|
|
]
|
|
|
|
# Non-updatable statuses that are in _ACTIVE_STACK_STATUSES (excludes DELETE_COMPLETE)
|
|
_NON_UPDATABLE_ACTIVE_STATUSES = [
|
|
"ROLLBACK_COMPLETE",
|
|
"ROLLBACK_IN_PROGRESS",
|
|
"UPDATE_ROLLBACK_IN_PROGRESS",
|
|
"UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS",
|
|
"DELETE_IN_PROGRESS",
|
|
]
|
|
|
|
_ALL_ACTIVE_STATUSES = _UPDATABLE_STATUSES + _NON_UPDATABLE_ACTIVE_STATUSES
|
|
|
|
# Strategy: generate a list of (suffix, status) pairs for stacks with the prefix
|
|
_stack_entry = st.tuples(
|
|
_safe_suffix,
|
|
st.sampled_from(_ALL_ACTIVE_STATUSES),
|
|
)
|
|
|
|
_stack_entries = st.lists(_stack_entry, min_size=1, max_size=30)
|
|
|
|
|
|
def _mock_cfn_client_with_statuses(entries: list[tuple[str, str]]) -> MagicMock:
|
|
"""Build a mock cfn_client returning stacks with given (suffix, status) pairs."""
|
|
client = MagicMock()
|
|
summaries = [
|
|
{"StackName": PREFIX + suffix, "StackStatus": status}
|
|
for suffix, status in entries
|
|
]
|
|
client.list_stacks.return_value = {"StackSummaries": summaries}
|
|
return client
|
|
|
|
|
|
@settings(max_examples=100)
|
|
@given(entries=_stack_entries)
|
|
def test_non_updatable_stacks_are_classified_correctly(
|
|
entries: list[tuple[str, str]],
|
|
) -> None:
|
|
"""Property 6: Non-updatable stacks are skipped.
|
|
|
|
**Validates: Requirements 4.2**
|
|
"""
|
|
cfn = _mock_cfn_client_with_statuses(entries)
|
|
|
|
result = discover_stacks(cfn, PREFIX)
|
|
|
|
assert len(result) == len(entries)
|
|
|
|
for stack, (_, status) in zip(result, entries):
|
|
if status in NON_UPDATABLE_STATUSES:
|
|
assert stack.updatable is False, (
|
|
f"Stack with status {status!r} should NOT be updatable"
|
|
)
|
|
else:
|
|
assert stack.updatable is True, (
|
|
f"Stack with status {status!r} should be updatable"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unit tests for stack discovery
|
|
# Requirements: 1.1, 1.2, 1.3, 1.4, 4.2
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDiscoverStacksEmpty:
|
|
"""Req 1.4: If no Target_Stacks are found, return an empty list."""
|
|
|
|
def test_empty_stack_list_returns_empty(self) -> None:
|
|
"""No stacks at all → empty result."""
|
|
client = MagicMock()
|
|
client.list_stacks.return_value = {"StackSummaries": []}
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert result == []
|
|
|
|
def test_no_matching_prefix_returns_empty(self) -> None:
|
|
"""Stacks exist but none match the prefix → empty result."""
|
|
client = MagicMock()
|
|
client.list_stacks.return_value = {
|
|
"StackSummaries": [
|
|
{"StackName": "OTHER-stack-1", "StackStatus": "CREATE_COMPLETE"},
|
|
{"StackName": "my-app-stack", "StackStatus": "UPDATE_COMPLETE"},
|
|
]
|
|
}
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert result == []
|
|
|
|
|
|
class TestDiscoverStacksPagination:
|
|
"""Pagination: discover_stacks must follow NextToken across pages."""
|
|
|
|
def test_two_pages_are_merged(self) -> None:
|
|
"""Two pages of results are combined into a single list."""
|
|
client = MagicMock()
|
|
client.list_stacks.side_effect = [
|
|
{
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-aaa", "StackStatus": "CREATE_COMPLETE"},
|
|
],
|
|
"NextToken": "page2",
|
|
},
|
|
{
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-bbb", "StackStatus": "UPDATE_COMPLETE"},
|
|
],
|
|
# No NextToken → last page
|
|
},
|
|
]
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert len(result) == 2
|
|
assert result[0].name == "CL-AppPipe-aaa"
|
|
assert result[1].name == "CL-AppPipe-bbb"
|
|
|
|
# Verify the second call included the NextToken
|
|
calls = client.list_stacks.call_args_list
|
|
assert len(calls) == 2
|
|
assert "NextToken" not in calls[0].kwargs
|
|
assert calls[1].kwargs["NextToken"] == "page2"
|
|
|
|
def test_three_pages_with_mixed_stacks(self) -> None:
|
|
"""Three pages, some stacks match prefix and some don't."""
|
|
client = MagicMock()
|
|
client.list_stacks.side_effect = [
|
|
{
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-p1", "StackStatus": "CREATE_COMPLETE"},
|
|
{"StackName": "OTHER-stack", "StackStatus": "CREATE_COMPLETE"},
|
|
],
|
|
"NextToken": "tok2",
|
|
},
|
|
{
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-p2", "StackStatus": "UPDATE_COMPLETE"},
|
|
],
|
|
"NextToken": "tok3",
|
|
},
|
|
{
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-p3", "StackStatus": "CREATE_COMPLETE"},
|
|
],
|
|
},
|
|
]
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert [s.name for s in result] == [
|
|
"CL-AppPipe-p1",
|
|
"CL-AppPipe-p2",
|
|
"CL-AppPipe-p3",
|
|
]
|
|
assert client.list_stacks.call_count == 3
|
|
|
|
|
|
class TestDiscoverStacksMixedUpdatability:
|
|
"""Req 4.2: Non-updatable stacks are marked updatable=False."""
|
|
|
|
def test_mixed_updatable_and_non_updatable(self) -> None:
|
|
"""Stacks with various statuses are classified correctly."""
|
|
client = MagicMock()
|
|
client.list_stacks.return_value = {
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-ok1", "StackStatus": "CREATE_COMPLETE"},
|
|
{"StackName": "CL-AppPipe-rb", "StackStatus": "ROLLBACK_COMPLETE"},
|
|
{"StackName": "CL-AppPipe-ok2", "StackStatus": "UPDATE_COMPLETE"},
|
|
{"StackName": "CL-AppPipe-del", "StackStatus": "DELETE_IN_PROGRESS"},
|
|
{"StackName": "CL-AppPipe-ok3", "StackStatus": "IMPORT_COMPLETE"},
|
|
]
|
|
}
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert len(result) == 5
|
|
# updatable stacks
|
|
assert result[0].updatable is True # CREATE_COMPLETE
|
|
assert result[2].updatable is True # UPDATE_COMPLETE
|
|
assert result[4].updatable is True # IMPORT_COMPLETE
|
|
# non-updatable stacks
|
|
assert result[1].updatable is False # ROLLBACK_COMPLETE
|
|
assert result[3].updatable is False # DELETE_IN_PROGRESS
|
|
|
|
def test_all_non_updatable(self) -> None:
|
|
"""Every discovered stack is non-updatable."""
|
|
client = MagicMock()
|
|
client.list_stacks.return_value = {
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-a", "StackStatus": "ROLLBACK_COMPLETE"},
|
|
{"StackName": "CL-AppPipe-b", "StackStatus": "DELETE_IN_PROGRESS"},
|
|
{"StackName": "CL-AppPipe-c", "StackStatus": "ROLLBACK_IN_PROGRESS"},
|
|
]
|
|
}
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert all(s.updatable is False for s in result)
|
|
|
|
def test_all_updatable(self) -> None:
|
|
"""Every discovered stack is updatable."""
|
|
client = MagicMock()
|
|
client.list_stacks.return_value = {
|
|
"StackSummaries": [
|
|
{"StackName": "CL-AppPipe-x", "StackStatus": "CREATE_COMPLETE"},
|
|
{"StackName": "CL-AppPipe-y", "StackStatus": "UPDATE_COMPLETE"},
|
|
]
|
|
}
|
|
|
|
result = discover_stacks(client, PREFIX)
|
|
|
|
assert all(s.updatable is True for s in result)
|