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

How to use check_mount_point method in autotest

Best Python code snippet using autotest_python

ephemeral_disk.py

Source:ephemeral_disk.py

...196        if partition in os.popen("fdisk -l {0} 2> /dev/null".format(partition)).read():197            return True198        else:199            return False200    def check_mount_point(self, partition):201        """ Check if partition is mounted, example:202        check_mount_point('/dev/xvdj1')203            Return True or False depending on the result.204            However it force is enabled, it tries to unmount205        """206        try:207            # Check if partition exists before enabling SWAP208            if not self.check_partition(partition):209                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))210                exit(2)211            # Confirm that mount/umount binary exists212            if not self.check_command('/bin/mount') and not self.check_command('/bin/umount'):213                logging.error('mount/umount command does not exist or cannot be accessible')214                exit(2)215            # Ensure that partition in question is already mounted or not216            # Returns True or False depending on the result217            # It is also important to check if is mounted as SWAP which uses a different place218            mount_points = os.popen("mount 2> /dev/null".format(partition)).read()219            with open('/proc/swaps', 'r') as filename:220                swap_points = filename.read()221                if partition in mount_points or partition in swap_points:222                    return True223                elif partition[:-1] in mount_points or partition[:-1] in swap_points:224                    return True225                else:226                    return False227        except Exception, e:228            logging.critical("Error while checking mount point")229            logging.error("{0}".format(e))230    def check_command(self, command):231        """ Check if command exists and can be executed.access232            Returns True or False depending on the result233        """234        if os.path.isfile(command) and os.access(command, os.X_OK):235            return True236        else:237            return False238    def delete_disk_partition(self, disk, part_number):239        """ Delete partitions using fdisk, example:240        delete('/dev/xvdj', '1')241            Then, deletes partition 1 of the242            disk /dev/xvdj. """243        # fdisk command to delete partition244        command = """echo "245d246{1}247w248" | /sbin/fdisk {0} 1>/dev/null """249        partitioning = command.format(disk, part_number)250        partition = "".join((disk, part_number))251        try:252            # Check if partition number is between 1 and 4253            if not part_number in str(partition):254                logging.error('Partition number must be between 1 and 4')255                exit(2)256            # Check if partition selected exist, otherwise raise an error257            if not self.check_partition(partition):258                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))259                exit(2)260            logging.info("Deleting partition... !")261            os.system(partitioning)262            logging.info("Partition deleted successfully")263        except Exception, e:264            logging.critical("Error while deleting disk partition {0} - Please double check if it is in use and try again".format(partition))265            logging.error("{0}".format(e))266    def enable_swap(self, partition):267        """ Enable SWAP partition previously created, example:268        enable('/dev/xvdj1')269            Then, formats partition as SWAP and then enable270            using SWAPON command in Linux"""271        try:272            # Check if partition exists before enabling SWAP273            if not self.check_partition(partition):274                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))275                exit(2)276            # Confirm that mkswap binary exists277            if not self.check_command('/sbin/mkswap'):278                logging.error('mkswap command does not exist or cannot be accessible')279                exit(2)280            # Double check if partition is already in use281            # and if Force is set unmount it before taking any action282            if self.check_mount_point(partition):283                if self.force:284                    self.force_unmount(partition)285                else:286                    logging.error('Partition {0} already mounted, cannot touch it!'.format(partition))287                    exit(2)288            # Creates SWAP using mkswap and enable it using swapon289            logging.info('Formating partition {0} as SWAP'.format(partition))290            os.system("mkswap {0} 1>/dev/null".format(partition))291            os.system('swapon {0} 1>/dev/null'.format(partition))292            logging.info('SWAP enabled successfully')293        except Exception, e:294            logging.critical("Error while enabling SWAP")295            logging.error("{0}".format(e))296    def disable_swap(self, partition):297        """ Disable SWAP partition, example298        disable('/dev/xvdj1') """299        try:300            # Check if partition exists before enabling SWAP301            if not self.check_partition(partition):302                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))303                exit(2)304             # Confirm that swapoff binary exists305            if not self.check_command('/sbin/swapoff'):306                logging.error('swapoff command does not exist or cannot be accessible')307                exit(2)308            # Double check if partition is already in use309            # and if Force is set unmount it before taking any action310            if self.check_mount_point(partition):311                if self.force:312                    self.force_unmount(partition)313                else:314                    logging.error('Partition {0} already mounted, cannot touch it!'.format(partition))315                    exit(2)316            os.system("swapoff {0} 1>/dev/null".format(partition))317            logging.info('SWAP disabled successfully')318        except Exception, e:319            logging.critical("Error while disabling SWAP")320            logging.error("{0}".format(e))321    def format_as_ext3(self, partition):322        """ Format partition previously created as EXT3, example:323        format_as_ext3('/dev/xvdj1')324        """325        try:326            # Check if partition selected exist, otherwise raise an error327            if not self.check_partition(partition):328                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))329                exit(2)330             # Confirm that mkfs.ext3 binary exists331            if not self.check_command('/sbin/mkfs.ext3'):332                logging.error('mkfs.ext3 command does not exist or cannot be accessible')333                exit(2)334            # Double check if partition is already in use335            # and if Force is set unmount it before taking any action336            if self.check_mount_point(partition):337                if self.force:338                    self.force_unmount(partition)339                else:340                    logging.error('Partition {0} already mounted, cannot touch it!'.format(partition))341                    exit(2)342            logging.info('Formating partition {0} as EXT3'.format(partition))343            os.system("mkfs.ext3 -q {0} 1>/dev/null".format(partition))344            logging.info('Partition formatted successfully as EXT3')345        except Exception, e:346            logging.critical("Error while formatting partition as EXT3")347            logging.error("{0}".format(e))348    def format_as_ext4(self, partition):349        """ Format partition previously created as EXT4, example:350        format_as_ext4('/dev/xvdj1')351        """352        try:353            # Check if partition selected exist, otherwise raise an error354            if not self.check_partition(partition):355                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))356                exit(2)357             # Confirm that mkfs.ext4 binary exists358            if not self.check_command('/sbin/mkfs.ext4'):359                logging.error('mkfs.ext4 command does not exist or cannot be accessible')360                exit(2)361            # Double check if partition is already in use362            # and if Force is set unmount it before taking any action363            if self.check_mount_point(partition):364                if self.force:365                    self.force_unmount(partition)366                else:367                    logging.error('Partition {0} already mounted, cannot touch it!'.format(partition))368                    exit(2)369            logging.info('Formating partition {0} as EXT4'.format(partition))370            os.system("mkfs.ext4 -q {0} 1>/dev/null".format(partition))371            logging.info('Partition formmatted successfully as EXT4')372        except Exception, e:373            logging.critical("Error while formatting partition as EXT4")374            logging.error("{0}".format(e))375    def format_as_xfs(self, partition):376        """ Format partition previously created as XFS, example:377        format_as_xfs('/dev/xvdj1')378        """379        try:380            # Check if partition selected exist, otherwise raise an error381            if not self.check_partition(partition):382                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))383                exit(2)384             # Confirm that mkfs.ext4 binary exists385            if not self.check_command('/sbin/mkfs.xfs'):386                logging.error('mkfs.xfs command does not exist or cannot be accessible')387                logging.info('Please make sure you have XFS file system tools installed')388                exit(2)389            # Double check if partition is already in use390            # and if Force is set unmount it before taking any action391            if self.check_mount_point(partition):392                if self.force:393                    self.force_unmount(partition)394                else:395                    logging.error('Partition {0} already mounted, cannot touch it!'.format(partition))396                    exit(2)397            logging.info('Formating partition {0} as XFS'.format(partition))398            os.system("mkfs.xfs -q {0} 1>/dev/null".format(partition))399            logging.info('Partition formmatted successfully as XFS')400        except Exception, e:401            logging.critical("Error while formatting partition as XFS")402            logging.error("{0}".format(e))403    def mount_partition(self, partition, mount_point):404        """ Mount partition previously created and formatted, example:405        mount_partition('/dev/xvdj2', '/mnt')406        """407        try:408            # Check if partition selected exist, otherwise raise an error409            if not self.check_partition(partition):410                logging.error('Partition {0} does not exist, please choose an existent one'.format(partition))411                exit(2)412             # Confirm that mount binary exists413            if not self.check_command('/bin/mount'):414                logging.error('mount command does not exist or cannot be accessible, please ensure you also have permission to mount')415                exit(2)416            # Double check if partition is already in use417            # and if Force is set unmount it before taking any action418            if self.check_mount_point(partition):419                if self.force:420                    self.force_unmount(partition)421                else:422                    logging.error('Partition {0} already mounted, cannot touch it!'.format(partition))423                    exit(2)424            logging.info('Mounting partition {0} in {1}'.format(partition, mount_point))425            if os.system("mount {0} {1} 1>/dev/null".format(partition, mount_point)):426                logging.info('Partition {0} was mounted successfully in {1}'.format(partition, mount_point))427        except Exception, e:428            logging.critical("Error while mounting partition {0} in {1}".format(partition, mount_point))429            logging.error("{0}".format(e))430    def force_unmount(self, partition):431        """ Force unmount partition once mounted432        force_unmount('/dev/xvdb1') or force_unmount('/dev/xvdb)433        """434        if self.force:435            # Check if partition selected exist, otherwise raise an error436            # Hint: module commands or subprocess can be used here to simplify437            if not self.check_partition(partition):438                logging.error('Partition {0} does not exist, please choose an existing one'.format(partition))439                exit(2)440            logging.info("Unmounting file system")441            for attempts in range(1,4):442                # Check if full partition or entire disk is mounted443                # Then umounts if true (swap and or FS)444                # This can be improved confirming if it is SWAP or FS rather than using both commands445                if self.check_mount_point(partition):446                    os.system("umount -l {0} 2>/dev/null".format(partition))447                    os.system("swapoff {0} 2>/dev/null".format(partition))448                elif self.check_mount_point(partition[:-1]):449                    os.system("umount -l {0} 2>/dev/null".format(partition[:-1]))450                    os.system("swapoff {0} 2>/dev/null".format(partition[:-1]))451                else:452                    # Break the loop If unmounted successfully otherwise try again453                    logging.info('Unmounted sucessfully!')...

