Jon Chery
126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
"""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"
|
|
|
|
class TestNarrowedException:
|
|
"""P4 (REQ-168): the copy_one_param except is narrowed to
|
|
ParameterNotFound; non-ParameterNotFound errors surface (not swallowed)."""
|
|
|
|
def test_parameter_not_found_proceeds_to_put(self):
|
|
"""A ParameterNotFound on the dest get_parameter (target absent) is
|
|
the expected 'proceed to put' path — not an error."""
|
|
from unittest import mock
|
|
import migrate_ssm_paths as m
|
|
|
|
class FakeExceptions:
|
|
ParameterNotFound = type("ParameterNotFound", (Exception,), {})
|
|
|
|
fake_client = mock.Mock()
|
|
fake_client.exceptions = FakeExceptions
|
|
# source get_parameter succeeds; dest get_parameter raises ParameterNotFound
|
|
fake_client.get_parameter.side_effect = [
|
|
{"Parameter": {"Value": "v", "Type": "String", "KeyId": None}},
|
|
FakeExceptions.ParameterNotFound(),
|
|
]
|
|
fake_client.put_parameter.return_value = {"Version": 1}
|
|
result = m.copy_one_param(fake_client, "/acdl/dev/c/out", "/nova/dev/c/out")
|
|
assert result == "copied"
|
|
fake_client.put_parameter.assert_called_once()
|
|
|
|
def test_non_parameter_not_found_error_is_raised(self):
|
|
"""A non-ParameterNotFound AWS error (e.g. ThrottlingException) on
|
|
the dest get_parameter is raised, not swallowed (P4, REQ-168)."""
|
|
from unittest import mock
|
|
import migrate_ssm_paths as m
|
|
|
|
class FakeExceptions:
|
|
ParameterNotFound = type("ParameterNotFound", (Exception,), {})
|
|
|
|
class ThrottlingException(Exception):
|
|
pass
|
|
|
|
fake_client = mock.Mock()
|
|
fake_client.exceptions = FakeExceptions
|
|
# source get_parameter succeeds; dest get_parameter raises Throttling
|
|
fake_client.get_parameter.side_effect = [
|
|
{"Parameter": {"Value": "v", "Type": "String", "KeyId": None}},
|
|
ThrottlingException("slow down"),
|
|
]
|
|
with pytest.raises(ThrottlingException):
|
|
m.copy_one_param(fake_client, "/acdl/dev/c/out", "/nova/dev/c/out")
|
|
fake_client.put_parameter.assert_not_called()
|