Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use lv_mount method in avocado

Best Python code snippet using avocado_python

lvm.py

Source:lvm.py

...81        params["lv_snapshot_name"] = params["get_state"]82        if mount_loc:83            # mount to avoid not-mounted errors84            try:85                lv_utils.lv_mount(params["vg_name"],86                                  params["lv_pointer_name"],87                                  mount_loc)88            except lv_utils.LVException:89                pass90            lv_utils.lv_umount(params["vg_name"],91                               params["lv_pointer_name"])92        try:93            logging.info("Restoring %s to state %s", vm_name, params["get_state"])94            lv_utils.lv_remove(params["vg_name"], params["lv_pointer_name"])95            lv_utils.lv_take_snapshot(params["vg_name"],96                                      params["lv_snapshot_name"],97                                      params["lv_pointer_name"])98        finally:99            if mount_loc:100                lv_utils.lv_mount(params["vg_name"],101                                  params["lv_pointer_name"],102                                  mount_loc)103    @classmethod104    def set(cls, params, object=None):105        """106        Store a state saving the current changes.107        All arguments match the base class.108        """109        vm_name = params["vms"]110        params["lv_snapshot_name"] = params["set_state"]111        logging.info("Taking a snapshot '%s' of %s", params["set_state"], vm_name)112        lv_utils.lv_take_snapshot(params["vg_name"],113                                  params["lv_pointer_name"],114                                  params["lv_snapshot_name"])115    @classmethod116    def unset(cls, params, object=None):117        """118        Remove a state with previous changes.119        All arguments match the base class and in addition:120        :raises: :py:class:`ValueError` if LV pointer state was used121        """122        vm_name = params["vms"]123        lv_pointer = params["lv_pointer_name"]124        if params["unset_state"] == lv_pointer:125            raise ValueError("Cannot unset built-in state '%s'" % lv_pointer)126        params["lv_snapshot_name"] = params["unset_state"]127        logging.info("Removing snapshot %s of %s", params["lv_snapshot_name"], vm_name)128        lv_utils.lv_remove(params["vg_name"], params["lv_snapshot_name"])129    @classmethod130    def check_root(cls, params, object=None):131        """132        Check whether a root state or essentially the object exists.133        All arguments match the base class.134        """135        vm_name = params["vms"]136        image_name = params["image_name"]137        logging.debug("Checking whether %s exists (root state requested)", vm_name)138        if lv_utils.lv_check(params["vg_name"], params["lv_name"]):139            logging.info("The required virtual machine %s's %s (%s) exists",140                         vm_name, image_name, params["lv_name"])141            return True142        else:143            logging.info("The required virtual machine %s's %s (%s) doesn't exist",144                         vm_name, image_name, params["lv_name"])145            return False146    @classmethod147    def set_root(cls, params, object=None):148        """149        Set a root state to provide object existence.150        All arguments match the base class.151        Create a disk, virtual group, thin pool and logical volume152        for each object.153        """154        vm_name = params["vms"]155        mount_loc = cls._get_image_mount_loc(params)156        logging.info("Creating original logical volume for %s", vm_name)157        vg_setup(params["vg_name"],158                 params["disk_vg_size"],159                 params["disk_basedir"],160                 params["disk_sparse_filename"],161                 params["use_tmpfs"] == "yes")162        lv_utils.lv_create(params["vg_name"],163                           params["lv_name"],164                           params["lv_size"],165                           # NOTE: call by key to keep good argument order which wasn't166                           # accepted upstream for backward API compatibility167                           pool_name=params["lv_pool_name"],168                           pool_size=params["lv_pool_size"])169        lv_utils.lv_take_snapshot(params["vg_name"],170                                  params["lv_name"],171                                  params["lv_pointer_name"])172        if mount_loc:173            if not os.path.exists(mount_loc):174                os.mkdir(mount_loc)175            lv_utils.lv_mount(params["vg_name"], params["lv_pointer_name"],176                              mount_loc, create_filesystem="ext4")177            # TODO: it is not correct for the LVM backend to expect QCOW2 images178            # but at the moment we have no better way to provide on states with179            # base image to take snapshots of180            super(LVMBackend, LVMBackend).set_root(params, object)181    @classmethod182    def unset_root(cls, params, object=None):183        """184        Unset a root state to prevent object existence.185        All arguments match the base class and in addition:186        :raises: :py:class:`exceptions.TestWarn` if permanent vm was detected187        Remove the disk, virtual group, thin pool and logical volume188        of each object.189        """190        vm_name = params["vms"]191        mount_loc = cls._get_image_mount_loc(params)192        logging.info("Removing original logical volume for %s", vm_name)193        try:194            if mount_loc:195                if lv_utils.vg_check(params["vg_name"]):196                    # mount to avoid not-mounted errors197                    try:198                        lv_utils.lv_mount(params["vg_name"],199                                          params["lv_pointer_name"],200                                          mount_loc)201                    except lv_utils.LVException:202                        pass203                    lv_utils.lv_umount(params["vg_name"],204                                       params["lv_pointer_name"])205                if os.path.exists(mount_loc):206                    try:207                        os.rmdir(mount_loc)208                    except OSError as ex:209                        logging.warning("No permanent vm can be removed automatically. If "210                                        "this is not a permanent test object, see the debug.")211                        raise exceptions.TestWarn("Permanent vm %s was detected but cannot be "212                                                  "removed automatically" % vm_name)...

