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

How to use _run_async_cmd method in localstack

Best Python code snippet using localstack_python

docker_cmd_client.py

Source:docker_cmd_client.py

...404            Util.rm_env_vars_file(env_file)405    def run_container(self, image_name: str, stdin=None, **kwargs) -> Tuple[bytes, bytes]:406        cmd, env_file = self._build_run_create_cmd("run", image_name, **kwargs)407        LOG.debug("Run container with cmd: %s", cmd)408        result = self._run_async_cmd(cmd, stdin, kwargs.get("name") or "", image_name)409        Util.rm_env_vars_file(env_file)410        return result411    def exec_in_container(412        self,413        container_name_or_id: str,414        command: Union[List[str], str],415        interactive=False,416        detach=False,417        env_vars: Optional[Dict[str, Optional[str]]] = None,418        stdin: Optional[bytes] = None,419        user: Optional[str] = None,420        workdir: Optional[str] = None,421    ) -> Tuple[bytes, bytes]:422        env_file = None423        cmd = self._docker_cmd()424        cmd.append("exec")425        if interactive:426            cmd.append("--interactive")427        if detach:428            cmd.append("--detach")429        if user:430            cmd += ["--user", user]431        if workdir:432            cmd += ["--workdir", workdir]433        if env_vars:434            env_flag, env_file = Util.create_env_vars_file_flag(env_vars)435            cmd += env_flag436        cmd.append(container_name_or_id)437        cmd += command if isinstance(command, List) else [command]438        LOG.debug("Execute in container cmd: %s", cmd)439        result = self._run_async_cmd(cmd, stdin, container_name_or_id)440        Util.rm_env_vars_file(env_file)441        return result442    def start_container(443        self,444        container_name_or_id: str,445        stdin=None,446        interactive: bool = False,447        attach: bool = False,448        flags: Optional[str] = None,449    ) -> Tuple[bytes, bytes]:450        cmd = self._docker_cmd() + ["start"]451        if flags:452            cmd.append(flags)453        if interactive:454            cmd.append("--interactive")455        if attach:456            cmd.append("--attach")457        cmd.append(container_name_or_id)458        LOG.debug("Start container with cmd: %s", cmd)459        return self._run_async_cmd(cmd, stdin, container_name_or_id)460    def _run_async_cmd(461        self, cmd: List[str], stdin: bytes, container_name: str, image_name=None462    ) -> Tuple[bytes, bytes]:463        kwargs = {464            "inherit_env": True,465            "asynchronous": True,466            "stderr": subprocess.PIPE,467            "outfile": self.default_run_outfile or subprocess.PIPE,468        }469        if stdin:470            kwargs["stdin"] = True471        try:472            process = run(cmd, **kwargs)473            stdout, stderr = process.communicate(input=stdin)474            if process.returncode != 0:...

cli.py

Source:cli.py

...84        return loop.run_until_complete(coro)85    except KeyboardInterrupt:86        print("Exiting")87    loop.close()88def _run_async_cmd(cmd: Callable[[], Any]) -> Any:89    """Run the coroutine and return its return value."""90    loop = asyncio.get_event_loop()91    _run_async(loop, cmd())92def _run_cmd(cmd: Callable[[], Any]) -> None:93    """Run a WAMP command.94    Uses `_run_async_cmd` to run the command, but handles return values95    and exceptions.96    """97    try:98        result = _run_async_cmd(cmd)99    except aiowamp.ErrorResponse as e:100        sys.exit(str(e))101    else:102        print(libwampli.human_result(result))103def _call_cmd(args: argparse.Namespace) -> None:104    """Call command."""105    async def cmd() -> Any:106        async with get_client_context(args) as client:107            return await client.call(args.uri, *call_args, kwargs=call_kwargs)108    call_args, call_kwargs = libwampli.parse_args(args.args)109    _run_cmd(cmd)110def _publish_cmd(args: argparse.Namespace) -> None:111    """Publish command."""112    async def cmd() -> None:...

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.