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

How to use get_partition_by_mountpoint method in lisa

Best Python code snippet using lisa_python

writeexts.py

Source:writeexts.py

...399                                            rootfs_uuid=rootfs_uuid)400        else:401            if device:402                # A partitioned disk or image - boot with partition UUID403                root_part = device.get_partition_by_mountpoint('/')404                root_guid = device.get_partition_uuid(root_part)405                self.generate_bootloader_config(mountpoint,406                                                root_guid=root_guid)407            else:408                # Unpartitioned and no initramfs - cannot boot with a UUID409                self.generate_bootloader_config(mountpoint)410        self.install_bootloader(mountpoint)411    def create_partition_mountpoints(self, device, system_dir):412        '''Create (or empty) partition mountpoints in the root filesystem413        Delete contents of partition mountpoints in the rootfs to leave an414        empty mount drectory (files are copied to the actual partition in415        create_partitioned_system()), or create an empty mount directory in416        the rootfs if the mount path doesn't exist.417        Args:418            device: A pyfdisk.py Device object describing the partitioning419            system_dir: A path to the Baserock rootfs to be modified420        '''421        for part in device.partitionlist:422            if hasattr(part, 'mountpoint') and part.mountpoint != '/':423                part_mount_dir = os.path.join(system_dir,424                                     re.sub('^/', '', part.mountpoint))425                if os.path.exists(part_mount_dir):426                    self.status(msg='Deleting files in mountpoint '427                                    'for %s partition' % part.mountpoint)428                    self.empty_dir(part_mount_dir)429                else:430                    self.status(msg='Creating empty mount directory '431                                    'for %s partition' % part.mountpoint)432                    os.mkdir(part_mount_dir)433    def create_orig(self, version_root, temp_root):434        '''Create the default "factory" system.'''435        orig = os.path.join(version_root, 'orig')436        self.status(msg='Creating orig subvolume')437        subprocess.check_call(['btrfs', 'subvolume', 'create', orig])438        self.status(msg='Copying files to orig subvolume')439        subprocess.check_call(['cp', '-a', temp_root + '/.', orig + '/.'])440        return orig441    def create_run(self, version_root):442        '''Create the 'run' snapshot.'''443        self.status(msg='Creating run subvolume')444        orig = os.path.join(version_root, 'orig')445        run = os.path.join(version_root, 'run')446        subprocess.check_call(447            ['btrfs', 'subvolume', 'snapshot', orig, run])448    def create_state_subvolume(self, system_dir, mountpoint, state_subdir):449        '''Create a shared state subvolume.450        We need to move any files added to the temporary rootfs by the451        configure extensions to their correct home. For example, they might452        have added keys in `/root/.ssh` which we now need to transfer to453        `/state/root/.ssh`.454        '''455        self.status(msg='Creating %s subvolume' % state_subdir)456        subvolume = os.path.join(mountpoint, 'state', state_subdir)457        subprocess.check_call(['btrfs', 'subvolume', 'create', subvolume])458        os.chmod(subvolume, 0o755)459        existing_state_dir = os.path.join(system_dir, state_subdir)460        self.move_dir_contents(existing_state_dir, subvolume)461    def move_dir_contents(self, source_dir, target_dir):462        '''Move all files source_dir, to target_dir'''463        n = self.__cmd_files_in_dir(['mv'], source_dir, target_dir)464        if n:465            self.status(msg='Moved %d files to %s' % (n, target_dir))466    def copy_dir_contents(self, source_dir, target_dir):467        '''Copy all files source_dir, to target_dir'''468        n = self.__cmd_files_in_dir(['cp', '-a', '-r'], source_dir, target_dir)469        if n:470            self.status(msg='Copied %d files to %s' % (n, target_dir))471    def empty_dir(self, directory):472        '''Empty the contents of a directory, but not the directory itself'''473        n = self.__cmd_files_in_dir(['rm', '-rf'], directory)474        if n:475            self.status(msg='Deleted %d files in %s' % (n, directory))476    def __cmd_files_in_dir(self, cmd, source_dir, target_dir=None):477        files = []478        if os.path.exists(source_dir):479            files = os.listdir(source_dir)480        for filename in files:481            filepath = os.path.join(source_dir, filename)482            add_params = [filepath, target_dir] if target_dir else [filepath]483            subprocess.check_call(cmd + add_params)484        return len(files)485    def complete_fstab_for_btrfs_layout(self, system_dir,486                                        rootfs_uuid=None, device=None):487        '''Fill in /etc/fstab entries for the default Btrfs disk layout.488        In the future we should move this code out of the write extension and489        in to a configure extension. To do that, though, we need some way of490        informing the configure extension what layout should be used. Right now491        a configure extension doesn't know if the system is going to end up as492        a Btrfs disk image, a tarfile or something else and so it can't come493        up with a sensible default fstab.494        Configuration extensions can already create any /etc/fstab that they495        like. This function only fills in entries that are missing, so if for496        example the user configured /home to be on a separate partition, that497        decision will be honoured and /state/home will not be created.498        '''499        shared_state_dirs = {'home', 'root', 'opt', 'srv', 'var'}500        fstab = Fstab(os.path.join(system_dir, 'etc', 'fstab'))501        existing_mounts = fstab.get_mounts()502        if '/' in existing_mounts:503            root_device = existing_mounts['/']504        else:505            root_device = (self.get_root_device() if rootfs_uuid is None else506                           'UUID=%s' % rootfs_uuid)507            fstab.add_line('%s  / btrfs defaults,rw,noatime 0 1' % root_device)508        # Add fstab entries for partitions509        part_mountpoints = set()510        if device:511            mount_parts = set(p for p in device.partitionlist512                          if hasattr(p, 'mountpoint') and p.mountpoint != '/')513            for part in mount_parts:514                if part.mountpoint not in existing_mounts:515                    # Get filesystem UUID516                    part_uuid = self.get_uuid(device.location,517                                              part.extent.start *518                                              device.sector_size)519                    self.status(msg='Adding fstab entry for %s '520                                    'partition' % part.mountpoint)521                    fstab.add_line('UUID=%s  %s %s defaults,rw,noatime '522                                   '0 2' % (part_uuid, part.mountpoint,523                                            part.filesystem))524                    part_mountpoints.add(part.mountpoint)525                else:526                    self.status(msg='WARNING: an entry already exists in '527                                    'fstab for %s partition, skipping' %528                                    part.mountpoint)529        # Add entries for state dirs530        all_mountpoints = set(existing_mounts.keys()) | part_mountpoints531        state_dirs_to_create = set()532        for state_dir in shared_state_dirs:533            mp = '/' + state_dir534            if mp not in all_mountpoints:535                state_dirs_to_create.add(state_dir)536                state_subvol = os.path.join('/state', state_dir)537                fstab.add_line(538                        '%s  /%s  btrfs subvol=%s,defaults,rw,noatime 0 2' %539                        (root_device, state_dir, state_subvol))540        fstab.write()541        return state_dirs_to_create542    def find_initramfs(self, temp_root):543        '''Check whether the rootfs has an initramfs.544        Uses the INITRAMFS_PATH option to locate it.545        '''546        if 'INITRAMFS_PATH' in os.environ:547            initramfs = os.path.join(temp_root, os.environ['INITRAMFS_PATH'])548            if not os.path.exists(initramfs):549                raise ExtensionError('INITRAMFS_PATH specified, '550                                     'but file does not exist')551            return initramfs552        return None553    def install_initramfs(self, initramfs_path, version_root):554        '''Install the initramfs outside of 'orig' or 'run' subvolumes.555        This is required because syslinux doesn't traverse subvolumes when556        loading the kernel or initramfs.557        '''558        self.status(msg='Installing initramfs')559        initramfs_dest = os.path.join(version_root, 'initramfs')560        subprocess.check_call(['cp', '-a', initramfs_path, initramfs_dest])561    def install_kernel(self, version_root, temp_root):562        '''Install the kernel outside of 'orig' or 'run' subvolumes'''563        self.status(msg='Installing kernel')564        image_names = ['vmlinuz', 'zImage', 'uImage']565        kernel_dest = os.path.join(version_root, 'kernel')566        for name in image_names:567            try_path = os.path.join(temp_root, 'boot', name)568            if os.path.exists(try_path):569                subprocess.check_call(['cp', '-a', try_path, kernel_dest])570                break571    def install_dtb(self, version_root, temp_root):572        '''Install the device tree outside of 'orig' or 'run' subvolumes'''573        self.status(msg='Installing devicetree')574        device_tree_path = self.get_dtb_path()575        dtb_dest = os.path.join(version_root, 'dtb')576        try_path = os.path.join(temp_root, device_tree_path)577        if os.path.exists(try_path):578            subprocess.check_call(['cp', '-a', try_path, dtb_dest])579        else:580            logging.error("Failed to find device tree %s", device_tree_path)581            raise ExtensionError(582                'Failed to find device tree %s' % device_tree_path)583    def get_dtb_path(self):584        return os.environ.get('DTB_PATH', '')585    def get_bootloader_install(self):586        # Do we actually want to install the bootloader?587        # Set this to "none" to prevent the install588        return os.environ.get('BOOTLOADER_INSTALL', 'extlinux')589    def get_bootloader_config_format(self):590        # The config format for the bootloader,591        # if not set we default to extlinux for x86592        return os.environ.get('BOOTLOADER_CONFIG_FORMAT', 'extlinux')593    def get_extra_kernel_args(self):594        return os.environ.get('KERNEL_ARGS', '')595    def get_root_device(self):596        return os.environ.get('ROOT_DEVICE', '/dev/sda')597    def generate_bootloader_config(self, *args, **kwargs):598        '''Install extlinux on the newly created disk image.'''599        config_function_dict = {600            'extlinux': self.generate_extlinux_config,601        }602        config_type = self.get_bootloader_config_format()603        if config_type in config_function_dict:604            config_function_dict[config_type](*args, **kwargs)605        else:606            raise ExtensionError(607                'Invalid BOOTLOADER_CONFIG_FORMAT %s' % config_type)608    def generate_extlinux_config(self, real_root,609                                 rootfs_uuid=None, root_guid=None):610        '''Generate the extlinux configuration file611        Args:612            real_root: Path to the mounted top level of the root filesystem613            rootfs_uuid: Specify a filesystem UUID which can be loaded using614                         an initramfs aware of filesystems615            root_guid: Specify a partition GUID, can be used without an616                       initramfs617        '''618        self.status(msg='Creating extlinux.conf')619        # To be compatible with u-boot, create the extlinux.conf file in620        # /extlinux/ rather than /621        # Syslinux, however, requires this to be in /, so create a symlink622        # as well623        config_path = os.path.join(real_root, 'extlinux')624        os.makedirs(config_path)625        config = os.path.join(config_path, 'extlinux.conf')626        os.symlink('extlinux/extlinux.conf', os.path.join(real_root,627                                                          'extlinux.conf'))628        ''' Please also update the documentation in the following files629            if you change these default kernel args:630            - kvm.write.help631            - rawdisk.write.help632            - virtualbox-ssh.write.help '''633        kernel_args = (634            'rw ' # ro ought to work, but we don't test that regularly635            'init=/sbin/init ' # default, but it doesn't hurt to be explicit636            'rootfstype=btrfs ' # required when using initramfs, also boots637                                # faster when specified without initramfs638            'rootflags=subvol=systems/default/run ') # boot runtime subvol639        # See init/do_mounts.c:182 in the kernel source, in the comment above640        # function name_to_dev_t(), for an explanation of the available641        # options for the kernel parameter 'root', particularly when using642        # GUID/UUIDs643        if rootfs_uuid:644            root_device = 'UUID=%s' % rootfs_uuid645        elif root_guid:646            root_device = 'PARTUUID=%s' % root_guid647        else:648            # Fall back to the root partition named in the cluster649            root_device = self.get_root_device()650        kernel_args += 'root=%s ' % root_device651        kernel_args += self.get_extra_kernel_args()652        with open(config, 'w') as f:653            f.write('default linux\n')654            f.write('timeout 1\n')655            f.write('label linux\n')656            f.write('kernel /systems/default/kernel\n')657            if rootfs_uuid is not None:658                f.write('initrd /systems/default/initramfs\n')659            if self.get_dtb_path() != '':660                f.write('devicetree /systems/default/dtb\n')661            f.write('append %s\n' % kernel_args)662    def install_bootloader(self, *args, **kwargs):663        install_function_dict = {664            'extlinux': self.install_bootloader_extlinux,665        }666        install_type = self.get_bootloader_install()667        if install_type in install_function_dict:668            install_function_dict[install_type](*args, **kwargs)669        elif install_type != 'none':670            raise ExtensionError(671                'Invalid BOOTLOADER_INSTALL %s' % install_type)672    def install_bootloader_extlinux(self, real_root):673        self.status(msg='Installing extlinux')674        subprocess.check_call(['extlinux', '--install', real_root])675        # FIXME this hack seems to be necessary to let extlinux finish676        subprocess.check_call(['sync'])677        time.sleep(2)678    def install_syslinux_blob(self, device, orig_root):679        '''Install Syslinux MBR blob680        This is the first stage of boot (for partitioned images) on x86681        machines. It is not required where there is no partition table. The682        syslinux bootloader is written to the MBR, and is capable of loading683        extlinux. This only works when the partition is set as bootable (MBR),684        or the legacy boot flag is set (GPT). The blob is built with extlinux,685        and found in the rootfs'''686        pt_format = device.partition_table_format.lower()687        if pt_format in ('gpb', 'mbr'):688            blob = 'mbr.bin'689        elif pt_format == 'gpt':690            blob = 'gptmbr.bin'691        blob_name = 'usr/share/syslinux/' + blob692        self.status(msg='Installing syslinux %s blob' % pt_format.upper())693        blob_location = os.path.join(orig_root, blob_name)694        if os.path.exists(blob_location):695            subprocess.check_call(['dd', 'if=%s' % blob_location,696                                         'of=%s' % device.location,697                                         'bs=440', 'count=1', 'conv=notrunc'])698        else:699            raise ExtensionError('MBR blob not found. Is this the correct'700                   'architecture? The MBR blob will only be built for x86'701                   'systems. You may wish to configure BOOTLOADER_INSTALL')702    def install_syslinux_menu(self, real_root, temp_root):703        '''Make syslinux/extlinux menu binary available.704        The syslinux boot menu is compiled to a file named menu.c32. Extlinux705        searches a few places for this file but it does not know to look inside706        our subvolume, so we copy it to the filesystem root.707        If the file is not available, the bootloader will still work but will708        not be able to show a menu.709        '''710        menu_file = os.path.join(temp_root, 'usr', 'share', 'syslinux',711                                 'menu.c32')712        if os.path.isfile(menu_file):713            self.status(msg='Copying menu.c32')714            shutil.copy(menu_file, real_root)715    def parse_attach_disks(self):716        '''Parse $ATTACH_DISKS into list of disks to attach.'''717        if 'ATTACH_DISKS' in os.environ:718            s = os.environ['ATTACH_DISKS']719            return s.split(':')720        else:721            return []722    def bootloader_config_is_wanted(self):723        '''Does the user want to generate a bootloader config?724        The user may set $BOOTLOADER_CONFIG_FORMAT to the desired725        format. 'extlinux' is the only allowed value, and is the default726        value for x86-32 and x86-64.727        '''728        def is_x86(arch):729            return (arch == 'x86_64' or730                    (arch.startswith('i') and arch.endswith('86')))731        value = os.environ.get('BOOTLOADER_CONFIG_FORMAT', '')732        if value == '':733            if not is_x86(os.uname()[-1]):734                return False735        return True736    def get_environment_boolean(self, variable, default='no'):737        '''Parse a yes/no boolean passed through the environment.'''738        value = os.environ.get(variable, default).lower()739        if value in ('no', '0', 'false'):740            return False741        elif value in ('yes', '1', 'true'):742            return True743        else:744            raise ExtensionError('Unexpected value for %s: %s' %745                                 (variable, value))746    def check_ssh_connectivity(self, ssh_host):747        try:748            output = ssh_runcmd(ssh_host, ['echo', 'test'])749        except ExtensionError as e:750            logging.error("Error checking SSH connectivity: %s", str(e))751            raise ExtensionError(752                'Unable to SSH to %s: %s' % (ssh_host, e))753        if output.strip() != 'test':754            raise ExtensionError(755                'Unexpected output from remote machine: %s' % output.strip())756    def is_device(self, location):757        try:758            st = os.stat(location)759            return stat.S_ISBLK(st.st_mode)760        except OSError as e:761            if e.errno == errno.ENOENT:762                return False763            raise764    def create_partitioned_system(self, temp_root, location):765        '''Deploy a bootable Baserock system with a custom partition layout.766        Called if USE_PARTITIONING=yes is set in the deployment options.767        '''768        part_spec = os.environ.get('PARTITION_FILE', 'partitioning/default')769        disk_size = self.get_disk_size()770        if not disk_size:771            raise writeexts.ExtensionError('DISK_SIZE is not defined')772        dev = partitioning.do_partitioning(location, disk_size,773                                           temp_root, part_spec)774        boot_partition_available = dev.get_partition_by_mountpoint('/boot')775        for part in dev.partitionlist:776            if not hasattr(part, 'mountpoint'):777                continue778            if part.mountpoint == '/':779                # Re-format the rootfs, to include needed extra features780                with pyfdisk.create_loopback(location,781                                             part.extent.start *782                                             dev.sector_size, part.size) as l:783                    self.mkfs_btrfs(l)784            self.status(msg='Mounting partition %d' % part.number)785            offset = part.extent.start * dev.sector_size786            with self.mount_partition(location,787                                      offset, part.size) as part_mount_dir:788                if part.mountpoint == '/':...

