How to use _resource_name_transformer method in localstack

Best Python code snippet using localstack_python

transformer_utility.py

Source:transformer_utility.py

...207        match = re.match(PATTERN_LOGSTREAM_ID, val)208        if match:209            return val210    return None211def _resource_name_transformer(key: str, val: str) -> str:212    if isinstance(val, str):213        match = re.match(PATTERN_ARN, val)214        if match:215            res = match.groups()[-1]216            if res.startswith("<") and res.endswith(">"):217                # value was already replaced218                return None219            if ":changeSet/" in val:220                return val.split(":changeSet/")[-1]221            if "/" in res:222                return res.split("/")[-1]223            if res.startswith("function:"):224                res = res.replace("function:", "")225                if "$" in res:...

test_snapshots.py

Source:test_snapshots.py

1import pytest2from localstack.testing.snapshots import SnapshotSession3from localstack.testing.snapshots.transformer import KeyValueBasedTransformer, SortingTransformer4from localstack.testing.snapshots.transformer_utility import _resource_name_transformer5class TestSnapshotManager:6    def test_simple_diff_nochange(self):7        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)8        sm.recorded_state = {"key_a": {"a": 3}}9        sm.match("key_a", {"a": 3})10        sm._assert_all()11    def test_simple_diff_change(self):12        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)13        sm.recorded_state = {"key_a": {"a": 3}}14        sm.match("key_a", {"a": 5})15        with pytest.raises(Exception) as ctx:16            sm._assert_all()17        ctx.match("Parity snapshot failed")18    def test_multiple_assertmatch_with_same_key_fail(self):19        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)20        sm.recorded_state = {"key_a": {"a": 3}}21        sm.match("key_a", {"a": 3})22        with pytest.raises(Exception) as ctx:23            sm.match("key_a", {"a": 3})24        ctx.match("used multiple times in the same test scope")25    def test_context_replacement(self):26        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)27        sm.add_transformer(28            KeyValueBasedTransformer(lambda k, v: v if k == "aaa" else None, replacement="A")29        )30        sm.recorded_state = {"key_a": {"aaa": "<A:1>", "bbb": "<A:1> hello"}}31        sm.match("key_a", {"aaa": "something", "bbb": "something hello"})32        sm._assert_all()33    def test_match_order_reference_replacement(self):34        """tests if the reference-replacement works as expected, e.g., using alphabetical order of keys"""35        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)36        sm.add_transformer(KeyValueBasedTransformer(_resource_name_transformer, "resource"))37        sm.recorded_state = {38            "subscription-attributes": {39                "Attributes": {40                    "ConfirmationWasAuthenticated": "true",41                    "Endpoint": "arn:aws:lambda:region:111111111111:function:<resource:1>",42                    "Owner": "111111111111",43                    "PendingConfirmation": "false",44                    "Protocol": "lambda",45                    "RawMessageDelivery": "false",46                    "RedrivePolicy": {47                        "deadLetterTargetArn": "arn:aws:sqs:region:111111111111:<resource:2>"48                    },49                    "SubscriptionArn": "arn:aws:sns:region:111111111111:<resource:4>:<resource:3>",50                    "TopicArn": "arn:aws:sns:region:111111111111:<resource:4>",51                },52                "ResponseMetadata": {"HTTPHeaders": {}, "HTTPStatusCode": 200},53            }54        }55        sm.match(56            "subscription-attributes",57            {58                "Attributes": {59                    "ConfirmationWasAuthenticated": "true",60                    "Owner": "111111111111",61                    "PendingConfirmation": "false",62                    "Protocol": "lambda",63                    "RawMessageDelivery": "false",64                    "RedrivePolicy": {65                        "deadLetterTargetArn": "arn:aws:sqs:region:111111111111:111112222233333"66                    },67                    "TopicArn": "arn:aws:sns:region:111111111111:rrrrrrrrrrrrrrrrr",68                    "SubscriptionArn": "arn:aws:sns:region:111111111111:rrrrrrrrrrrrrrrrr:azazazazazazazaza",69                    "Endpoint": "arn:aws:lambda:region:111111111111:function:aaaaabbbbb",70                },71                "ResponseMetadata": {"HTTPHeaders": {}, "HTTPStatusCode": 200},72            },73        )74        sm._assert_all()75    def test_replacement_key_value(self):76        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)77        sm.add_transformer(78            KeyValueBasedTransformer(79                # returns last two characters of value -> only this should be replaced80                lambda k, v: v[-2:] if k == "aaa" else None,81                replacement="A",82                replace_reference=False,83            )84        )85        sm.recorded_state = {86            "key_a": {"aaa": "hellA", "aab": "this is a test", "b": {"aaa": "another teA"}}87        }88        sm.match("key_a", {"aaa": "helloo", "aab": "this is a test", "b": {"aaa": "another test"}})89        sm._assert_all()90    def test_dot_in_skip_verification_path(self):91        sm = SnapshotSession(scope_key="A", verify=True, file_path="", update=False)92        sm.recorded_state = {93            "key_a": {"aaa": "hello", "aab": "this is a test", "b": {"a.aa": "another test"}}94        }95        sm.match(96            "key_a",97            {"aaa": "hello", "aab": "this is a test-fail", "b": {"a.aa": "another test-fail"}},98        )99        with pytest.raises(Exception) as ctx:  # asserts it fail without skipping100            sm._assert_all()101        ctx.match("Parity snapshot failed")102        skip_path = ["$..aab", "$..b.a.aa"]103        with pytest.raises(Exception) as ctx:  # asserts it fails if fields are not escaped104            sm._assert_all(skip_verification_paths=skip_path)105        ctx.match("Parity snapshot failed")106        skip_path_escaped = ["$..aab", "$..b.'a.aa'"]107        sm._assert_all(skip_verification_paths=skip_path_escaped)108def test_sorting_transformer():109    original_dict = {110        "a": {111            "b": [112                {"name": "c-123"},113                {"name": "a-123"},114                {"name": "b-123"},115            ]116        },117        "a2": {118            "b": [119                {"name": "b-123"},120                {"name": "a-123"},121                {"name": "c-123"},122            ]123        },124    }125    sorted_items = [126        {"name": "a-123"},127        {"name": "b-123"},128        {"name": "c-123"},129    ]130    transformer = SortingTransformer("b", lambda x: x["name"])131    transformed_dict = transformer.transform(original_dict)132    assert transformed_dict["a"]["b"] == sorted_items...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.