lvsetup.py

Source:lvsetup.py

...133    def mount_unmount_lv(self):134        """135        Mounts and unmounts the filesystem on the logical volume.136        """137        lv_utils.lv_mount(self.vg_name, self.lv_name, self.mount_loc)138        lv_utils.lv_umount(self.vg_name, self.lv_name)139    @avocado.fail_on(lv_utils.LVException)140    def test(self):141        """142        A volume group with given name is created in the ramdisk. It then143        creates a logical volume, mounts and unmounts it.144        """145        self.create_lv()146        lv_utils.lv_mount(self.vg_name, self.lv_name, self.mount_loc,147                          create_filesystem=self.fs_name)148        lv_utils.lv_umount(self.vg_name, self.lv_name)149        self.mount_unmount_lv()150    @avocado.fail_on(lv_utils.LVException)151    def test_vg_reactivate(self):152        """153        Deactivate, export, import and activate a volume group.154        """155        self.create_lv()156        lv_utils.lv_mount(self.vg_name, self.lv_name, self.mount_loc,157                          create_filesystem=self.fs_name)158        lv_utils.lv_umount(self.vg_name, self.lv_name)159        lv_utils.vg_reactivate(self.vg_name, export=True)160        self.mount_unmount_lv()161    @avocado.fail_on(lv_utils.LVException)162    def test_lv_snapshot(self):163        """164        Takes a snapshot from the logical and merges snapshot with the165        logical volume.166        """167        self.create_lv()168        lv_utils.lv_mount(self.vg_name, self.lv_name, self.mount_loc,169                          create_filesystem=self.fs_name)170        lv_utils.lv_umount(self.vg_name, self.lv_name)171        lv_utils.lv_take_snapshot(self.vg_name, self.lv_name,172                                  self.lv_snap_name,173                                  self.lv_snapshot_size)174        lv_utils.lv_revert(self.vg_name, self.lv_name, self.lv_snap_name)175        self.mount_unmount_lv()176    def tearDown(self):177        """178        Cleans up loop device.179        """180        self.delete_lv()181        if not self.disk:182            disk.delete_loop_device(self.device)

uecbuild.py

Source:uecbuild.py