lissuite.py

Source:lissuite.py

...125        # 10MB for log file creation126        root_partition_buffer_space = 10485760127        # allowed limit of +- 1MB128        boot_partition_buffer_space = 1048576129        root_partition = df.get_partition_by_mountpoint("/")130        boot_partition = df.get_partition_by_mountpoint("/boot")131        assert root_partition, "fail to get root partition"132        assert boot_partition, "fail to get boot partition"133        ramdisk_size_factor = 1134        if root_partition.name != boot_partition.name:135            ramdisk_size_factor = 2136        lis_path = lisdriver.download()137        os_version = node.os.information.release.split(".")138        version = os_version[0] + os_version[1]139        lib_module_required_space = rpm.get_file_size(140            f"{lis_path}/RPMS{version}/kmod-microsoft-hyper*x86_64.rpm"141        )142        uname = node.tools[Uname]143        kernel_version = uname.get_linux_information().kernel_version_raw144        ramdisk_required_space = stat.get_total_size(...

df.py

Source:df.py

...36                    percentage_blocks_used=int(df_entry["percentage_use"]),37                )38            )39        return partition_info40    def get_partition_by_mountpoint(41        self, mountpoint: str, force_run: bool = False42    ) -> Optional[PartitionInfo]:43        # get `df` entry for the partition with the given mountpoint44        df_partition_info = self.get_partitions(force_run)45        for partition in df_partition_info:46            if partition.mountpoint == mountpoint:47                return partition...

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.