1# Copyright 2015 The Chromium 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 os
6import stat
7import unittest
8
9import mock
10from pyfakefs import fake_filesystem_unittest
11from py_utils import cloud_storage
12
13from dependency_manager import archive_info
14from dependency_manager import cloud_storage_info
15from dependency_manager import exceptions
16
17class CloudStorageInfoTest(unittest.TestCase):
18  def testInitCloudStorageInfoErrors(self):
19    # Must specify cloud storage information atomically.
20    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
21                      None, None, None, None)
22    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
23                      'cs_bucket', None, None, None)
24    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
25                      None, 'cs_hash', None, None)
26    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
27                      None, None, 'download_path', None)
28    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
29                      None, None, None, 'cs_remote_path')
30    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
31                      None, 'cs_hash', 'download_path', 'cs_remote_path')
32    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
33                      'cs_bucket', None, 'download_path', 'cs_remote_path')
34    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
35                      'cs_bucket', 'cs_hash', None, 'cs_remote_path')
36    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
37                      'cs_bucket', 'cs_hash', 'download_path', None)
38
39  def testInitWithVersion(self):
40    self.assertRaises(
41        ValueError, cloud_storage_info.CloudStorageInfo, None, None, None,
42        'cs_remote_path', version_in_cs='version_in_cs')
43    self.assertRaises(
44        ValueError, cloud_storage_info.CloudStorageInfo, None, 'cs_hash',
45        'download_path', 'cs_remote_path', version_in_cs='version_in_cs')
46
47    cs_info = cloud_storage_info.CloudStorageInfo(
48        'cs_bucket', 'cs_hash', 'download_path', 'cs_remote_path',
49        version_in_cs='version_in_cs')
50    self.assertEqual('cs_hash', cs_info._cs_hash)
51    self.assertEqual('cs_bucket', cs_info._cs_bucket)
52    self.assertEqual('cs_remote_path', cs_info._cs_remote_path)
53    self.assertEqual('download_path', cs_info._download_path)
54    self.assertEqual('version_in_cs', cs_info._version_in_cs)
55
56  def testInitWithArchiveInfoErrors(self):
57    zip_info = archive_info.ArchiveInfo(
58        'download_path', 'unzip_location', 'path_within_archive')
59    self.assertRaises(
60        ValueError, cloud_storage_info.CloudStorageInfo, None, None, None, None,
61        archive_info=zip_info)
62    self.assertRaises(
63        ValueError, cloud_storage_info.CloudStorageInfo, None, None, None,
64        'cs_remote_path', archive_info=zip_info)
65    self.assertRaises(
66        ValueError, cloud_storage_info.CloudStorageInfo, 'cs_bucket', 'cs_hash',
67        None, 'cs_remote_path', archive_info=zip_info)
68    self.assertRaises(ValueError, cloud_storage_info.CloudStorageInfo,
69                      'cs_bucket', 'cs_hash',
70                      'cs_remote_path', None, version_in_cs='version',
71                      archive_info=zip_info)
72
73
74  def testInitWithArchiveInfo(self):
75    zip_info = archive_info.ArchiveInfo(
76        'download_path', 'unzip_location', 'path_within_archive')
77    cs_info = cloud_storage_info.CloudStorageInfo(
78        'cs_bucket', 'cs_hash', 'download_path', 'cs_remote_path',
79        archive_info=zip_info)
80    self.assertEqual('cs_hash', cs_info._cs_hash)
81    self.assertEqual('cs_bucket', cs_info._cs_bucket)
82    self.assertEqual('cs_remote_path', cs_info._cs_remote_path)
83    self.assertEqual('download_path', cs_info._download_path)
84    self.assertEqual(zip_info, cs_info._archive_info)
85    self.assertFalse(cs_info._version_in_cs)
86
87  def testInitWithVersionAndArchiveInfo(self):
88    zip_info = archive_info.ArchiveInfo(
89        'download_path', 'unzip_location', 'path_within_archive')
90    cs_info = cloud_storage_info.CloudStorageInfo(
91        'cs_bucket', 'cs_hash', 'download_path',
92        'cs_remote_path', version_in_cs='version_in_cs',
93        archive_info=zip_info)
94    self.assertEqual('cs_hash', cs_info._cs_hash)
95    self.assertEqual('cs_bucket', cs_info._cs_bucket)
96    self.assertEqual('cs_remote_path', cs_info._cs_remote_path)
97    self.assertEqual('download_path', cs_info._download_path)
98    self.assertEqual(zip_info, cs_info._archive_info)
99    self.assertEqual('version_in_cs', cs_info._version_in_cs)
100
101  def testInitMinimumCloudStorageInfo(self):
102    cs_info = cloud_storage_info.CloudStorageInfo(
103        'cs_bucket',
104        'cs_hash', 'download_path',
105        'cs_remote_path')
106    self.assertEqual('cs_hash', cs_info._cs_hash)
107    self.assertEqual('cs_bucket', cs_info._cs_bucket)
108    self.assertEqual('cs_remote_path', cs_info._cs_remote_path)
109    self.assertEqual('download_path', cs_info._download_path)
110    self.assertFalse(cs_info._version_in_cs)
111    self.assertFalse(cs_info._archive_info)
112
113
114class TestGetRemotePath(fake_filesystem_unittest.TestCase):
115  def setUp(self):
116    self.setUpPyfakefs()
117    self.config_path = '/test/dep_config.json'
118    self.fs.CreateFile(self.config_path, contents='{}')
119    self.download_path = '/foo/download_path'
120    self.fs.CreateFile(
121        self.download_path, contents='1010110', st_mode=stat.S_IWOTH)
122    self.cs_info = cloud_storage_info.CloudStorageInfo(
123        'cs_bucket', 'cs_hash', self.download_path, 'cs_remote_path',
124        version_in_cs='1.2.3.4',)
125
126  def tearDown(self):
127    self.tearDownPyfakefs()
128
129  @mock.patch(
130      'py_utils.cloud_storage.GetIfHashChanged')
131  def testGetRemotePathNoArchive(self, cs_get_mock):
132    def _GetIfHashChangedMock(cs_path, download_path, bucket, file_hash):
133      del cs_path, bucket, file_hash
134      if not os.path.exists(download_path):
135        self.fs.CreateFile(download_path, contents='1010001010101010110101')
136    cs_get_mock.side_effect = _GetIfHashChangedMock
137    # All of the needed information is given, and the downloaded path exists
138    # after calling cloud storage.
139    self.assertEqual(
140        os.path.abspath(self.download_path),
141        self.cs_info.GetRemotePath())
142    self.assertTrue(os.stat(self.download_path).st_mode & stat.S_IXUSR)
143
144    # All of the needed information is given, but the downloaded path doesn't
145    # exists after calling cloud storage.
146    self.fs.RemoveObject(self.download_path)
147    cs_get_mock.side_effect = [True]  # pylint: disable=redefined-variable-type
148    self.assertRaises(
149        exceptions.FileNotFoundError, self.cs_info.GetRemotePath)
150
151  @mock.patch(
152      'dependency_manager.dependency_manager_util.UnzipArchive')
153  @mock.patch(
154      'dependency_manager.cloud_storage_info.cloud_storage.GetIfHashChanged') # pylint: disable=line-too-long
155  def testGetRemotePathWithArchive(self, cs_get_mock, unzip_mock):
156    def _GetIfHashChangedMock(cs_path, download_path, bucket, file_hash):
157      del cs_path, bucket, file_hash
158      if not os.path.exists(download_path):
159        self.fs.CreateFile(download_path, contents='1010001010101010110101')
160    cs_get_mock.side_effect = _GetIfHashChangedMock
161
162    unzip_path = os.path.join(
163        os.path.dirname(self.download_path), 'unzip_dir')
164    path_within_archive = os.path.join('path', 'within', 'archive')
165    dep_path = os.path.join(unzip_path, path_within_archive)
166    def _UnzipFileMock(archive_file, unzip_location, tmp_location=None):
167      del archive_file, tmp_location
168      self.fs.CreateFile(dep_path)
169      self.fs.CreateFile(os.path.join(unzip_location, 'extra', 'path'))
170      self.fs.CreateFile(os.path.join(unzip_location, 'another_extra_path'))
171    unzip_mock.side_effect = _UnzipFileMock
172
173    # Create a stale directory that's expected to get deleted
174    stale_unzip_path_glob = os.path.join(
175        os.path.dirname(self.download_path), 'unzip_dir_*')
176    stale_path = os.path.join(
177        os.path.dirname(self.download_path), 'unzip_dir_stale')
178    self.fs.CreateDirectory(stale_path)
179    self.fs.CreateFile(os.path.join(stale_path, 'some_file'))
180
181    self.assertFalse(os.path.exists(dep_path))
182    zip_info = archive_info.ArchiveInfo(
183        self.download_path, unzip_path, path_within_archive,
184        stale_unzip_path_glob)
185    self.cs_info = cloud_storage_info.CloudStorageInfo(
186        'cs_bucket', 'cs_hash', self.download_path, 'cs_remote_path',
187        version_in_cs='1.2.3.4', archive_info=zip_info)
188
189    self.assertFalse(unzip_mock.called)
190    self.assertEqual(
191        os.path.abspath(dep_path),
192        self.cs_info.GetRemotePath())
193    self.assertTrue(os.path.exists(dep_path))
194    self.assertTrue(stat.S_IMODE(os.stat(os.path.abspath(dep_path)).st_mode) &
195                    (stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR))
196    unzip_mock.assert_called_once_with(self.download_path, unzip_path)
197
198    # Stale directory should have been deleted
199    self.assertFalse(os.path.exists(stale_path))
200
201    # Should not need to unzip a second time, but should return the same path.
202    unzip_mock.reset_mock()
203    self.assertTrue(os.path.exists(dep_path))
204    self.assertEqual(
205        os.path.abspath(dep_path),
206        self.cs_info.GetRemotePath())
207    self.assertTrue(stat.S_IMODE(os.stat(os.path.abspath(dep_path)).st_mode) &
208                    (stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR))
209    self.assertFalse(unzip_mock.called)
210
211
212  @mock.patch(
213      'py_utils.cloud_storage.GetIfHashChanged')
214  def testGetRemotePathCloudStorageErrors(self, cs_get_mock):
215    cs_get_mock.side_effect = cloud_storage.CloudStorageError
216    self.assertRaises(cloud_storage.CloudStorageError,
217                      self.cs_info.GetRemotePath)
218
219    cs_get_mock.side_effect = cloud_storage.ServerError
220    self.assertRaises(cloud_storage.ServerError,
221                      self.cs_info.GetRemotePath)
222
223    cs_get_mock.side_effect = cloud_storage.NotFoundError
224    self.assertRaises(cloud_storage.NotFoundError,
225                      self.cs_info.GetRemotePath)
226
227    cs_get_mock.side_effect = cloud_storage.PermissionError
228    self.assertRaises(cloud_storage.PermissionError,
229                      self.cs_info.GetRemotePath)
230
231    cs_get_mock.side_effect = cloud_storage.CredentialsError
232    self.assertRaises(cloud_storage.CredentialsError,
233                      self.cs_info.GetRemotePath)
234