Files
acdl/tests/test_decommission.py
T
Jon Chery 134f85d2df feat(P34): decommission alias + CMDB validation (REQ-92, REQ-93, REQ-94)
---ci---
project: acdl
phase: 34
milestone: v1.8
status: execute
---/ci---

- DynamoDB acdl-change-requests table added to terraform/platform/main.tf
  (PK changeRequestId, SK submittedAt, SSE via CMK, PITR).
- validate_change_request Lambda action added to contract_ingestor.py:
  queries CMDB, asserts status=approved + consumerRepo match.
- decommission_transform() added to contract_resolver.py: zeroes all
  counts (desired_count, min/max_capacity) + sets deletion_protection=false.
- Decommission mode added to deploy pipeline + both deploy workflows
  (mode: decommission + changeRequestId input). Byte-identical.
- run_platform.sh --decommission flag: validates CR, resolves with
  deletion_protection=false (step 1), then decommission_transform
  (step 2). HITL SRE gates documented.
- docs/consumer-guide.md: new "Decommissioning a stack" section with
  CR request, trigger, 2-step HITL SRE gates, CMK deletion window, uptime.

Tests: +14 (318 -> 332). All pass.
2026-07-22 22:18:28 +00:00

107 lines
4.5 KiB
Python

"""Tests for the decommission transform (REQ-92) + decommission pipeline mode."""
import json
import sys
from pathlib import Path
import pytest
import yaml
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
ROOT = Path(__file__).resolve().parent.parent
class TestDecommissionTransform:
"""REQ-92: decommission_transform zeroes counts + disables deletion protection."""
def test_decommission_transform_zeros_desired_count(self):
from core.contract_resolver import decommission_transform
stack = {
"version": "1.0.0",
"stack": {"name": "test", "kind": "l1", "depth": 1},
"resources": [
{"id": "svc", "type": "aws:ecs:service", "module": "ecs-service@1.0.0",
"inputs": {"desired_count": 3}, "outputs": {}, "nfrs": {"deletion_protection": True}},
],
}
result = decommission_transform(stack)
assert result["resources"][0]["inputs"]["desired_count"] == 0
assert result["resources"][0]["nfrs"]["deletion_protection"] is False
def test_decommission_transform_zeros_min_max_capacity(self):
from core.contract_resolver import decommission_transform
stack = {
"version": "1.0.0",
"stack": {"name": "test", "kind": "l1", "depth": 1},
"resources": [
{"id": "asg", "type": "aws:autoscaling:group", "module": "asg@1.0.0",
"inputs": {"min_capacity": 2, "max_capacity": 10}, "outputs": {}, "nfrs": {}},
],
}
result = decommission_transform(stack)
assert result["resources"][0]["inputs"]["min_capacity"] == 0
assert result["resources"][0]["inputs"]["max_capacity"] == 0
assert result["resources"][0]["nfrs"]["deletion_protection"] is False
def test_decommission_transform_sets_deletion_protection_false_on_all(self):
from core.contract_resolver import decommission_transform
stack = {
"version": "1.0.0",
"stack": {"name": "test", "kind": "l2", "depth": 1},
"resources": [
{"id": "s3", "type": "aws:s3:bucket", "module": "s3@1.0.0",
"inputs": {}, "outputs": {}, "nfrs": {"deletion_protection": True}},
{"id": "rds", "type": "aws:rds:instance", "module": "rds@1.0.0",
"inputs": {}, "outputs": {}, "nfrs": {"deletion_protection": True}},
],
}
result = decommission_transform(stack)
for res in result["resources"]:
assert res["nfrs"]["deletion_protection"] is False
def test_decommission_transform_handles_empty_nfrs(self):
from core.contract_resolver import decommission_transform
stack = {
"version": "1.0.0",
"stack": {"name": "test", "kind": "l1", "depth": 1},
"resources": [
{"id": "s3", "type": "aws:s3:bucket", "module": "s3@1.0.0",
"inputs": {}, "outputs": {}},
],
}
result = decommission_transform(stack)
assert result["resources"][0]["nfrs"]["deletion_protection"] is False
class TestDecommissionPipelineContract:
"""REQ-92: decommission mode in the deploy pipeline contract + workflows."""
def test_deploy_workflow_has_decommission_mode(self):
wf = yaml.safe_load(open(ROOT / ".github/workflows/deploy.yml"))
on_key = "on" if "on" in wf else True
inputs = wf[on_key]["workflow_call"]["inputs"]
assert "mode" in inputs
assert "decommission" in inputs["mode"]["description"]
def test_deploy_workflow_has_change_request_id_input(self):
wf = yaml.safe_load(open(ROOT / ".github/workflows/deploy.yml"))
on_key = "on" if "on" in wf else True
inputs = wf[on_key]["workflow_call"]["inputs"]
assert "changeRequestId" in inputs
def test_deploy_workflow_decommission_requires_change_request_id(self):
wf_text = open(ROOT / ".github/workflows/deploy.yml").read()
assert "changeRequestId" in wf_text
assert "decommission" in wf_text
def test_deploy_workflows_byte_identical(self):
gitea = open(ROOT / ".gitea/workflows/deploy.yml", "rb").read()
github = open(ROOT / ".github/workflows/deploy.yml", "rb").read()
assert gitea == github
def test_consumer_guide_has_decommission_section(self):
guide = open(ROOT / "docs/consumer-guide.md").read()
assert "Decommissioning a stack" in guide
assert "decommission-gate-sre" in guide
assert "decommission-destroy-sre" in guide