• 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
10from autotest_lib.client.cros.power import sys_power
11from cros import storage as storage_mod
12
13
14class 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