feat(P3): Nova rebrand — SSM path + tag keys (REQ-161/162)
SSM path /acdl/{env}/{contractId}/{output} → /nova/... across
core/output_publisher + contract resolver + consumer docs. New
scripts/migrate_ssm_paths.py (copy/verify/delete, dry-run default).
AWS tag keys acdl:owner|environment|contract|cost-center|ref → nova:*
across terraform tagging + ABAC session policies (iam:ResourceTag/acdl:*
→ iam:ResourceTag/nova:*). nova_tagging.py hard mode (D-109 warn→hard).
tagging-standard.json tag-key values → nova:*. New
scripts/untag_acdl_keys.py (remove old acdl:* tags, dry-run default).
Test fixtures updated; pytest + run_ci.sh PASS.
---ci---
project: acdl
phase: 3
milestone: v1.15
status: execute
---/ci---
This commit is contained in:
Jon Chery
+1
-1
@@ -20,7 +20,7 @@
|
||||
"policy": "require-resource-labels",
|
||||
"severity": "medium",
|
||||
"result": "fail",
|
||||
"message": "Pod missing required label acdl:owner.",
|
||||
"message": "Pod missing required label nova:owner.",
|
||||
"resource": "default/Pod/acdl-bad-app",
|
||||
"namespace": "default",
|
||||
"kind": "Pod",
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Unit tests for scripts/migrate_ssm_paths.py path-mapping logic (REQ-161, P3).
|
||||
|
||||
Tests the pure ``map_path()`` function (the AWS I/O glue is thin boto3 around
|
||||
it). The script does not need live AWS to be importable.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
||||
|
||||
from migrate_ssm_paths import map_path # noqa: E402
|
||||
|
||||
|
||||
class TestMapPath:
|
||||
def test_basic_dev_path(self):
|
||||
assert map_path("/acdl/dev/svc-x/output") == "/nova/dev/svc-x/output"
|
||||
|
||||
def test_basic_contract_path(self):
|
||||
assert map_path("/acdl/dev/c-1/vpc_id") == "/nova/dev/c-1/vpc_id"
|
||||
|
||||
def test_qa_env(self):
|
||||
assert map_path("/acdl/qa/c-2/db_endpoint") == "/nova/qa/c-2/db_endpoint"
|
||||
|
||||
def test_prod_env(self):
|
||||
assert map_path("/acdl/prod/c-3/distribution_domain_name") == "/nova/prod/c-3/distribution_domain_name"
|
||||
|
||||
def test_dr_env(self):
|
||||
assert map_path("/acdl/dr/c-4/bucket_arn") == "/nova/dr/c-4/bucket_arn"
|
||||
|
||||
def test_deep_nested_path(self):
|
||||
assert map_path("/acdl/dev/contract-001/nested/deep/output") == "/nova/dev/contract-001/nested/deep/output"
|
||||
|
||||
def test_preserves_trailing_segment(self):
|
||||
# The output name segment is preserved verbatim
|
||||
assert map_path("/acdl/dev/c/secret_token") == "/nova/dev/c/secret_token"
|
||||
|
||||
def test_custom_prefixes(self):
|
||||
assert map_path("/acdl/dev/c/x", "/acdl", "/nova") == "/nova/dev/c/x"
|
||||
assert map_path("/old/dev/c/x", "/old", "/new") == "/new/dev/c/x"
|
||||
|
||||
def test_raises_on_nonmatching_path(self):
|
||||
with pytest.raises(ValueError, match="does not start with source prefix"):
|
||||
map_path("/nova/dev/c/output")
|
||||
|
||||
def test_raises_on_path_not_segment_prefixed(self):
|
||||
# /acdl-platform is NOT a path-segment match for /acdl (no trailing /)
|
||||
with pytest.raises(ValueError, match="does not start with source prefix"):
|
||||
map_path("/acdl-platform-key")
|
||||
|
||||
def test_raises_on_empty_path(self):
|
||||
with pytest.raises(ValueError):
|
||||
map_path("")
|
||||
|
||||
def test_raises_on_just_prefix(self):
|
||||
# Exactly /acdl (no trailing slash) is not a valid parameter path
|
||||
with pytest.raises(ValueError):
|
||||
map_path("/acdl")
|
||||
|
||||
def test_round_trip_identity(self):
|
||||
# map_path is its own inverse when source/dest are swapped
|
||||
src = "/acdl/dev/svc-x/output"
|
||||
mapped = map_path(src, "/acdl", "/nova")
|
||||
back = map_path(mapped, "/nova", "/acdl")
|
||||
assert back == src
|
||||
|
||||
def test_idempotent_on_already_migrated(self):
|
||||
# If somehow a /nova/ path is passed with default args, it raises
|
||||
# (the script filters by source prefix before mapping)
|
||||
with pytest.raises(ValueError):
|
||||
map_path("/nova/dev/c/output")
|
||||
|
||||
def test_preserves_value_segment_exactly(self):
|
||||
# Hyphens, dots, underscores in output names are preserved
|
||||
assert map_path("/acdl/dev/c-1/my.output-name_2") == "/nova/dev/c-1/my.output-name_2"
|
||||
@@ -50,12 +50,12 @@ class TestPublishToSsm:
|
||||
outputs = {"bucket_name": "acdl-spike-bucket", "secret_token": "s3cret"}
|
||||
results = publish_to_ssm(outputs, "dev", "contract-001")
|
||||
|
||||
assert results["bucket_name"] == "/acdl/dev/contract-001/bucket_name"
|
||||
assert results["secret_token"] == "/acdl/dev/contract-001/secret_token"
|
||||
assert results["bucket_name"] == "/nova/dev/contract-001/bucket_name"
|
||||
assert results["secret_token"] == "/nova/dev/contract-001/secret_token"
|
||||
|
||||
# Verify the parameter landed in SSM correctly
|
||||
param = ssm.get_parameter(
|
||||
Name="/acdl/dev/contract-001/bucket_name", WithDecryption=True
|
||||
Name="/nova/dev/contract-001/bucket_name", WithDecryption=True
|
||||
)
|
||||
assert param["Parameter"]["Type"] == "SecureString"
|
||||
assert param["Parameter"]["Value"] == "acdl-spike-bucket"
|
||||
@@ -72,7 +72,7 @@ class TestPublishToSsm:
|
||||
with mock_aws():
|
||||
ssm = boto3.client("ssm", region_name="us-east-1")
|
||||
publish_to_ssm({"vpc_id": "vpc-123"}, "dev", "c-1")
|
||||
param = ssm.get_parameter(Name="/acdl/dev/c-1/vpc_id", WithDecryption=True)
|
||||
param = ssm.get_parameter(Name="/nova/dev/c-1/vpc_id", WithDecryption=True)
|
||||
assert param["Parameter"]["Type"] == "SecureString"
|
||||
|
||||
def test_publish_skips_none_and_empty_values(self, monkeypatch):
|
||||
@@ -112,7 +112,7 @@ class TestPublishToSsm:
|
||||
publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
|
||||
# Second publish with a new value should overwrite, not error
|
||||
publish_to_ssm({"vpc_id": "vpc-2"}, "dev", "c-1")
|
||||
param = ssm.get_parameter(Name="/acdl/dev/c-1/vpc_id", WithDecryption=True)
|
||||
param = ssm.get_parameter(Name="/nova/dev/c-1/vpc_id", WithDecryption=True)
|
||||
assert param["Parameter"]["Value"] == "vpc-2"
|
||||
|
||||
def test_publish_continues_on_single_failure(self, monkeypatch):
|
||||
@@ -142,7 +142,7 @@ class TestPublishToSsm:
|
||||
results = publish_to_ssm(
|
||||
{"good": "val", "bad": "val"}, "dev", "c-1"
|
||||
)
|
||||
assert results["good"] == "/acdl/dev/c-1/good"
|
||||
assert results["good"] == "/nova/dev/c-1/good"
|
||||
assert results["bad"] is None
|
||||
|
||||
|
||||
@@ -156,7 +156,7 @@ class TestFormatComment:
|
||||
comment = format_comment(outputs, "dev", "contract-001")
|
||||
assert "acdl-spike-bucket" in comment
|
||||
assert "vpc-abc123" in comment
|
||||
assert "### ACDL Deploy Outputs (dev)" in comment
|
||||
assert "### Nova Deploy Outputs (dev)" in comment
|
||||
assert "`contract-001`" in comment
|
||||
|
||||
def test_sensitive_outputs_show_published_to_ssm(self):
|
||||
@@ -175,12 +175,12 @@ class TestFormatComment:
|
||||
def test_ssm_path_included_when_results_provided(self):
|
||||
outputs = {"bucket_name": "my-bucket", "secret_token": "s3cret"}
|
||||
ssm_results = {
|
||||
"bucket_name": "/acdl/dev/contract-001/bucket_name",
|
||||
"secret_token": "/acdl/dev/contract-001/secret_token",
|
||||
"bucket_name": "/nova/dev/contract-001/bucket_name",
|
||||
"secret_token": "/nova/dev/contract-001/secret_token",
|
||||
}
|
||||
comment = format_comment(outputs, "dev", "contract-001", ssm_results)
|
||||
assert "/acdl/dev/contract-001/bucket_name" in comment
|
||||
assert "/acdl/dev/contract-001/secret_token" in comment
|
||||
assert "/nova/dev/contract-001/bucket_name" in comment
|
||||
assert "/nova/dev/contract-001/secret_token" in comment
|
||||
|
||||
def test_dash_shown_when_ssm_results_provided_but_missing(self):
|
||||
outputs = {"bucket_name": "my-bucket"}
|
||||
@@ -193,12 +193,12 @@ class TestFormatComment:
|
||||
outputs = {"bucket_name": "my-bucket"}
|
||||
comment = format_comment(outputs, "dev", "contract-001", ssm_results=None)
|
||||
# No SSM column content when ssm_results is None
|
||||
assert "/acdl/" not in comment or "get-parameter" in comment # only footer
|
||||
assert "/nova/" not in comment or "get-parameter" in comment # only footer
|
||||
|
||||
def test_ssm_footer_contains_correct_path(self):
|
||||
outputs = {"bucket_name": "b"}
|
||||
comment = format_comment(outputs, "dev", "contract-001")
|
||||
assert "/acdl/dev/contract-001/<output_name>" in comment
|
||||
assert "/nova/dev/contract-001/<output_name>" in comment
|
||||
|
||||
def test_skips_none_and_empty_values(self):
|
||||
outputs = {"real": "val", "none_val": None, "empty": ""}
|
||||
@@ -355,7 +355,7 @@ class TestCli:
|
||||
output = captured.getvalue()
|
||||
assert "cli-bucket" in output
|
||||
assert "vpc-1" in output
|
||||
assert "### ACDL Deploy Outputs (dev)" in output
|
||||
assert "### Nova Deploy Outputs (dev)" in output
|
||||
finally:
|
||||
op.boto3 = saved_boto3
|
||||
sys.argv = old_argv
|
||||
@@ -411,8 +411,8 @@ class TestKmsFailLoud:
|
||||
with mock_aws():
|
||||
ssm = boto3.client("ssm", region_name="us-east-1")
|
||||
results = publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
|
||||
assert results["vpc_id"] == "/acdl/dev/c-1/vpc_id"
|
||||
param = ssm.get_parameter(Name="/acdl/dev/c-1/vpc_id", WithDecryption=True)
|
||||
assert results["vpc_id"] == "/nova/dev/c-1/vpc_id"
|
||||
param = ssm.get_parameter(Name="/nova/dev/c-1/vpc_id", WithDecryption=True)
|
||||
assert param["Parameter"]["Type"] == "SecureString"
|
||||
|
||||
def test_kms_set_takes_precedence_over_allow_default(self, monkeypatch):
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""Unit tests for scripts/untag_acdl_keys.py key-list logic (REQ-162, P3).
|
||||
|
||||
Tests the pure ``acdl_keys_in()`` + ``keys_to_untag()`` functions (the AWS
|
||||
I/O glue is thin boto3 around them). The script does not need live AWS to
|
||||
be importable.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
||||
|
||||
from untag_acdl_keys import acdl_keys_in, keys_to_untag, DEFAULT_LEGACY_KEYS # noqa: E402
|
||||
|
||||
|
||||
class TestAcdlKeysIn:
|
||||
def test_empty(self):
|
||||
assert acdl_keys_in([]) == []
|
||||
|
||||
def test_no_acdl_keys(self):
|
||||
assert acdl_keys_in(["nova:owner", "nova:contract", "Name"]) == []
|
||||
|
||||
def test_all_acdl_keys(self):
|
||||
keys = ["acdl:owner", "acdl:contract", "acdl:environment", "acdl:cost-center", "acdl:ref"]
|
||||
assert acdl_keys_in(keys) == list(keys)
|
||||
|
||||
def test_mixed_keys(self):
|
||||
keys = ["acdl:owner", "nova:owner", "Name", "acdl:cost-center"]
|
||||
assert acdl_keys_in(keys) == ["acdl:owner", "acdl:cost-center"]
|
||||
|
||||
def test_preserves_input_order(self):
|
||||
keys = ["acdl:ref", "nova:owner", "acdl:owner", "acdl:contract"]
|
||||
assert acdl_keys_in(keys) == ["acdl:ref", "acdl:owner", "acdl:contract"]
|
||||
|
||||
def test_custom_legacy_set(self):
|
||||
# Only removing acdl:owner + acdl:ref (subset)
|
||||
legacy = ("acdl:owner", "acdl:ref")
|
||||
keys = ["acdl:owner", "acdl:contract", "acdl:ref", "nova:owner"]
|
||||
assert acdl_keys_in(keys, legacy_keys=legacy) == ["acdl:owner", "acdl:ref"]
|
||||
|
||||
def test_default_legacy_keys_all_5(self):
|
||||
assert len(DEFAULT_LEGACY_KEYS) == 5
|
||||
assert "acdl:owner" in DEFAULT_LEGACY_KEYS
|
||||
assert "acdl:contract" in DEFAULT_LEGACY_KEYS
|
||||
assert "acdl:environment" in DEFAULT_LEGACY_KEYS
|
||||
assert "acdl:cost-center" in DEFAULT_LEGACY_KEYS
|
||||
assert "acdl:ref" in DEFAULT_LEGACY_KEYS
|
||||
|
||||
def test_duplicates_not_duplicated_in_output(self):
|
||||
# List comprehension preserves duplicates in input; the API dedups via set
|
||||
# but the function is a faithful list filter. Duplicates are unusual but
|
||||
# the function does not dedup (the UntagResources API tolerates the same
|
||||
# key once; real GetResources never returns duplicate keys).
|
||||
keys = ["acdl:owner", "acdl:owner"]
|
||||
assert acdl_keys_in(keys) == ["acdl:owner", "acdl:owner"]
|
||||
|
||||
|
||||
class TestKeysToUntag:
|
||||
def test_empty_tags(self):
|
||||
assert keys_to_untag({"ResourceARN": "arn:...", "Tags": []}) == []
|
||||
|
||||
def test_no_tags_key(self):
|
||||
assert keys_to_untag({"ResourceARN": "arn:..."}) == []
|
||||
|
||||
def test_with_acdl_keys(self):
|
||||
resource = {
|
||||
"ResourceARN": "arn:aws:s3:::my-bucket",
|
||||
"Tags": [
|
||||
{"Key": "acdl:owner", "Value": "acdl"},
|
||||
{"Key": "nova:owner", "Value": "acdl"},
|
||||
{"Key": "acdl:cost-center", "Value": "acdl-default"},
|
||||
{"Key": "Name", "Value": "my-bucket"},
|
||||
],
|
||||
}
|
||||
assert keys_to_untag(resource) == ["acdl:owner", "acdl:cost-center"]
|
||||
|
||||
def test_with_only_nova_keys(self):
|
||||
resource = {
|
||||
"ResourceARN": "arn:aws:s3:::my-bucket",
|
||||
"Tags": [
|
||||
{"Key": "nova:owner", "Value": "acdl"},
|
||||
{"Key": "nova:contract", "Value": "platform"},
|
||||
{"Key": "Name", "Value": "my-bucket"},
|
||||
],
|
||||
}
|
||||
assert keys_to_untag(resource) == []
|
||||
|
||||
def test_all_5_acdl_keys(self):
|
||||
resource = {
|
||||
"ResourceARN": "arn:aws:ecs:us-east-1:123:cluster/x",
|
||||
"Tags": [
|
||||
{"Key": "acdl:owner", "Value": "v"},
|
||||
{"Key": "acdl:contract", "Value": "v"},
|
||||
{"Key": "acdl:environment", "Value": "v"},
|
||||
{"Key": "acdl:cost-center", "Value": "v"},
|
||||
{"Key": "acdl:ref", "Value": "v"},
|
||||
],
|
||||
}
|
||||
result = keys_to_untag(resource)
|
||||
assert result == ["acdl:owner", "acdl:contract", "acdl:environment", "acdl:cost-center", "acdl:ref"]
|
||||
|
||||
def test_custom_legacy_keys(self):
|
||||
resource = {
|
||||
"ResourceARN": "arn:...",
|
||||
"Tags": [
|
||||
{"Key": "acdl:owner", "Value": "v"},
|
||||
{"Key": "acdl:contract", "Value": "v"},
|
||||
],
|
||||
}
|
||||
# Only targeting acdl:owner
|
||||
assert keys_to_untag(resource, legacy_keys=("acdl:owner",)) == ["acdl:owner"]
|
||||
|
||||
def test_malformed_tag_entry_skipped(self):
|
||||
# A tag entry without a Key is skipped gracefully
|
||||
resource = {
|
||||
"ResourceARN": "arn:...",
|
||||
"Tags": [
|
||||
{"Value": "no-key"},
|
||||
{"Key": "acdl:owner", "Value": "v"},
|
||||
"not-a-dict",
|
||||
],
|
||||
}
|
||||
assert keys_to_untag(resource) == ["acdl:owner"]
|
||||
|
||||
|
||||
class TestRunDryRunNoClient:
|
||||
def test_dry_run_returns_empty_summary_without_client(self):
|
||||
from untag_acdl_keys import run
|
||||
summary = run(apply=False, client=None)
|
||||
assert summary["listed"] == 0
|
||||
assert summary["untagged"] == 0
|
||||
assert summary["keys_removed"] == 0
|
||||
assert summary["errors"] == 0
|
||||
|
||||
def test_dry_run_apply_false_no_mutation(self):
|
||||
# apply=False with a client still only lists (no untag)
|
||||
from untag_acdl_keys import run
|
||||
|
||||
class FakeClient:
|
||||
def get_paginator(self, name):
|
||||
class P:
|
||||
def paginate(self, **kw):
|
||||
return iter([{"ResourceMappingList": []}])
|
||||
return P()
|
||||
|
||||
summary = run(apply=False, client=FakeClient())
|
||||
assert summary["listed"] == 0
|
||||
assert summary["untagged"] == 0
|
||||
Reference in New Issue
Block a user