1#!/usr/bin/env python2import argparse3import os4import subprocess5import sys6import tempfile7DEFAULT_KERNEL = "kernels/maverick-server-uec-amd64-vmlinuz-virtual"8DEFAULT_IMAGE = "images/maverick-server-uec-amd64.img"9DOMAIN_EXISTS_ERROR = """Domain already exists (and could be running)! Manually shutdown and remove it with:10    virsh shutdown %(name)s (force with: virsh destroy %(name)s)11    virsh undefine %(name)s"""12def execute(args):13    with open(os.devnull, "r+") as f:14        subprocess.call(args, stdout=f, stderr=f)15def write_template(input_path, output_path, replacements):16    with open(input_path, "r") as input_file:17        with open(output_path, "w") as output_file:18            output_file.write(input_file.read() % replacements)19TTY_COLORS = {20	"blue":   34,21	"red":    31,22	"white":  39,23	"yellow": 33,24}25def tty_format(str, color="red", bold=True):26	tty_color = TTY_COLORS.get(color, TTY_COLORS["white"])27	tty_format = "1" if bold else "0"28	return "\033[%s;%sm%s\033[0m" % (tty_format, tty_color, str)29def log(msg):30	print(tty_format(msg, "blue", True))31def err(msg):32	print(tty_format(msg, "red", True))33# FIXME: Add global exception handler which prints exception value using err and stacktrace after that34if __name__ == "__main__":35    parser = argparse.ArgumentParser(36		description='Depoy Ubuntu UEC images to KVM/LVM based infrastructures',37		formatter_class=argparse.ArgumentDefaultsHelpFormatter38	)39    parser.add_argument('name', help='FQDN and identifier of domain')40    parser.add_argument('--vg', default='vms', help='name of LVM volume group')41    parser.add_argument('--size', default='5G', help='size of root partition, e.g., 1000M, 5G')42    parser.add_argument('--memory', default=1048576, help='size of RAM in bytes, e.g., 1048576, 2097152')43    parser.add_argument('--vcpus', default=2, help='number of virtual CPUs')44    parser.add_argument('--image', default=DEFAULT_IMAGE, help='reference root image file')45    parser.add_argument('--kernel', default=DEFAULT_KERNEL, help='kernel image to use')46    args = parser.parse_args()47    lv_path = os.path.join("/dev", args.vg, args.name)48    lv_mount = tempfile.mkdtemp("-uecbuild-%s-root" % args.name)49    base_dir = os.path.dirname(os.path.abspath(__file__))50    replacements = {51        "name": args.name,52        "memory": args.memory,53        "vcpus": args.vcpus,54        "kernel": os.path.join(base_dir, args.kernel),55        "lv_path": lv_path,56    }57    # check preconditions58    if os.getuid() != 0:59        raise Exception("Root privileges required")60    if os.path.exists(lv_path):61        raise Exception("LV already exists! Manually remove it with: umount %s; lvremove %s" % (lv_path, lv_path))62    if execute(["virsh", "domstate", args.name]) == 0:63        raise Exception(DOMAIN_EXISTS_ERROR % {"name": args.name})64    # create lv65    log("Creating LV at %s" % lv_path)66    execute(["lvcreate", "-n", args.name, "-L", args.size, args.vg])67    # copy root to lv68    log("Copying root image to LV")69    execute(["dd", "if=%s" % args.image, "of=%s" % lv_path, "bs=1M"])70    # check & resize lv71    log("Checking and resizing LV")72    execute(["e2fsck", "-f", lv_path])73    execute(["resize2fs", lv_path])74    # mount lv75    log("Mounting LV")76    execute(["mount", lv_path, lv_mount])77    # write hostname78    log("Writing hostname to LV's /etc/hostname")79    with open(os.path.join(lv_mount, "etc", "hostname"), "w") as f:80        f.write(args.name + "\n")81    82    # write user-data & meta-data83    log("Writing user-data & meta-data to LV")84    nocloud_dir = os.path.join(lv_mount, "var", "lib", "cloud", "data", "cache", "nocloud")85    os.makedirs(nocloud_dir)86    for template in ["user-data", "meta-data"]:87        write_template(88            input_path=os.path.join(base_dir, "templates", template),89            output_path=os.path.join(nocloud_dir, template),90            replacements=replacements91        )92    # unmount LV & remove tempdir93    log("Unmounting LV")94    execute(["umount", lv_mount])95    execute(["rmdir", lv_mount])96    97    # create domain xml98    log("Adding domain to libvirt")99    domain_xml = tempfile.mkstemp("-uecbuild-%s.xml" % args.name)100    write_template(101        input_path=os.path.join(base_dir, "templates", "domain.xml"),102        output_path=domain_xml[1],103        replacements=replacements104    )105    # define domain xml106    execute(["virsh", "define", domain_xml[1]])107    # remove domain xml108    os.remove(domain_xml[1])109    # start domain110    log("Starting domain %s" % args.name)...

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.