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

How to use unmount_partition method in autotest

Best Python code snippet using autotest_python

mynode_usb_extras.py

Source:mynode_usb_extras.py

...79    is_opendime = False80    if mount_partition(partition, "temp_check"):81        if os.path.isfile("/mnt/usb_extras/temp_check/support/opendime.png"):82            is_opendime = True83    unmount_partition("temp_check")84    return is_opendime85################################86## Device Handlers87################################88usb_device_id = 089                90class UsbDeviceHandler:91    def __init__(self):92        global usb_device_id93        self.id = usb_device_id94        usb_device_id = usb_device_id + 195    def to_dict(self):96        raise NotImplementedError97class OpendimeHandler(UsbDeviceHandler):98    def __init__(self, block_device, partition):99        super().__init__()100        self.device = block_device101        self.device_type = "opendime"102        self.partition = partition103        self.folder_name = f"opendime_{self.id}"104        self.state = "loading_1"105        self.http_server = None106        self.http_server_thread = None107    def to_dict(self):108        dict = {}109        dict["id"] = self.id110        dict["device_type"] = self.device_type111        dict["device"] = self.device112        dict["partition"] = self.partition113        dict["folder_name"] = self.folder_name114        dict["port"] = self.port115        dict["state"] = self.state116        return dict117    def start(self):118        try:119            if mount_partition(self.partition, self.folder_name, "rw"):120                # Check device state121                self.state = "loading_2"122                try:123                    readme_file = f"/mnt/usb_extras/{self.folder_name}/README.txt"124                    private_key_file = f"/mnt/usb_extras/{self.folder_name}/private-key.txt"125                    if os.path.isfile(readme_file):126                        with open(readme_file) as f:127                            content = f.read()128                            if "This Opendime is fresh and unused. It hasn't picked a private key yet." in content:129                                self.state = "new"130                                log_message("  Opendime in state 'new'")131                    if os.path.isfile(private_key_file):132                        with open(private_key_file) as f:133                            content = f.read()134                            if "SEALED" in content:135                                self.state = "sealed"136                                log_message("  Opendime in state 'sealed'")137                            else:138                                self.state = "unsealed"139                                log_message("  Opendime in state 'unsealed'")140                except Exception as e:141                    self.state = "error_reading_opendime"142                self.port = 8010 + (self.id % 10)143                self.http_server = HTTPServer(('', self.port), web_handler_from(f"/mnt/usb_extras/{self.folder_name}"))144                self.http_server_thread = Thread(target = self.http_server.serve_forever)145                self.http_server_thread.setDaemon(True)146                self.http_server_thread.start()147                return True148            else:149                log_message("Error mounting partition for opendime")150                return False151        except Exception as e:152            unmount_partition(self.folder_name)153            log_message("Opendime Start Exception: {}".format(str(e)))154            return False155    def stop(self):156        try:157            if self.http_server:158                self.http_server.shutdown()159            unmount_partition(self.folder_name)160        except Exception as e:161            log_message("Opendime Stop Exception: {}".format(str(e)))162################################163## check_usb_devices()164################################165def check_usb_devices():166    try:167        # if new event, reset state168        # if no new event and in state (mounted), jump to state machine169        170        # Set initial state171        set_usb_extras_state("detecting")172        os.system("umount /mnt/usb_extras")173        # Detect drives174        drives = find_unmounted_drives()175        log_message(f"Drives: {drives}")176        # Check exactly one extra drive found177        drive_count = len(drives)178        if drive_count == 0:179            log_message("No USB extras found.")180        else:181            set_usb_extras_state("processing")182            for drive in drives:183                # Check drive for partitions184                drive = drives[0]185                partitions = find_partitions_for_drive(drive)186                log_message(f"Drive {drive} paritions: {partitions}")187                num_partitions = len(partitions)188                if num_partitions == 0:189                    log_message("No partitions found. Nothing to do.")190                elif num_partitions == 1:191                    # Process partition192                    partition = partitions[0]193                    log_message("One partition found! Scanning...")194                    if check_partition_for_opendime(partition):195                        log_message("Found Opendime!")196                        opendime = OpendimeHandler(drive, partition)197                        if opendime.start():198                            add_usb_device(opendime)199                        else:200                            opendime.stop()201                    else:202                        log_message(f"Drive {drive} could not be detected.")203                else:204                    log_message(f"{num_partitions} partitions found. Not sure what to do.")205        # Successful scan post init or usb action detected, mark homepage refresh206        os.system("touch /tmp/homepage_needs_refresh")207            208    except Exception as e:209        log_message("Exception: {}".format(str(e)))210        set_usb_extras_state("error")211        reset_usb_devices()212        log_message("Caught exception. Delaying 30s.")213        time.sleep(30)214################################215## Main216################################217def main():218    # Setup219    os.system("mkdir -p /mnt/usb_extras")220    # Start fresh and check USB devices once on startup 221    unmount_partition("*")222    reset_usb_devices()223    check_usb_devices()224    # Monitor USB and re-check on add/remove225    context = pyudev.Context()226    monitor = pyudev.Monitor.from_netlink(context)227    monitor.filter_by(subsystem='usb')228    # this is module level logger, can be ignored229    log_message("Starting to monitor for usb")230    monitor.start()231    log_message("Waiting on USB Event...")232    set_usb_extras_state("waiting")233    for device in iter(monitor.poll, None):234        log_message("")235        log_message("Got USB event: %s", device.action)236        if device.action == 'add':237            check_usb_devices()238        else:239            # HANDLE DEVICE REMOVAL BETTER? This resets all and re-scans240            reset_usb_devices()241            check_usb_devices()242        log_message("Waiting on USB Event...")243        set_usb_extras_state("waiting")244    245@atexit.register246def goodbye():247    log_message("ATEXIT: Resetting devices")248    unmount_partition("*")249    reset_usb_devices()250    log_message("ATEXIT: Done")251# This is the main entry point for the program252if __name__ == "__main__":253    while True:254        try:255            main()256        except Exception as e:257            set_usb_extras_state("error")258            log_message("Main Exception: {}".format(str(e)))259            log_message("Caught exception. Delaying 30s.")260            unmount_partition("*")261            reset_usb_devices()...

mount.py

Source:mount.py

...26def unmount(mountpoint):27    subprocess.run(['umount', mountpoint])28    # The mount point should be empty, so we can use rmdir.29    os.rmdir(mountpoint)30def unmount_partition(partition):31    subprocess.run(['umount', '/dev/' + partition])32def unmount_all_partitions(device):33    partitions = usb_info.get_partitions(device)34    for partition in partitions:...

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.