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

How to use _create_container_for_task method in avocado

Best Python code snippet using avocado_python

podman.py

Source:podman.py

...146        major, minor, _ = await self.python_version147        eggs = self.get_eggs_paths(major, minor)148        for egg, to in eggs:149            await self.podman.copy_to_container(where, egg, to)150    async def _create_container_for_task(151        self, runtime_task, env_args, test_output=None152    ):153        mount_status_server_socket = False154        mounted_status_server_socket = "/tmp/.status_server.sock"155        status_server_uri = runtime_task.task.status_services[0].uri156        if ":" not in status_server_uri:157            # a unix domain socket is being used158            mount_status_server_socket = True159            runtime_task.task.status_services[0].uri = mounted_status_server_socket160        _, _, python_binary = await self.python_version161        entry_point_args = [python_binary, "-m", "avocado.core.nrunner", "task-run"]162        test_opts = ()163        if runtime_task.task.category == "test":164            runnable_uri = runtime_task.task.runnable.uri165            try:166                test_path, _ = runnable_uri.split(":", 1)167            except ValueError:168                test_path = runnable_uri169            if os.path.exists(test_path):170                to = os.path.join("/tmp", test_path)171                runtime_task.task.runnable.uri = os.path.join("/tmp", runnable_uri)172                test_opts = ("-v", f"{os.path.abspath(test_path)}:{to}:ro")173        task = runtime_task.task174        entry_point_args.extend(task.get_command_args())175        entry_point = json.dumps(entry_point_args)176        entry_point_arg = "--entrypoint=" + entry_point177        if mount_status_server_socket:178            status_server_opts = (179                "--privileged",180                "-v",181                f"{status_server_uri}:{mounted_status_server_socket}",182            )183        else:184            status_server_opts = ("--net=host",)185        output_opts = ()186        if test_output:187            podman_output = runtime_task.task.runnable.output_dir188            output_opts = (189                "-v",190                (f"{test_output}:" f"{os.path.expanduser(podman_output)}"),191            )192        image, _ = self._get_image_from_cache(runtime_task)193        if not image:194            image = self.config.get("spawner.podman.image")195        envs = [f"-e={k}={v}" for k, v in env_args.items()]196        try:197            # pylint: disable=W0201198            _, stdout, _ = await self.podman.execute(199                "create",200                *status_server_opts,201                *output_opts,202                *test_opts,203                entry_point_arg,204                *envs,205                image,206            )207        except PodmanException as ex:208            msg = f"Could not create podman container: {ex}"209            runtime_task.status = msg210            return False211        return stdout.decode().strip()212    async def spawn_task(self, runtime_task):213        self.create_task_output_dir(runtime_task)214        podman_bin = self.config.get("spawner.podman.bin")215        try:216            # pylint: disable=W0201217            self.podman = Podman(podman_bin)218        except PodmanException as ex:219            runtime_task.status = str(ex)220            return False221        major, minor, _ = await self.python_version222        # Return only the "to" location223        eggs = self.get_eggs_paths(major, minor)224        destination_eggs = ":".join(map(lambda egg: str(egg[1]), eggs))225        env_args = {"PYTHONPATH": destination_eggs}226        output_dir_path = self.task_output_dir(runtime_task)227        container_id = await self._create_container_for_task(228            runtime_task, env_args, output_dir_path229        )230        runtime_task.spawner_handle = container_id231        await self.deploy_avocado(container_id)232        try:233            # pylint: disable=W0201234            returncode, _, _ = await self.podman.start(container_id)235        except PodmanException as ex:236            msg = f"Could not start container: {ex}"237            runtime_task.status = msg238            LOG.error(msg)239            return False240        return returncode == 0241    def create_task_output_dir(self, runtime_task):...

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.