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