1# Copyright (c) 2011 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
5import dbus
6
7from autotest_lib.client.bin import test
8from autotest_lib.client.common_lib import error
9from autotest_lib.client.cros.cros_disks import CrosDisksTester
10
11class CrosDisksAPITester(CrosDisksTester):
12
13    # See MountErrorType defined in system_api/dbus/cros-disks/dbus-constants.h
14    MOUNT_ERROR_INVALID_DEVICE_PATH = 100
15
16    def __init__(self, test):
17        super(CrosDisksAPITester, self).__init__(test)
18
19    def get_tests(self):
20        return [
21            self.test_enumerate_devices,
22            self.test_enumerate_auto_mountable_devices,
23            self.test_get_device_properties,
24            self.test_get_device_properties_of_nonexistent_device,
25            self.test_enumerate_auto_mountable_devices_are_not_on_boot_device,
26            self.test_enumerate_auto_mountable_devices_are_not_virtual,
27            self.test_mount_nonexistent_device,
28            self.test_mount_boot_device_rejected,
29            self.test_unmount_nonexistent_device,
30        ]
31
32    def validate_disk_properties(self, disk):
33        # Disk properties provided by the API
34        disk_properties = (
35            ('DeviceFile', dbus.String),
36            ('DeviceIsDrive', dbus.Boolean),
37            ('DeviceIsMediaAvailable', dbus.Boolean),
38            ('DeviceIsOnBootDevice', dbus.Boolean),
39            ('DeviceIsVirtual', dbus.Boolean),
40            ('DeviceIsMounted', dbus.Boolean),
41            ('DeviceIsReadOnly', dbus.Boolean),
42            ('DeviceMediaType', dbus.UInt32),
43            ('DeviceMountPaths', dbus.Array),
44            ('DevicePresentationHide', dbus.Boolean),
45            ('DeviceSize', dbus.UInt64),
46            ('DriveIsRotational', dbus.Boolean),
47            ('DriveModel', dbus.String),
48            ('IdLabel', dbus.String),
49            ('NativePath', dbus.String),
50            ('FileSystemType', dbus.String),
51        )
52
53        for (prop_name, prop_value_type) in disk_properties:
54            # Check if all disk properties are set.
55            if prop_name not in disk:
56                raise error.TestFail("disk.%s not found" % prop_name)
57
58            # Check if each disk property has the right data type.
59            prop_value = disk[prop_name]
60            if not isinstance(prop_value, prop_value_type):
61                raise error.TestFail(
62                        "disk.%s is %s, but %s expected"
63                        % (prop_name, type(prop_value), prop_value_type))
64
65        # Check if DeviceFile has a proper value.
66        if not disk['DeviceFile']:
67            raise error.TestFail(
68                    "disk.DeviceFile should not be empty")
69
70        # Check if the values of DeviceIsMounted and DeviceMountPaths
71        # are consistent.
72        mount_paths = disk['DeviceMountPaths']
73        if disk['DeviceIsMounted']:
74            if len(mount_paths) == 0:
75                raise error.TestFail(
76                        "disk.DeviceMountPaths should not be empty "
77                        "if disk.DeviceIsMounted is true")
78        else:
79            if len(mount_paths) != 0:
80                raise error.TestFail(
81                        "disk.DeviceMountPaths should be empty "
82                        "if disk.DeviceIsMounted is false")
83
84        if mount_paths.signature != dbus.Signature('s'):
85            raise error.TestFail(
86                    "disk.DeviceMountPaths should contain only strings")
87
88        for mount_path in mount_paths:
89            if not mount_path:
90                raise error.TestFail(
91                        "disk.DeviceMountPaths should not contain any "
92                        "empty string")
93
94    def test_enumerate_devices(self):
95        # Check if EnumerateDevices method returns a list of devices.
96        devices = self.cros_disks.enumerate_devices()
97        for device in devices:
98            if not device or not isinstance(device, dbus.String):
99                raise error.TestFail(
100                        "device returned by EnumerateDevices "
101                        "should be a non-empty string")
102
103    def test_enumerate_auto_mountable_devices(self):
104        # Check if EnumerateAutoMountableDevices method returns a list
105        # of devices.
106        devices = self.cros_disks.enumerate_auto_mountable_devices()
107        for device in devices:
108            if not device or not isinstance(device, dbus.String):
109                raise error.TestFail(
110                        "device returned by EnumerateAutoMountableDevices "
111                        "should be a non-empty string")
112
113    def test_enumerate_auto_mountable_devices_are_not_on_boot_device(self):
114        # Make sure EnumerateAutoMountableDevices method does not return
115        # any device that is on the boot device.
116        devices = self.cros_disks.enumerate_auto_mountable_devices()
117        for device in devices:
118            properties = self.cros_disks.get_device_properties(device)
119            if properties['DeviceIsOnBootDevice']:
120                raise error.TestFail(
121                        "device returned by EnumerateAutoMountableDevices "
122                        "should not be on boot device")
123
124    def test_enumerate_auto_mountable_devices_are_not_virtual(self):
125        # Make sure EnumerateAutoMountableDevices method does not return
126        # any device that is virtual.
127        devices = self.cros_disks.enumerate_auto_mountable_devices()
128        for device in devices:
129            properties = self.cros_disks.get_device_properties(device)
130            if properties['DeviceIsVirtual']:
131                raise error.TestFail(
132                        "device returned by EnumerateAutoMountableDevices "
133                        "should not be virtual")
134
135    def test_get_device_properties(self):
136        # Check if GetDeviceProperties method returns valid properties.
137        devices = self.cros_disks.enumerate_devices()
138        for device in devices:
139            properties = self.cros_disks.get_device_properties(device)
140            self.validate_disk_properties(properties)
141
142    def test_get_device_properties_of_nonexistent_device(self):
143        try:
144            properties = self.cros_disks.get_device_properties('/nonexistent')
145        except dbus.DBusException:
146            return
147        raise error.TestFail(
148            "GetDeviceProperties of a nonexistent device should fail")
149
150    def test_mount_nonexistent_device(self):
151        self.cros_disks.mount('/dev/nonexistent', '', [])
152        self.cros_disks.expect_mount_completion({
153            'source_path': '/dev/nonexistent',
154            'mount_path':  '',
155        })
156
157    def test_mount_boot_device_rejected(self):
158        # Check if EnumerateDevices method returns a list of devices.
159        devices = self.cros_disks.enumerate_devices()
160        for device in devices:
161            properties = self.cros_disks.get_device_properties(device)
162            self.validate_disk_properties(properties)
163            if not properties['DeviceIsOnBootDevice']:
164                continue
165
166            self.cros_disks.mount(device, '', [])
167            self.cros_disks.expect_mount_completion({
168                'source_path': device,
169                'mount_path': '',
170                'status': self.MOUNT_ERROR_INVALID_DEVICE_PATH
171            })
172
173    def test_unmount_nonexistent_device(self):
174        try:
175            self.cros_disks.unmount('/dev/nonexistent', [])
176        except dbus.DBusException:
177            return
178        raise error.TestFail("Unmounting a nonexistent device should fail")
179
180
181class platform_CrosDisksDBus(test.test):
182    version = 1
183
184    def run_once(self, *args, **kwargs):
185        tester = CrosDisksAPITester(self)
186        tester.run(*args, **kwargs)
187