How to use deploy_avocado method in avocado

Best Python code snippet using avocado_python

plugin_interfaces.py

Source:plugin_interfaces.py

...300    Spawners that uses any type of isolation model would be a possible children301    of this.302    """303    @abc.abstractmethod304    async def deploy_avocado(self, where):305        """Avocado deployment for the isolated environment.306        This method should be executed before spawning the task.307        :param where: handle for where the spawner should deploy avocado.308                      You can use this parameter to pass any valid argument,309                      like a container, vm, or any other isolated environment310                      object or identifier.311        """312    @abc.abstractmethod313    async def deploy_artifacts(self):314        """Basic artifacts deployment for the isolated environment.315        During this stage test references (i.e: mytest.py), data files (i.e:316        mytest.py.data/), and any other basic requirement should be deployed to317        the isolated environment. Please keep in mind that test requirements318        defined at the requirements resolver level are not part of this...

podman.py

Source:podman.py

...140            self._PYTHON_VERSIONS_CACHE[image] = result141        return self._PYTHON_VERSIONS_CACHE[image]142    async def deploy_artifacts(self):143        pass144    async def deploy_avocado(self, where):145        # Deploy all the eggs to container inside /tmp/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):242        output_dir_path = self.task_output_dir(runtime_task)243        output_podman_path = "~/avocado/job-results/spawner/task"244        os.makedirs(output_dir_path, exist_ok=True)245        runtime_task.task.setup_output_dir(output_podman_path)...

shared_customize_vm.py

Source:shared_customize_vm.py

...27destination_avocado_path = "/tmp/utils/avocado"28###############################################################################29# HELPERS30###############################################################################31def deploy_avocado(vm):32    """33    Deploy the avocado package to a vm.34    :param vm: vm to deploy to (must be compatible)35    :type vm: :py:class:`virttest.qemu_vm.VM`36    """37    # TODO: scp does not seem to raise exception if path does not exist38    log.info(f"Deploying avocado utilities at {vm.params['main_vm']}")39    log.debug(f"Deploying utilities from {source_avocado_path} on host to "40              f"{destination_avocado_path} on the virtual machine.")41    vm.session.cmd("mkdir -p " + destination_avocado_path)42    vm.session.cmd("touch " + os.path.join(destination_avocado_path, "__init__.py"))43    vm.copy_files_to(source_avocado_path, destination_avocado_path, timeout=180)44def deploy_data(vm, folder_name, dst_folder_name="", custom_src_path="", timeout=60):45    """46    Deploy data to a vm.47    :param vm: vm to deploy to (must be compatible)48    :type vm: :py:class:`virttest.qemu_vm.VM`49    :param str folder_name: data folder name (default path is 'guest')50    :param params: deploy configuration51    :type params: {str, str}52    :param str custom_src_path: custom path to the src data folder53    :param str custom_dst_name: custom folder name of the dst data folder54    :param int timeout: copying timeout55    """56    if custom_src_path == "":57        src_path = os.path.join(vm.params["suite_path"], folder_name)58    else:59        src_path = os.path.join(custom_src_path, folder_name)60    tmp_dir = vm.params.get("tmp_dir", "/tmp")61    folder_name = dst_folder_name if dst_folder_name else folder_name62    dst_path = os.path.join(tmp_dir, folder_name)63    cmd = f"rm -fr" if vm.params.get("os_type", "linux") == "linux" else "rmdir /s /q"64    cmd += f" {dst_path}"65    vm.session.cmd(cmd)66    vm.copy_files_to(src_path, dst_path, timeout=timeout)67def handle_ssh_authorized_keys(vm):68    """69    Deploy an SSH key to a vm.70    :param vm: vm to deploy to (must be compatible)71    :type vm: :py:class:`virttest.qemu_vm.VM`72    """73    ssh_authorized_keys = os.environ['SSHKEY'] if 'SSHKEY' in os.environ else ""74    if ssh_authorized_keys == "":75        return76    log.info(f"Enabled ssh key '{ssh_authorized_keys}'")77    tmpfile = tempfile.NamedTemporaryFile(delete=False)78    tmpfile.write((ssh_authorized_keys + '\n').encode())79    tmpfile.close()80    vm.session.cmd('mkdir -p /root/.ssh')81    vm.copy_files_to(tmpfile.name, '/root/.ssh/authorized_keys')82    os.unlink(tmpfile.name)83###############################################################################84# TEST MAIN85###############################################################################86@error_context.context_aware87def run(test, params, env):88    """89    Main test run.90    :param test: test object91    :type test: :py:class:`avocado_vt.test.VirtTest`92    :param params: extended dictionary of parameters93    :type params: :py:class:`virttest.utils_params.Params`94    :param env: environment object95    :type env: :py:class:`virttest.utils_env.Env`96    """97    vmnet = env.get_vmnet()98    vm, session, params = vmnet.get_single_vm_with_session_and_params()99    os_type = params.get("os_type", "linux")100    os_variant = params.get("os_variant", "ibs")101    tmp_dir = params.get("tmp_dir", "/tmp")102    # main deployment part103    log.info(f"Deploying customized data to {tmp_dir} on {params['main_vm']}")104    deploy_data(vm, "data/")105    # WARNING: control file must add path to utils to the pythonpath106    log.info(f"Deploying customized test utilities to {tmp_dir} on {params['main_vm']}")107    deploy_data(vm, "utils/")108    # as avocado utils are deployed in a subdirectory, deploy them after the general utils109    if params.get_boolean("guest_avocado_enabled"):110        if os.path.exists(source_avocado_path):111            deploy_avocado(vm)112        else:113            log.warning("No source avocado path found and could be deployed")114    # additional deployment part115    additional_deployment_path = params.get("additional_deployment_dir", "/mnt/local/packages")116    destination_packages_path = params.get("deployed_packages_path", "/tmp/packages")117    if additional_deployment_path is not None and os.path.isdir(additional_deployment_path):118        log.info(f"Deploying additional packages and data to {tmp_dir} on {params['main_vm']}")119        # careful about the splitting process - since we perform deleting need to validate here120        additional_deployment_path = additional_deployment_path.rstrip("/")121        deploy_data(vm, os.path.basename(additional_deployment_path), "packages",122                    os.path.dirname(additional_deployment_path), 60)123    else:124        raise exceptions.TestError("Additional deployment path %s does not exist (current dir: "125                                   "%s)" % (additional_deployment_path, os.getcwd()))...

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.