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

How to use wait_is_container_running method in localstack

Best Python code snippet using localstack_python

bootstrap.py

Source:bootstrap.py

...396        # also checks the edge port health status397        return super().is_up()398    def is_container_running(self) -> bool:399        return DOCKER_CLIENT.is_container_running(self.container.name)400    def wait_is_container_running(self, timeout=None) -> bool:401        return poll_condition(self.is_container_running, timeout)402    def do_run(self):403        if DOCKER_CLIENT.is_container_running(self.container.name):404            raise ContainerExists(405                'LocalStack container named "%s" is already running' % self.container.name406            )407        return self.container.run()408    def do_shutdown(self):409        try:410            CmdDockerClient().stop_container(411                self.container.name, timeout=10412            )  # giving the container some time to stop413        except Exception as e:414            LOG.info("error cleaning up localstack container %s: %s", self.container.name, e)415class ContainerExists(Exception):416    pass417def prepare_docker_start():418    # prepare environment for docker start419    container_name = config.MAIN_CONTAINER_NAME420    if DOCKER_CLIENT.is_container_running(container_name):421        raise ContainerExists('LocalStack container named "%s" is already running' % container_name)422    if config.dirs.tmp != config.dirs.functions and not config.LAMBDA_REMOTE_DOCKER:423        # Logger is not initialized at this point, so the warning is displayed via print424        print(425            f"WARNING: The detected temp folder for localstack ({config.dirs.tmp}) is not equal to the "426            f"HOST_TMP_FOLDER environment variable set ({config.dirs.functions})."427        )428    os.environ[ENV_SCRIPT_STARTING_DOCKER] = "1"429    # make sure temp folder exists430    mkdir(config.dirs.tmp)431    try:432        chmod_r(config.dirs.tmp, 0o777)433    except Exception:434        pass435def configure_container(container: LocalstackContainer):436    """437    Configuration routine for the LocalstackContainer.438    """439    # get additional configured flags440    user_flags = config.DOCKER_FLAGS441    user_flags = extract_port_flags(user_flags, container.ports)442    container.additional_flags.extend(shlex.split(user_flags))443    # get additional parameters from plugins444    hooks.configure_localstack_container.run(container)445    # construct default port mappings446    service_ports = config.SERVICE_PORTS447    if service_ports.get("edge") == 0:448        service_ports.pop("edge")449    for port in service_ports.values():450        if port:451            container.ports.add(port)452    for port in range(config.EXTERNAL_SERVICE_PORTS_START, config.EXTERNAL_SERVICE_PORTS_END):453        container.ports.add(port)454    if config.DEVELOP:455        container.ports.add(config.DEVELOP_PORT)456    # environment variables457    # pass through environment variables defined in config458    for env_var in config.CONFIG_ENV_VARS:459        value = os.environ.get(env_var, None)460        if value is not None:461            container.env_vars[env_var] = value462    container.env_vars["DOCKER_HOST"] = f"unix://{config.DOCKER_SOCK}"463    container.env_vars["HOST_TMP_FOLDER"] = config.dirs.functions  # TODO: rename env var464    # TODO this is default now, remove once a considerate time is passed465    # to activate proper signal handling466    container.env_vars["SET_TERM_HANDLER"] = "1"467    configure_volume_mounts(container)468    # mount docker socket469    container.volumes.append((config.DOCKER_SOCK, config.DOCKER_SOCK))470    container.additional_flags.append("--privileged")471def configure_volume_mounts(container: LocalstackContainer):472    if not config.LEGACY_DIRECTORIES:473        container.volumes.add(VolumeBind(config.VOLUME_DIR, DEFAULT_VOLUME_DIR))474        return475    source_dirs = config.dirs476    target_dirs = Directories.legacy_for_container()477    # default shared directories478    for name in Directories.default_bind_mounts:479        src = getattr(source_dirs, name, None)480        target = getattr(target_dirs, name, None)481        if src and target:482            container.volumes.add(VolumeBind(src, target))483    # shared tmp folder484    container.volumes.add(VolumeBind(source_dirs.tmp, target_dirs.tmp))485    # data_dir mounting and environment variables486    if source_dirs.data:487        container.volumes.add(VolumeBind(source_dirs.data, target_dirs.data))488        container.env_vars["DATA_DIR"] = target_dirs.data489@log_duration()490def prepare_host():491    """492    Prepare the host environment for running LocalStack, this should be called before start_infra_*.493    """494    if os.environ.get(constants.LOCALSTACK_INFRA_PROCESS) in constants.TRUE_STRINGS:495        return496    setup_logging()497    hooks.prepare_host.run()498def start_infra_in_docker():499    prepare_docker_start()500    container = LocalstackContainer()501    # create and prepare container502    configure_container(container)503    container.truncate_log()504    # printing the container log is the current way we're occupying the terminal505    log_printer = FileListener(container.logfile, print)506    log_printer.start()507    # Set up signal handler, to enable clean shutdown across different operating systems.508    #  There are subtle differences across operating systems and terminal emulators when it509    #  comes to handling of CTRL-C - in particular, Linux sends SIGINT to the parent process,510    #  whereas MacOS sends SIGINT to the process group, which can result in multiple SIGINT signals511    #  being received (e.g., when running the localstack CLI as part of an "npm run .." script).512    #  Hence, using a shutdown handler and synchronization event here, to avoid inconsistencies.513    def shutdown_handler(*args):514        with shutdown_event_lock:515            if shutdown_event.is_set():516                return517            shutdown_event.set()518        print("Shutting down...")519        server.shutdown()520        log_printer.close()521    shutdown_event = threading.Event()522    shutdown_event_lock = threading.RLock()523    signal.signal(signal.SIGINT, shutdown_handler)524    # start the Localstack container as a Server525    server = LocalstackContainerServer(container)526    try:527        server.start()528        server.join()529    except KeyboardInterrupt:530        print("ok, bye!")531        shutdown_handler()532def start_infra_in_docker_detached(console):533    """534    An alternative to start_infra_in_docker where the terminal is not blocked by the follow on the logfile.535    """536    console.log("preparing environment")537    try:538        prepare_docker_start()539    except ContainerExists as e:540        console.print(str(e))541        return542    # create and prepare container543    console.log("configuring container")544    container = LocalstackContainer()545    configure_container(container)546    container.truncate_log()547    # start the Localstack container as a Server548    console.log("starting container")549    server = LocalstackContainerServer(container)550    server.start()551    server.wait_is_container_running()552    console.log("detaching")553def wait_container_is_ready(timeout: Optional[float] = None):554    """Blocks until the localstack main container is running and the ready marker has been printed."""555    container_name = config.MAIN_CONTAINER_NAME556    started = time.time()557    def is_container_running():558        return DOCKER_CLIENT.is_container_running(container_name)559    if not poll_condition(is_container_running, timeout=timeout):560        return False561    stream = DOCKER_CLIENT.stream_container_logs(container_name)562    # create a timer that will terminate the log stream after the remaining timeout563    timer = None564    if timeout:565        waited = time.time() - started...

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.