usbupd.py

Source:usbupd.py

...29        self.retry_count = 030        self.job_id = None31        self.start_usb_detection()32        syslog("usbupd: init: Local USB Update is initialized")33    def check_mount_point(self):34        """35        Function looks for mount point and sends the swupdate config to Softwareupdate service36        """37        try:38            if self.retry_count == RETRY_COUNT:39                self.retry_count = 040                return False41            for partition in disk_partitions(all=False):42                if DEVICE_PART1 in partition.device:43                    update_path = os.path.join(partition.mountpoint, UPDATE_PACKAGE_NAME)44                    if os.path.isfile(update_path):45                        local_update_config["image"] = update_path46                        syslog("usbupd: check_mount_point: Starting config parsing...")47                        ret = self.process_config(local_update_config)48                        if ret:49                            syslog("usbupd: check_mount_point: Starting Software Update...")50                            gobject.source_remove(self.job_id)51                            self.start_swupdate(False)52                            self.retry_count = 053                    self.job_id = None54                    return False55            self.retry_count += 156            return True57        except Exception as e:58            syslog("usbupd: check_mount_point: %s" % e)59            if self.job_id is not None:60                gobject.source_remove(self.job_id)61                self.job_id = None62            self.retry_count = 063    def device_event(self,observer, device):64        """65        USB Device Event signal66        """67        try:68            if device.action == "add":69                syslog("usbupd: device_event: USB inserted")70                if self.job_id is None:71                    self.job_id = gobject.timeout_add(TIMEOUT,self.check_mount_point)72            if device.action == "remove":73                syslog("usbupd: device_event: USB removed")74                # reset device leds75                if self.device_svc:76                    self.device_svc.DeviceUpdateReset()77                if self.retry_count != 0:78                    self.retry_count = 079                if self.job_id is not None:80                    gobject.source_remove(self.job_id)81                    self.job_id = None82        except Exception as e:83            syslog("usbupd: device_event: %s" % e)84            if self.job_id is not None:85                gobject.source_remove(self.job_id)86                self.job_id = None87            self.start_usb_detection()88    def start_usb_detection(self):89        """90        Starts listening to udev events for usb activity91        """92        try:93            #Remove comments to enable usb local update on boot94            #if os.path.exists(DEVICE_PART1):95            #    syslog("start_usb_detection: Mount point exists")96            #    self.check_mount_point()97            context = Context()98            monitor = Monitor.from_netlink(context)99            monitor.filter_by(subsystem='usb')100            observer = MonitorObserver(monitor)101            observer.connect('device-event', self.device_event)102            monitor.start()103        except Exception as e:...

