How to use run_for_max_seconds method in localstack

Best Python code snippet using localstack_python

common.py

Source:common.py

1from localstack import config2# TODO: remove imports from here (need to update any client code that imports these from utils.common)3from localstack.utils.archives import get_unzipped_size, is_zip_file, untar, unzip  # noqa4# TODO: remove imports from here (need to update any client code that imports these from utils.common)5from localstack.utils.collections import (  # noqa6    DelSafeDict,7    HashableList,8    PaginatedList,9    ensure_list,10    is_list_or_tuple,11    is_none_or_empty,12    is_sub_dict,13    items_equivalent,14    last_index_of,15    merge_dicts,16    merge_recursive,17    remove_attributes,18    remove_none_values_from_dict,19    rename_attributes,20    select_attributes,21    to_unique_items_list,22)23# TODO: remove imports from here (need to update any client code that imports these from utils.common)24from localstack.utils.crypto import (  # noqa25    PEM_CERT_END,26    PEM_CERT_START,27    PEM_KEY_END_REGEX,28    PEM_KEY_START_REGEX,29    generate_ssl_cert,30)31# TODO: remove imports from here (need to update any client code that imports these from utils.common)32from localstack.utils.files import (  # noqa33    TMP_FILES,34    chmod_r,35    chown_r,36    cleanup_tmp_files,37    cp_r,38    disk_usage,39    ensure_readable,40    file_exists_not_empty,41    get_or_create_file,42    is_empty_dir,43    load_file,44    mkdir,45    new_tmp_dir,46    new_tmp_file,47    replace_in_file,48    rm_rf,49    save_file,50)51# TODO: remove imports from here (need to update any client code that imports these from utils.common)52from localstack.utils.functions import (  # noqa53    call_safe,54    empty_context_manager,55    prevent_stack_overflow,56    run_safe,57)58# TODO: remove imports from here (need to update any client code that imports these from utils.common)59from localstack.utils.http import (  # noqa60    NetrcBypassAuth,61    _RequestsSafe,62    download,63    get_proxies,64    make_http_request,65    parse_request_data,66    replace_response_content,67    safe_requests,68)69# TODO: remove imports from here (need to update any client code that imports these from utils.common)70from localstack.utils.json import (  # noqa71    CustomEncoder,72    FileMappedDocument,73    JsonObject,74    assign_to_path,75    canonical_json,76    clone,77    clone_safe,78    extract_from_jsonpointer_path,79    extract_jsonpath,80    fix_json_keys,81    json_safe,82    parse_json_or_yaml,83    try_json,84)85# TODO: remove imports from here (need to update any client code that imports these from utils.common)86from localstack.utils.net import (  # noqa87    PortNotAvailableException,88    PortRange,89    get_free_tcp_port,90    is_ip_address,91    is_ipv4_address,92    is_port_open,93    port_can_be_bound,94    resolve_hostname,95    wait_for_port_closed,96    wait_for_port_open,97    wait_for_port_status,98)99# TODO: remove imports from here (need to update any client code that imports these from utils.common)100from localstack.utils.numbers import format_bytes, format_number, is_number  # noqa101# TODO: remove imports from here (need to update any client code that imports these from utils.common)102from localstack.utils.objects import (  # noqa103    ArbitraryAccessObj,104    Mock,105    ObjectIdHashComparator,106    SubtypesInstanceManager,107    fully_qualified_class_name,108    get_all_subclasses,109    keys_to_lower,110    not_none_or,111    recurse_object,112)113# TODO: remove imports from here (need to update any client code that imports these from utils.common)114from localstack.utils.platform import (  # noqa115    get_arch,116    get_os,117    in_docker,118    is_debian,119    is_linux,120    is_mac_os,121    is_windows,122)123# TODO: remove imports from here (need to update any client code that imports these from utils.common)124from localstack.utils.run import (  # noqa125    CaptureOutput,126    ShellCommandThread,127    get_os_user,128    is_command_available,129    is_root,130    kill_process_tree,131    run,132    run_for_max_seconds,133)134# TODO: remove imports from here (need to update any client code that imports these from utils.common)135from localstack.utils.strings import (  # noqa136    base64_to_hex,137    camel_to_snake_case,138    canonicalize_bool_to_str,139    convert_to_printable_chars,140    first_char_to_lower,141    first_char_to_upper,142    is_base64,143    is_string,144    is_string_or_bytes,145    long_uid,146    md5,147    short_uid,148    snake_to_camel_case,149    str_insert,150    str_remove,151    str_startswith_ignore_case,152    str_to_bool,153    to_bytes,154    to_str,155    truncate,156)157# TODO: remove imports from here (need to update any client code that imports these from utils.common)158from localstack.utils.sync import (  # noqa159    poll_condition,160    retry,161    sleep_forever,162    synchronized,163    wait_until,164)165# TODO: remove imports from here (need to update any client code that imports these from utils.common)166from localstack.utils.tail import FileListener  # noqa167# TODO: remove imports from here (need to update any client code that imports these from utils.common)168from localstack.utils.threads import (  # noqa169    TMP_PROCESSES,170    TMP_THREADS,171    FuncThread,172    cleanup_threads_and_processes,173    parallelize,174    start_thread,175    start_worker_thread,176)177# TODO: remove imports from here (need to update any client code that imports these from utils.common)178from localstack.utils.time import (  # noqa179    TIMESTAMP_FORMAT,180    TIMESTAMP_FORMAT_MICROS,181    TIMESTAMP_FORMAT_TZ,182    epoch_timestamp,183    isoformat_milliseconds,184    mktime,185    now,186    now_utc,187    parse_timestamp,188    timestamp,189    timestamp_millis,190)191# TODO: remove imports from here (need to update any client code that imports these from utils.common)192from localstack.utils.urls import path_from_url  # noqa193# TODO: remove imports from here (need to update any client code that imports these from utils.common)194from localstack.utils.xml import obj_to_xml, strip_xmlns  # noqa195# TODO: move somewhere sensible (probably localstack.runtime)196class ExternalServicePortsManager(PortRange):197    """Manages the ports used for starting external services like ElasticSearch, OpenSearch,..."""198    def __init__(self):199        super().__init__(config.EXTERNAL_SERVICE_PORTS_START, config.EXTERNAL_SERVICE_PORTS_END)200external_service_ports = ExternalServicePortsManager()201# TODO: replace references with config.get_protocol/config.edge_ports_info202get_service_protocol = config.get_protocol203# TODO: replace references to safe_run with localstack.utils.run.run...

