How to use filter_event_based_on_event_format method in localstack

Best Python code snippet using localstack_python

events_starter.py

Source:events_starter.py

...51        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))122        _dump_events_to_files(events)123        event_rules = self.events_backend.rules124        for event_envelope in events:125            event = event_envelope["event"]126            event_bus = event.get("EventBusName") or DEFAULT_EVENT_BUS_NAME127            matchine_rules = [r for r in event_rules.values() if r.event_bus_name == event_bus]128            if not matchine_rules:129                continue130            formatted_event = {131                "version": "0",132                "id": event_envelope["uuid"],133                "detail-type": event.get("DetailType"),134                "source": event.get("Source"),135                "account": TEST_AWS_ACCOUNT_ID,136                "time": datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),137                "region": self.region,138                "resources": event.get("Resources", []),139                "detail": json.loads(event.get("Detail", "{}")),140            }141            targets = []142            for rule in matchine_rules:143                if filter_event_based_on_event_format(self, rule.name, formatted_event):144                    targets.extend(self.events_backend.list_targets_by_rule(rule.name)["Targets"])145            # process event146            process_events(formatted_event, targets)147        content = {148            "FailedEntryCount": 0,  # TODO: dynamically set proper value when refactoring149            "Entries": list(map(lambda event: {"EventId": event["uuid"]}, events)),150        }151        self.response_headers.update(152            {"Content-Type": APPLICATION_AMZ_JSON_1_1, "x-amzn-RequestId": short_uid()}153        )154        return json.dumps(content), self.response_headers155    rule_model._generate_arn = rule_model_generate_arn156    events_handler.put_events = events_handler_put_events157def start_scheduler():...

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.