How to use filter_event_with_target_input_path method in localstack

Best Python code snippet using localstack_python

events_starter.py

Source:events_starter.py

...16from localstack.utils.aws.message_forwarding import send_event_to_target17from localstack.utils.common import extract_jsonpath, short_uid, truncate18LOG = logging.getLogger(__name__)19CONTENT_BASE_FILTER_KEYWORDS = ["prefix", "anything-but", "numeric", "cidr", "exists"]20def filter_event_with_target_input_path(target: Dict, event: Dict) -> Dict:21    input_path = target.get("InputPath")22    if input_path:23        event = extract_jsonpath(event, input_path)24    return event25# TODO: unclear shared responsibility for filtering with filter_event_with_content_base_parameter26def handle_prefix_filtering(event_pattern, value):27    for element in event_pattern:28        if isinstance(element, (int, str)):29            if str(element) == str(value):30                return True31        elif isinstance(element, dict) and "prefix" in element:32            if value.startswith(element.get("prefix")):33                return True34        elif isinstance(element, dict) and "anything-but" in element:35            if element.get("anything-but") != value:36                return True37        elif "numeric" in element:38            return handle_numeric_conditions(element.get("numeric"), value)39        elif isinstance(element, list):40            if value in list:41                return True42    return False43def handle_numeric_conditions(conditions: List[Any], value: float):44    for i in range(0, len(conditions), 2):45        if conditions[i] == "<" and not (value < conditions[i + 1]):46            return False47        if conditions[i] == ">" and not (value > conditions[i + 1]):48            return False49        if conditions[i] == "<=" and not (value <= conditions[i + 1]):50            return False51        if conditions[i] == ">=" and not (value >= conditions[i + 1]):52            return False53    return True54# TODO: refactor/simplify55def filter_event_based_on_event_format(self, rule_name: str, event: Dict[str, Any]):56    def filter_event(event_pattern_filter: Dict[str, Any], event: Dict[str, Any]):57        for key, value in event_pattern_filter.items():58            event_value = event.get(key.lower())59            if event_value is None:60                return False61            if event_value and isinstance(event_value, dict):62                for key_a, value_a in event_value.items():63                    if key_a == "ip":64                        # TODO add IP-Address check here65                        continue66                    if isinstance(value.get(key_a), (int, str)):67                        if value_a != value.get(key_a):68                            return False69                    if isinstance(value.get(key_a), list) and value_a not in value.get(key_a):70                        if not handle_prefix_filtering(value.get(key_a), value_a):71                            return False72            elif isinstance(value, list) and not identify_content_base_parameter_in_pattern(value):73                if (74                    isinstance(event_value, list)75                    and get_two_lists_intersection(value, event_value) == []76                ):77                    return False78                elif (79                    not isinstance(event_value, list)80                    and isinstance(event_value, (str, int))81                    and event_value not in value82                ):83                    return False84            elif isinstance(value, list) and identify_content_base_parameter_in_pattern(value):85                if not filter_event_with_content_base_parameter(value, event_value):86                    return False87            elif isinstance(value, (str, dict)):88                try:89                    value = json.loads(value) if isinstance(value, str) else value90                    if isinstance(value, dict) and not filter_event(value, event_value):91                        return False92                except json.decoder.JSONDecodeError:93                    return False94        return True95    rule_information = self.events_backend.describe_rule(rule_name)96    if not rule_information:97        LOG.info('Unable to find rule "%s" in backend: %s', rule_name, rule_information)98        return False99    if rule_information.event_pattern._pattern:100        event_pattern = rule_information.event_pattern._pattern101        if not filter_event(event_pattern, event):102            return False103    return True104def process_events(event: Dict, targets: List[Dict]):105    for target in targets:106        arn = target["Arn"]107        changed_event = filter_event_with_target_input_path(target, event)108        try:109            send_event_to_target(arn, changed_event, aws_stack.get_events_target_attributes(target))110        except Exception as e:111            LOG.info(f"Unable to send event notification {truncate(event)} to target {target}: {e}")112def apply_patches():113    # Fix events ARN114    def rule_model_generate_arn(self, name):115        return "arn:aws:events:{region_name}:{account_id}:rule/{name}".format(116            region_name=self.region_name, account_id=TEST_AWS_ACCOUNT_ID, name=name117        )118    # specific logic for put_events which forwards matching events to target listeners119    def events_handler_put_events(self):120        entries = self._get_param("Entries")121        events = list(map(lambda event: {"event": event, "uuid": str(uuid.uuid4())}, entries))...

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.