KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use build_mapping_obj method in localstack

Best Python code snippet using localstack_python

lambda_api.py

Source:lambda_api.py

...174            INVALID_PARAMETER_VALUE_EXCEPTION,175            "BatchSize {} exceeds the max of {}".format(batch_size, batch_size_entry[1]),176        )177    return batch_size178def build_mapping_obj(data) -> Dict:179    mapping = {}180    function_name = data["FunctionName"]181    enabled = data.get("Enabled", True)182    batch_size = data.get("BatchSize")183    mapping["UUID"] = str(uuid.uuid4())184    mapping["FunctionArn"] = func_arn(function_name)185    mapping["LastProcessingResult"] = "OK"186    mapping["StateTransitionReason"] = "User action"187    mapping["LastModified"] = format_timestamp_for_event_source_mapping()188    mapping["State"] = "Enabled" if enabled in [True, None] else "Disabled"189    mapping["ParallelizationFactor"] = data.get("ParallelizationFactor") or 1190    mapping["Topics"] = data.get("Topics") or []191    mapping["MaximumRetryAttempts"] = data.get("MaximumRetryAttempts") or -1192    if "SelfManagedEventSource" in data:193        source_arn = data["SourceAccessConfigurations"][0]["URI"]194        mapping["SelfManagedEventSource"] = data["SelfManagedEventSource"]195        mapping["SourceAccessConfigurations"] = data["SourceAccessConfigurations"]196    else:197        source_arn = data["EventSourceArn"]198        mapping["EventSourceArn"] = source_arn199        mapping["StartingPosition"] = data.get("StartingPosition") or "LATEST"200    batch_size = check_batch_size_range(source_arn, batch_size)201    mapping["BatchSize"] = batch_size202    if data.get("DestinationConfig"):203        mapping["DestinationConfig"] = data.get("DestinationConfig")204    return mapping205def format_timestamp(timestamp=None):206    timestamp = timestamp or datetime.utcnow()207    return isoformat_milliseconds(timestamp) + "+0000"208def format_timestamp_for_event_source_mapping():209    # event source mappings seem to use a different time format (required for Terraform compat.)210    return datetime.utcnow().timestamp()211def add_event_source(data):212    region = LambdaRegion.get()213    mapping = build_mapping_obj(data)214    region.event_source_mappings.append(mapping)215    EventSourceListener.start_listeners(mapping)216    return mapping217def update_event_source(uuid_value, data):218    region = LambdaRegion.get()219    function_name = data.get("FunctionName") or ""220    enabled = data.get("Enabled", True)221    for mapping in region.event_source_mappings:222        if uuid_value == mapping["UUID"]:223            if function_name:224                mapping["FunctionArn"] = func_arn(function_name)225            batch_size = data.get("BatchSize")226            if "SelfManagedEventSource" in mapping:227                batch_size = check_batch_size_range(...

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.