TIM4.py

Source:TIM4.py

...64    except FileNotFoundError:65        clear()66        print("Google Drive is not mounted at {}".format(volume))67        print("Running Backup Search")68        check_mount_point()69    return None70# Get User  input to find Google Drive Volume71def check_mount_point():72    mount_point = input("Please enter the Volume Label that Google Drive is mounted at:\n")73    mount_point.strip(':')74    try:75        os.chdir(mount_point +  ':')76        clear()77        print("Found Google Drive")78    except FileNotFoundError:79        clear()80        print("\'" + mount_point + ":\' does not exist")81        menu  =  {82            'text' : "1. Try a new Volume\n2. Exit",83            "menu_options" : {84                '1': check_mount_point,85                '2': on_close86            }87        }88        menu_action(menu,check_mount_point)89        # menu_action = input("1. Try a New Volume\n2. Exit")90        # if menu_action == '1':91        #     self.check_mount_point()92        #     return None93        # elif menu_action == '2':94        #     self.on_close()95        #     return  None96        # else:97        #     self.unexpected_action()98        #     self.check_mount_point()99            100#Utility Functions for use from main method101#######################################102def set_cwd(dir):103    os.chdir(dir)104    print("cwd set")105def clear():106    os.system('cls')107    return None108def on_close():109    print("\n\nClosing Session\n\n")110    quit()111def unexpected_action():112    clear()...

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.