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 5 import os 6 import stat 7 import unittest 8 9 import mock 10 from pyfakefs import fake_filesystem_unittest 11 from py_utils import cloud_storage 12 13 from dependency_manager import archive_info 14 from dependency_manager import cloud_storage_info 15 from dependency_manager import exceptions 16 17 class 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 114 class 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] 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
