• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2019 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.
4import os
5
6from autotest_lib.client.bin import utils
7from autotest_lib.client.common_lib import error
8from autotest_lib.client.cros.enterprise import enterprise_policy_base
9
10from telemetry.core import exceptions
11
12
13class policy_DownloadDirectory(
14        enterprise_policy_base.EnterprisePolicyTest):
15    """
16    Test effect of DownloadDirectory policy on Chrome OS.
17
18    This policy can only be set to default or Google Drive. Test each case,
19    and verify the test download file goes to the proper location.
20
21    """
22    version = 1
23
24    POLICY_NAME = 'DownloadDirectory'
25    _DOWNLOAD_BASE = ('http://commondatastorage.googleapis.com/'
26                      'chromiumos-test-assets-public/audio_power/')
27    DOWNLOAD_DIR = '/home/chronos/user/Downloads/download'
28    POLICY_SETTING = {'PromptForDownloadLocation': False}
29
30    def run_once(self, case):
31        """
32        Setup and run the test configured for the specified test case.
33
34        @param case: Name of the test case to run.
35
36        """
37        if case:
38            self.POLICY_SETTING[self.POLICY_NAME] = '${google_drive}'
39
40        self.setup_case(user_policies=self.POLICY_SETTING,
41                        real_gaia=True)
42
43        self.GDRIVE_DIR = self._get_Gdrive_path()
44        self._wait_for_mount_ready()
45        self._clear_test_locations()
46        self._the_test(case)
47        self._clear_test_locations()
48
49    def _get_Gdrive_path(self):
50        """Returns the path for the Google Drive Mountpoint."""
51        self.GDRIVE_BASE = self._get_mount(False)
52        return '{}/root/download'.format(self.GDRIVE_BASE)
53
54    def _download_test_file(self):
55        """Loads to the test URL which automatically downloads the test file."""
56        try:
57            self.navigate_to_url(self._DOWNLOAD_BASE)
58        # This page empty with just a test download, so it will timeout.
59        #  TODO: crbug:1058141, invesitgate the telemetry crash caused here.
60        except (exceptions.TimeoutException, exceptions.InitializationError):
61            pass
62
63
64    def _the_test(self, case):
65        """
66        Download the test file, and verify it is in the proper directory,
67        and not in the incorrect directory
68
69        @param case: If the download location was set to Gdrive or not.
70
71        """
72        self._download_test_file()
73
74        if case:
75            if (not os.path.isfile(self.GDRIVE_DIR) or
76                    os.path.isfile(self.DOWNLOAD_DIR)):
77                raise error.TestError(
78                    'Download file not in Gdrive Dir, OR found in downloads.')
79        else:
80            if (not os.path.isfile(self.DOWNLOAD_DIR) or
81                    os.path.isfile(self.GDRIVE_DIR)):
82                raise error.TestError(
83                    'Download file in Gdrive Dir, OR not found in downloads.')
84
85    def _clear_test_locations(self):
86        """Deletes the 'download' file in both test locations."""
87        for test_dir in [self.DOWNLOAD_DIR, self.GDRIVE_DIR]:
88            try:
89                # Adding the * incase the file was downloaded multiple times.
90                utils.system_output('rm {}*'.format(test_dir))
91            # If there is no file present this will return -1. Thats OK!
92            except error.CmdError:
93                pass
94
95    def _wait_for_mount_ready(self):
96        """Wait for the mount to be ready."""
97        def _mount_ready():
98            try:
99                utils.system_output('ls {}/root/'.format(self.GDRIVE_BASE))
100                return True
101            except error.CmdError:
102                return False
103
104        utils.poll_for_condition(
105            lambda: (_mount_ready()),
106            exception=error.TestFail('mounts not ready ever'),
107            timeout=15,
108            sleep_interval=1,
109            desc='Polling for mounts to be ready.')
110
111    def _get_mount(self, case):
112        """Get the Google Drive mount path."""
113        e_msg = 'Should have found mountpoint but did not!'
114        # It may take some time until drivefs is started, so poll for the
115        # mountpoint until timeout.
116        utils.poll_for_condition(
117            lambda: (self._find_drivefs_mount() is not None),
118            exception=error.TestFail(e_msg),
119            timeout=10,
120            sleep_interval=1,
121            desc='Polling for Google Drive to load.')
122
123        mountpoint = self._find_drivefs_mount()
124        return mountpoint
125
126    def _find_drivefs_mount(self):
127        """Return the mount point of the drive if found, else return None."""
128        for mount in utils.mounts():
129            if mount['type'] == 'fuse.drivefs':
130                return mount['dest']
131        return None
132