1 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 # Author: Cosimo Alfarano <cosimo.alfarano (at] collabora.co.uk> 5 6 import logging, os 7 8 from autotest_lib.client.common_lib import error 9 from autotest_lib.client.cros import rtc 10 from autotest_lib.client.cros.power import sys_power 11 from cros import storage as storage_mod 12 13 14 class hardware_UsbMount(storage_mod.StorageTester): 15 version = 1 16 SECS_TO_SUSPEND = 10 17 _tmpfile = None 18 19 20 def cleanup(self): 21 # if the test fails with the device unmounted and before re-mounting, 22 # the test will be unable to properly cleanup, since we have no way to 23 # remove a tmp file from an unmounted device. 24 # For instance, this can happen if suspend won't work (e.g. it will 25 # reboot instead). 26 if self._tmpfile and os.path.isfile(self._tmpfile): 27 logging.debug('cleanup(): removing %s', self._tmpfile) 28 os.remove(self._tmpfile) 29 30 self.scanner.unmount_all() 31 32 super(hardware_UsbMount, self).cleanup() 33 34 35 def run_once(self, mount_cycles=10, filter_dict={'bus':'usb'}): 36 """ 37 @param mount_cycles: how many times to mount/unount Default: 10. 38 @param filter_dict: storage dictionary filter. 39 Default: match any device connected on the USB bus. 40 """ 41 # wait_for_device() returns (device_dictionary, 42 # time_spent_looking_for_it), and only the dictionary is relevant for 43 # this test 44 storage = self.wait_for_device(filter_dict, cycles=1, 45 mount_volume=True)[0] 46 47 if not os.path.ismount(storage['mountpoint']): 48 raise error.TestFail('filesystem %s mount failed' % filter_dict) 49 50 storage_filter = {'fs_uuid': storage['fs_uuid']} 51 # We cannot use autotemp.tempfile since we should close the descriptors 52 # everytime the storage device is un-mounted. 53 self._tmpfile = os.path.join(storage['mountpoint'], 54 'tempfile_usb_mount.tmp') 55 56 while mount_cycles: 57 mount_cycles -= 1 58 # Create a 1MiB random file and checksum it. 59 storage_mod.create_file(self._tmpfile, 1*1024*1024) 60 chksum = storage_mod.checksum_file(self._tmpfile) 61 62 logging.debug('storage to umount %s', storage) 63 64 # Umount the volume. 65 self.scanner.umount_volume(storage_dict=storage) 66 storage = self.wait_for_device(storage_filter, 67 mount_volume=False)[0] 68 if os.path.ismount(storage['mountpoint']): 69 raise error.TestFail('filesystem %s unmount failed ' % 70 storage_filter) 71 72 # Mount the volume back. 73 self.scanner.mount_volume(storage_dict=storage) 74 storage = self.wait_for_device(storage_filter, 75 mount_volume=False)[0] 76 if not os.path.ismount(storage['mountpoint']): 77 raise error.TestFail('filesystem %s mount failed' % 78 storage_filter) 79 80 # Check that the created file exists and has the same content. 81 if not os.path.isfile(self._tmpfile): 82 raise error.TestFail('%s: file not present after remounting' % 83 self._tmpfile) 84 85 if chksum != storage_mod.checksum_file(self._tmpfile): 86 raise error.TestFail('%s: file content changed after ' 87 'remounting' % self._tmpfile) 88 89 # Mount it, suspend and verify that after suspend-to-ram the 90 # device is still mounted 91 self.scanner.mount_volume(storage_dict=storage) 92 storage = self.wait_for_device(storage_filter, mount_volume=False)[0] 93 if not os.path.ismount(storage['mountpoint']): 94 raise error.TestFail('filesystem %s mount failed ' % storage) 95 96 sys_power.do_suspend(self.SECS_TO_SUSPEND) 97 98 # mount_volume=False because we don't want the method to mount if 99 # unmonted: we need to check its actual status right after suspend 100 storage = self.wait_for_device(storage_filter, mount_volume=False)[0] 101 102 if not os.path.ismount(storage['mountpoint']): 103 raise error.TestFail('filesystem %s not mounted after suspend' % 104 storage_filter) 105 106 if not os.path.isfile(self._tmpfile): 107 raise error.TestFail('%s: file not present anymore after ' 108 'remounting' % self._tmpfile) 109 110 if chksum != storage_mod.checksum_file(self._tmpfile): 111 raise error.TestFail('%s: file content changed after remounting' % 112 self._tmpfile) 113