completion.py

Source:completion.py

1import os2import time3from datetime import timedelta4import warnings5from envs.graph_env import GraphEnv6from utils import draw_graph7from stable_baselines3 import PPO89warnings.simplefilter(action='ignore', category=FutureWarning)101112def complete_graph(graph, model_id, run_for_min_seconds=5, run_for_max_seconds=10, draw_generated_graphs=False):13  timerStart = time.time()14  counter = 01516  # Init gym graph environment17  env = GraphEnv()18  env.init(base_graph=graph)1920  # Load model21  model = PPO.load(f"{os.path.join(os.path.dirname(os.path.realpath(__file__)), '../..')}/saved_models/{model_id}/best_model.zip", env=env)2223  # Generate multiple graphs24  generated_graphs = []2526  print(f"Graph completion started (model {model_id}).")2728  while True:29    done = False30    obs = env.reset()3132    while not done:33      action, _ = model.predict(obs, deterministic=False)34      obs, _, done, info = env.step(action)3536      current_time = time.time()-timerStart37      if current_time >= run_for_min_seconds and generated_graphs:38        print("")39        print(f"Finished after {timedelta(seconds=current_time)}")4041        print("Graph completion succeeded!")4243        # Sort generated graphs by score44        generated_graphs.sort(key=lambda graph: graph[0], reverse=True)45        46        print(f"{len(generated_graphs)}/{counter} graphs were valid.")47        print(f"Best score was {generated_graphs[0][0]}.")4849        # Return best graph50        return generated_graphs[0]5152      if current_time >= run_for_max_seconds:53        print("")54        print(f"Finished after {timedelta(seconds=current_time)}")5556        print("Graph completion failed!")5758        return False5960      if done:61        counter = counter+16263        if info["graph_valid"]:64          print("â", end="")65          generated_graphs.append((info["mood_score"], info["graph"]))66          if draw_generated_graphs == "only_valid":67            draw_graph(info["graph"])68        else:69          print(".", end="")7071        if draw_generated_graphs == "all":72          draw_graph(info["graph"], layout=None)7374  # should never be reached
...

complete_graph.py

Source:complete_graph.py

1from fastapi import APIRouter23from data import base_graph4from completion import complete_graph5from utils import draw_graph67router = APIRouter()89@router.get("/complete-graph/{edge_string}", tags=["complete_graph"])10def get_completed_graph(edge_string: str):11  '''12  Completes the graph with nodes:1314  - 0 - Michel15  - 1 - Elias16  - 2 - Sofia17  - 3 - Jorge18  - 4 - Juan19  - 5 - Manuel20  - 6 - Laura21  - 7 - Stefan22  - 8 - Robin23  - 9 - Daniel2425  and edges given via edge_string parameter.2627  The edge_string parameter should come in the format used in the following examples:2829  - `/complete-graph/0-1_1-2_8-9` (translates into array of 3 edges `[(0, 1), (1, 2), (8, 9)]`)30  - `/complete-graph/-` (translates into empty array of 0 edges `[]`)31  32  '''3334  if edge_string == "-":35    edges = []36  else:37    edges = [edge for edge in edge_string.split("_")]38    edges = [(int(edge.split("-")[0]), int(edge.split("-")[1])) for edge in edges]3940  # use nodes from base_graph41  input_graph = base_graph42  # add given edges to achieve graph from interface43  input_graph.add_edges_from(edges)4445  # generate completed graph, sorted by descending mood scores46  completed_graphs = complete_graph(input_graph, "2022-06-28_09-50-50", run_for_min_seconds=5, run_for_max_seconds=120)4748  if completed_graphs:49    best_graph = completed_graphs[1]50    mood_score = completed_graphs[0]51    response = {52      "status": "success",53      "completed_graph_edge_list": list(best_graph.edges),54      "completed_graph_mood_score": mood_score,55    }56    draw_graph(best_graph)57  else:58    response = {59      "status": "failure",60      "completed_graph_edge_list": [],61      "completed_graph_mood_score": 0,62    }63  
...

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.