Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -744,36 +744,6 @@ def stepfunctions_api():
# def custom(fn: Callable[[dict], dict]) -> Transformer:
# return GenericTransformer(fn)

@staticmethod
def eventbridge_api_destination(snapshot, connection_name: str):
"""
Add common transformers for EventBridge connection tests.

Args:
snapshot: The snapshot instance to add transformers to
connection_name: The name of the connection to transform in the snapshot
"""
snapshot.add_transformer(snapshot.transform.regex(connection_name, "<connection-name>"))
snapshot.add_transformer(
snapshot.transform.key_value("ApiDestinationArn", reference_replacement=False)
)
snapshot.add_transformer(
snapshot.transform.key_value("ConnectionArn", reference_replacement=False)
)
return snapshot

@staticmethod
def eventbridge_connection(snapshot, connection_name: str):
"""
Add common transformers for EventBridge connection tests.
Args:
snapshot: The snapshot instance to add transformers to
connection_name: The name of the connection to transform in the snapshot
"""
snapshot.add_transformer(snapshot.transform.regex(connection_name, "<connection-name>"))
snapshot.add_transformer(TransformerUtility.resource_name())
return snapshot


def _sns_pem_file_token_transformer(key: str, val: str) -> str:
if isinstance(val, str) and key.lower() == "SigningCertURL".lower():
Expand Down
29 changes: 29 additions & 0 deletions tests/aws/services/events/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import pytest

from localstack.testing.snapshots.transformer_utility import TransformerUtility
from localstack.utils.aws.arns import get_partition
from localstack.utils.strings import short_uid
from localstack.utils.sync import retry
Expand Down Expand Up @@ -445,3 +446,31 @@ def _create_api_destination(**kwargs):
)

return _create_api_destination


#############################
# Common Transformer Fixtures
#############################


@pytest.fixture
def api_destination_snapshot(snapshot, destination_name):
snapshot.add_transformers_list(
[
snapshot.transform.regex(destination_name, "<destination-name>"),
snapshot.transform.key_value("ApiDestinationArn", reference_replacement=False),
snapshot.transform.key_value("ConnectionArn", reference_replacement=False),
]
)
return snapshot


@pytest.fixture
def connection_snapshot(snapshot, connection_name):
snapshot.add_transformers_list(
[
snapshot.transform.regex(connection_name, "<connection-name>"),
TransformerUtility.resource_name(),
]
)
return snapshot
Loading
Loading