Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _get_events_tmp_dir method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...137    # maps rule name to job_id138    rule_scheduled_jobs: Dict[str, str]139    def __init__(self):140        self.rule_scheduled_jobs = {}141def _get_events_tmp_dir():142    return os.path.join(config.dirs.tmp, EVENTS_TMP_DIR)143def _create_and_register_temp_dir():144    tmp_dir = _get_events_tmp_dir()145    if not os.path.exists(tmp_dir):146        mkdir(tmp_dir)147        TMP_FILES.append(tmp_dir)148    return tmp_dir149def _dump_events_to_files(events_with_added_uuid):150    try:151        _create_and_register_temp_dir()152        current_time_millis = int(round(time.time() * 1000))153        for event in events_with_added_uuid:154            target = os.path.join(155                _get_events_tmp_dir(),156                "%s_%s" % (current_time_millis, event["uuid"]),157            )158            save_file(target, json.dumps(event["event"]))159    except Exception as e:160        LOG.info("Unable to dump events to tmp dir %s: %s", _get_events_tmp_dir(), e)161def handle_numeric_conditions(conditions: List[Any], value: float):162    for i in range(0, len(conditions), 2):163        if conditions[i] == "<" and not (value < conditions[i + 1]):164            return False165        if conditions[i] == ">" and not (value > conditions[i + 1]):166            return False167        if conditions[i] == "<=" and not (value <= conditions[i + 1]):168            return False169        if conditions[i] == ">=" and not (value >= conditions[i + 1]):170            return False171    return True172def check_valid_numeric_content_base_rule(list_of_operators):173    if len(list_of_operators) > 4:174        return False...

events_listener.py

Source:events_listener.py

...38    pattern = r"<CreateDate>([^<]+) ([^<+]+)(\+[^<]*)?</CreateDate>"39    replacement = r"<CreateDate>\1T\2Z</CreateDate>"40    replace_response_content(response, pattern, replacement)41def _create_and_register_temp_dir():42    tmp_dir = _get_events_tmp_dir()43    if not os.path.exists(tmp_dir):44        mkdir(tmp_dir)45        TMP_FILES.append(tmp_dir)46    return tmp_dir47def _dump_events_to_files(events_with_added_uuid):48    try:49        _create_and_register_temp_dir()50        current_time_millis = int(round(time.time() * 1000))51        for event in events_with_added_uuid:52            target = os.path.join(53                _get_events_tmp_dir(), "%s_%s" % (current_time_millis, event["uuid"])54            )55            save_file(target, json.dumps(event["event"]))56    except Exception as e:57        LOG.info("Unable to dump events to tmp dir %s: %s", _get_events_tmp_dir(), e)58def _get_events_tmp_dir():59    return os.path.join(config.dirs.tmp, EVENTS_TMP_DIR)60def get_scheduled_rule_func(data):61    def func(*args, **kwargs):62        rule_name = data.get("Name")63        client = aws_stack.connect_to_service("events")64        targets = client.list_targets_by_rule(Rule=rule_name)["Targets"]65        if targets:66            LOG.debug(67                "Notifying %s targets in response to triggered Events rule %s",68                len(targets),69                rule_name,70            )71        for target in targets:72            arn = target.get("Arn")...

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.