| OLD | NEW |
| (Empty) |
| 1 # -*- coding: utf-8 -*- | |
| 2 # Copyright 2013 Google Inc. All Rights Reserved. | |
| 3 # | |
| 4 # Licensed under the Apache License, Version 2.0 (the "License"); | |
| 5 # you may not use this file except in compliance with the License. | |
| 6 # You may obtain a copy of the License at | |
| 7 # | |
| 8 # http://www.apache.org/licenses/LICENSE-2.0 | |
| 9 # | |
| 10 # Unless required by applicable law or agreed to in writing, software | |
| 11 # distributed under the License is distributed on an "AS IS" BASIS, | |
| 12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| 13 # See the License for the specific language governing permissions and | |
| 14 # limitations under the License. | |
| 15 """Tests for ls command.""" | |
| 16 | |
| 17 from __future__ import absolute_import | |
| 18 | |
| 19 import posixpath | |
| 20 import re | |
| 21 import subprocess | |
| 22 import sys | |
| 23 | |
| 24 import gslib | |
| 25 from gslib.cs_api_map import ApiSelector | |
| 26 import gslib.tests.testcase as testcase | |
| 27 from gslib.tests.testcase.integration_testcase import SkipForS3 | |
| 28 from gslib.tests.util import ObjectToURI as suri | |
| 29 from gslib.tests.util import unittest | |
| 30 from gslib.util import IS_WINDOWS | |
| 31 from gslib.util import Retry | |
| 32 from gslib.util import UTF8 | |
| 33 | |
| 34 | |
| 35 class TestLs(testcase.GsUtilIntegrationTestCase): | |
| 36 """Integration tests for ls command.""" | |
| 37 | |
| 38 def test_blank_ls(self): | |
| 39 self.RunGsUtil(['ls']) | |
| 40 | |
| 41 def test_empty_bucket(self): | |
| 42 bucket_uri = self.CreateBucket() | |
| 43 self.AssertNObjectsInBucket(bucket_uri, 0) | |
| 44 | |
| 45 def test_empty_bucket_with_b(self): | |
| 46 bucket_uri = self.CreateBucket() | |
| 47 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 48 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 49 def _Check1(): | |
| 50 stdout = self.RunGsUtil(['ls', '-b', suri(bucket_uri)], | |
| 51 return_stdout=True) | |
| 52 self.assertEqual('%s/\n' % suri(bucket_uri), stdout) | |
| 53 _Check1() | |
| 54 | |
| 55 def test_bucket_with_Lb(self): | |
| 56 """Tests ls -Lb.""" | |
| 57 bucket_uri = self.CreateBucket() | |
| 58 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 59 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 60 def _Check1(): | |
| 61 stdout = self.RunGsUtil(['ls', '-Lb', suri(bucket_uri)], | |
| 62 return_stdout=True) | |
| 63 self.assertIn(suri(bucket_uri), stdout) | |
| 64 self.assertNotIn('TOTAL:', stdout) | |
| 65 _Check1() | |
| 66 | |
| 67 def test_bucket_with_lb(self): | |
| 68 """Tests ls -lb.""" | |
| 69 bucket_uri = self.CreateBucket() | |
| 70 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 71 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 72 def _Check1(): | |
| 73 stdout = self.RunGsUtil(['ls', '-lb', suri(bucket_uri)], | |
| 74 return_stdout=True) | |
| 75 self.assertIn(suri(bucket_uri), stdout) | |
| 76 self.assertNotIn('TOTAL:', stdout) | |
| 77 _Check1() | |
| 78 | |
| 79 def test_bucket_list_wildcard(self): | |
| 80 """Tests listing multiple buckets with a wildcard.""" | |
| 81 random_prefix = self.MakeRandomTestString() | |
| 82 bucket1_name = self.MakeTempName('bucket', prefix=random_prefix) | |
| 83 bucket2_name = self.MakeTempName('bucket', prefix=random_prefix) | |
| 84 bucket1_uri = self.CreateBucket(bucket_name=bucket1_name) | |
| 85 bucket2_uri = self.CreateBucket(bucket_name=bucket2_name) | |
| 86 # This just double checks that the common prefix of the two buckets is what | |
| 87 # we think it should be (based on implementation detail of CreateBucket). | |
| 88 # We want to be careful when setting a wildcard on buckets to make sure we | |
| 89 # don't step outside the test buckets to affect other buckets. | |
| 90 common_prefix = posixpath.commonprefix([suri(bucket1_uri), | |
| 91 suri(bucket2_uri)]) | |
| 92 self.assertTrue(common_prefix.startswith( | |
| 93 '%s://%sgsutil-test-test_bucket_list_wildcard-bucket-' % | |
| 94 (self.default_provider, random_prefix))) | |
| 95 wildcard = '%s*' % common_prefix | |
| 96 | |
| 97 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 98 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 99 def _Check1(): | |
| 100 stdout = self.RunGsUtil(['ls', '-b', wildcard], return_stdout=True) | |
| 101 expected = set([suri(bucket1_uri) + '/', suri(bucket2_uri) + '/']) | |
| 102 actual = set(stdout.split()) | |
| 103 self.assertEqual(expected, actual) | |
| 104 _Check1() | |
| 105 | |
| 106 def test_nonexistent_bucket_with_ls(self): | |
| 107 """Tests a bucket that is known not to exist.""" | |
| 108 stderr = self.RunGsUtil( | |
| 109 ['ls', '-lb', 'gs://%s' % self.nonexistent_bucket_name], | |
| 110 return_stderr=True, expected_status=1) | |
| 111 self.assertIn('404', stderr) | |
| 112 | |
| 113 stderr = self.RunGsUtil( | |
| 114 ['ls', '-Lb', 'gs://%s' % self.nonexistent_bucket_name], | |
| 115 return_stderr=True, expected_status=1) | |
| 116 self.assertIn('404', stderr) | |
| 117 | |
| 118 stderr = self.RunGsUtil( | |
| 119 ['ls', '-b', 'gs://%s' % self.nonexistent_bucket_name], | |
| 120 return_stderr=True, expected_status=1) | |
| 121 self.assertIn('404', stderr) | |
| 122 | |
| 123 def test_list_missing_object(self): | |
| 124 """Tests listing a non-existent object.""" | |
| 125 bucket_uri = self.CreateBucket() | |
| 126 stderr = self.RunGsUtil(['ls', suri(bucket_uri, 'missing')], | |
| 127 return_stderr=True, expected_status=1) | |
| 128 self.assertIn('matched no objects', stderr) | |
| 129 | |
| 130 def test_with_one_object(self): | |
| 131 bucket_uri = self.CreateBucket() | |
| 132 obj_uri = self.CreateObject(bucket_uri=bucket_uri, contents='foo') | |
| 133 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 134 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 135 def _Check1(): | |
| 136 stdout = self.RunGsUtil(['ls', suri(bucket_uri)], return_stdout=True) | |
| 137 self.assertEqual('%s\n' % obj_uri, stdout) | |
| 138 _Check1() | |
| 139 | |
| 140 def test_subdir(self): | |
| 141 """Tests listing a bucket subdirectory.""" | |
| 142 bucket_uri = self.CreateBucket(test_objects=1) | |
| 143 k1_uri = bucket_uri.clone_replace_name('foo') | |
| 144 k1_uri.set_contents_from_string('baz') | |
| 145 k2_uri = bucket_uri.clone_replace_name('dir/foo') | |
| 146 k2_uri.set_contents_from_string('bar') | |
| 147 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 148 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 149 def _Check1(): | |
| 150 stdout = self.RunGsUtil(['ls', '%s/dir' % suri(bucket_uri)], | |
| 151 return_stdout=True) | |
| 152 self.assertEqual('%s\n' % suri(k2_uri), stdout) | |
| 153 stdout = self.RunGsUtil(['ls', suri(k1_uri)], return_stdout=True) | |
| 154 self.assertEqual('%s\n' % suri(k1_uri), stdout) | |
| 155 _Check1() | |
| 156 | |
| 157 def test_versioning(self): | |
| 158 """Tests listing a versioned bucket.""" | |
| 159 bucket1_uri = self.CreateBucket(test_objects=1) | |
| 160 bucket2_uri = self.CreateVersionedBucket(test_objects=1) | |
| 161 self.AssertNObjectsInBucket(bucket1_uri, 1, versioned=True) | |
| 162 bucket_list = list(bucket1_uri.list_bucket()) | |
| 163 | |
| 164 objuri = [bucket1_uri.clone_replace_key(key).versionless_uri | |
| 165 for key in bucket_list][0] | |
| 166 self.RunGsUtil(['cp', objuri, suri(bucket2_uri)]) | |
| 167 self.RunGsUtil(['cp', objuri, suri(bucket2_uri)]) | |
| 168 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 169 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 170 def _Check2(): | |
| 171 stdout = self.RunGsUtil(['ls', '-a', suri(bucket2_uri)], | |
| 172 return_stdout=True) | |
| 173 self.assertNumLines(stdout, 3) | |
| 174 stdout = self.RunGsUtil(['ls', '-la', suri(bucket2_uri)], | |
| 175 return_stdout=True) | |
| 176 self.assertIn('%s#' % bucket2_uri.clone_replace_name(bucket_list[0].name), | |
| 177 stdout) | |
| 178 self.assertIn('metageneration=', stdout) | |
| 179 _Check2() | |
| 180 | |
| 181 def test_etag(self): | |
| 182 """Tests that listing an object with an etag.""" | |
| 183 bucket_uri = self.CreateBucket() | |
| 184 obj_uri = self.CreateObject(bucket_uri=bucket_uri, contents='foo') | |
| 185 # TODO: When testcase setup can use JSON, match against the exact JSON | |
| 186 # etag. | |
| 187 etag = obj_uri.get_key().etag.strip('"\'') | |
| 188 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 189 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 190 def _Check1(): | |
| 191 stdout = self.RunGsUtil(['ls', '-l', suri(bucket_uri)], | |
| 192 return_stdout=True) | |
| 193 if self.test_api == ApiSelector.XML: | |
| 194 self.assertNotIn(etag, stdout) | |
| 195 else: | |
| 196 self.assertNotIn('etag=', stdout) | |
| 197 _Check1() | |
| 198 | |
| 199 def _Check2(): | |
| 200 stdout = self.RunGsUtil(['ls', '-le', suri(bucket_uri)], | |
| 201 return_stdout=True) | |
| 202 if self.test_api == ApiSelector.XML: | |
| 203 self.assertIn(etag, stdout) | |
| 204 else: | |
| 205 self.assertIn('etag=', stdout) | |
| 206 _Check2() | |
| 207 | |
| 208 def _Check3(): | |
| 209 stdout = self.RunGsUtil(['ls', '-ale', suri(bucket_uri)], | |
| 210 return_stdout=True) | |
| 211 if self.test_api == ApiSelector.XML: | |
| 212 self.assertIn(etag, stdout) | |
| 213 else: | |
| 214 self.assertIn('etag=', stdout) | |
| 215 _Check3() | |
| 216 | |
| 217 @SkipForS3('S3 bucket configuration values are not supported via ls.') | |
| 218 def test_location(self): | |
| 219 """Tests listing a bucket with location constraint.""" | |
| 220 bucket_uri = self.CreateBucket() | |
| 221 bucket_suri = suri(bucket_uri) | |
| 222 | |
| 223 # No location info | |
| 224 stdout = self.RunGsUtil(['ls', '-lb', bucket_suri], | |
| 225 return_stdout=True) | |
| 226 self.assertNotIn('Location constraint', stdout) | |
| 227 | |
| 228 # Default location constraint is US | |
| 229 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 230 return_stdout=True) | |
| 231 self.assertIn('Location constraint:\t\tUS', stdout) | |
| 232 | |
| 233 @SkipForS3('S3 bucket configuration values are not supported via ls.') | |
| 234 def test_logging(self): | |
| 235 """Tests listing a bucket with logging config.""" | |
| 236 bucket_uri = self.CreateBucket() | |
| 237 bucket_suri = suri(bucket_uri) | |
| 238 | |
| 239 # No logging info | |
| 240 stdout = self.RunGsUtil(['ls', '-lb', bucket_suri], | |
| 241 return_stdout=True) | |
| 242 self.assertNotIn('Logging configuration', stdout) | |
| 243 | |
| 244 # Logging configuration is absent by default | |
| 245 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 246 return_stdout=True) | |
| 247 self.assertIn('Logging configuration:\t\tNone', stdout) | |
| 248 | |
| 249 # Enable and check | |
| 250 self.RunGsUtil(['logging', 'set', 'on', '-b', bucket_suri, | |
| 251 bucket_suri]) | |
| 252 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 253 return_stdout=True) | |
| 254 self.assertIn('Logging configuration:\t\tPresent', stdout) | |
| 255 | |
| 256 # Disable and check | |
| 257 self.RunGsUtil(['logging', 'set', 'off', bucket_suri]) | |
| 258 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 259 return_stdout=True) | |
| 260 self.assertIn('Logging configuration:\t\tNone', stdout) | |
| 261 | |
| 262 @SkipForS3('S3 bucket configuration values are not supported via ls.') | |
| 263 def test_web(self): | |
| 264 """Tests listing a bucket with website config.""" | |
| 265 bucket_uri = self.CreateBucket() | |
| 266 bucket_suri = suri(bucket_uri) | |
| 267 | |
| 268 # No website configuration | |
| 269 stdout = self.RunGsUtil(['ls', '-lb', bucket_suri], | |
| 270 return_stdout=True) | |
| 271 self.assertNotIn('Website configuration', stdout) | |
| 272 | |
| 273 # Website configuration is absent by default | |
| 274 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 275 return_stdout=True) | |
| 276 self.assertIn('Website configuration:\t\tNone', stdout) | |
| 277 | |
| 278 # Initialize and check | |
| 279 self.RunGsUtil(['web', 'set', '-m', 'google.com', bucket_suri]) | |
| 280 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 281 return_stdout=True) | |
| 282 self.assertIn('Website configuration:\t\tPresent', stdout) | |
| 283 | |
| 284 # Clear and check | |
| 285 self.RunGsUtil(['web', 'set', bucket_suri]) | |
| 286 stdout = self.RunGsUtil(['ls', '-Lb', bucket_suri], | |
| 287 return_stdout=True) | |
| 288 self.assertIn('Website configuration:\t\tNone', stdout) | |
| 289 | |
| 290 def test_list_sizes(self): | |
| 291 """Tests various size listing options.""" | |
| 292 bucket_uri = self.CreateBucket() | |
| 293 self.CreateObject(bucket_uri=bucket_uri, contents='x' * 2048) | |
| 294 | |
| 295 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 296 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 297 def _Check1(): | |
| 298 stdout = self.RunGsUtil(['ls', '-l', suri(bucket_uri)], | |
| 299 return_stdout=True) | |
| 300 self.assertIn('2048', stdout) | |
| 301 _Check1() | |
| 302 | |
| 303 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 304 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 305 def _Check2(): | |
| 306 stdout = self.RunGsUtil(['ls', '-L', suri(bucket_uri)], | |
| 307 return_stdout=True) | |
| 308 self.assertIn('2048', stdout) | |
| 309 _Check2() | |
| 310 | |
| 311 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 312 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 313 def _Check3(): | |
| 314 stdout = self.RunGsUtil(['ls', '-al', suri(bucket_uri)], | |
| 315 return_stdout=True) | |
| 316 self.assertIn('2048', stdout) | |
| 317 _Check3() | |
| 318 | |
| 319 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 320 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 321 def _Check4(): | |
| 322 stdout = self.RunGsUtil(['ls', '-lh', suri(bucket_uri)], | |
| 323 return_stdout=True) | |
| 324 self.assertIn('2 KiB', stdout) | |
| 325 _Check4() | |
| 326 | |
| 327 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 328 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 329 def _Check5(): | |
| 330 stdout = self.RunGsUtil(['ls', '-alh', suri(bucket_uri)], | |
| 331 return_stdout=True) | |
| 332 self.assertIn('2 KiB', stdout) | |
| 333 _Check5() | |
| 334 | |
| 335 @unittest.skipIf(IS_WINDOWS, | |
| 336 'Unicode handling on Windows requires mods to site-packages') | |
| 337 def test_list_unicode_filename(self): | |
| 338 """Tests listing an object with a unicode filename.""" | |
| 339 # Note: This test fails on Windows (command.exe). I was able to get ls to | |
| 340 # output Unicode filenames correctly by hacking the UniStream class code | |
| 341 # shown at | |
| 342 # http://stackoverflow.com/questions/878972/windows-cmd-encoding-change-caus
es-python-crash/3259271 | |
| 343 # into the start of gslib/commands/ls.py, along with no-op flush and | |
| 344 # isastream functions (as an experiment). However, even with that change, | |
| 345 # the current test still fails, since it also needs to run that | |
| 346 # stdout/stderr-replacement code. That UniStream class replacement really | |
| 347 # needs to be added to the site-packages on Windows python. | |
| 348 object_name = u'Аудиоархив' | |
| 349 object_name_bytes = object_name.encode(UTF8) | |
| 350 bucket_uri = self.CreateVersionedBucket() | |
| 351 key_uri = self.CreateObject(bucket_uri=bucket_uri, contents='foo', | |
| 352 object_name=object_name) | |
| 353 self.AssertNObjectsInBucket(bucket_uri, 1, versioned=True) | |
| 354 stdout = self.RunGsUtil(['ls', '-ael', suri(key_uri)], | |
| 355 return_stdout=True) | |
| 356 self.assertIn(object_name_bytes, stdout) | |
| 357 if self.default_provider == 'gs': | |
| 358 self.assertIn(str(key_uri.generation), stdout) | |
| 359 self.assertIn( | |
| 360 'metageneration=%s' % key_uri.get_key().metageneration, stdout) | |
| 361 if self.test_api == ApiSelector.XML: | |
| 362 self.assertIn(key_uri.get_key().etag.strip('"\''), stdout) | |
| 363 else: | |
| 364 # TODO: When testcase setup can use JSON, match against the exact JSON | |
| 365 # etag. | |
| 366 self.assertIn('etag=', stdout) | |
| 367 elif self.default_provider == 's3': | |
| 368 self.assertIn(key_uri.version_id, stdout) | |
| 369 self.assertIn(key_uri.get_key().etag.strip('"\''), stdout) | |
| 370 | |
| 371 def test_list_gzip_content_length(self): | |
| 372 """Tests listing a gzipped object.""" | |
| 373 file_size = 10000 | |
| 374 file_contents = 'x' * file_size | |
| 375 fpath = self.CreateTempFile(contents=file_contents, file_name='foo.txt') | |
| 376 key_uri = self.CreateObject() | |
| 377 self.RunGsUtil(['cp', '-z', 'txt', suri(fpath), suri(key_uri)]) | |
| 378 | |
| 379 # Use @Retry as hedge against bucket listing eventual consistency. | |
| 380 @Retry(AssertionError, tries=3, timeout_secs=1) | |
| 381 def _Check1(): | |
| 382 stdout = self.RunGsUtil(['ls', '-L', suri(key_uri)], return_stdout=True) | |
| 383 self.assertRegexpMatches(stdout, r'Content-Encoding:\s+gzip') | |
| 384 find_content_length_re = r'Content-Length:\s+(?P<num>\d)' | |
| 385 self.assertRegexpMatches(stdout, find_content_length_re) | |
| 386 m = re.search(find_content_length_re, stdout) | |
| 387 content_length = int(m.group('num')) | |
| 388 self.assertGreater(content_length, 0) | |
| 389 self.assertLess(content_length, file_size) | |
| 390 _Check1() | |
| 391 | |
| 392 def test_output_chopped(self): | |
| 393 """Tests that gsutil still succeeds with a truncated stdout.""" | |
| 394 bucket_uri = self.CreateBucket(test_objects=2) | |
| 395 | |
| 396 # Run Python with the -u flag so output is not buffered. | |
| 397 gsutil_cmd = [ | |
| 398 sys.executable, '-u', gslib.GSUTIL_PATH, 'ls', suri(bucket_uri)] | |
| 399 # Set bufsize to 0 to make sure output is not buffered. | |
| 400 p = subprocess.Popen(gsutil_cmd, stdout=subprocess.PIPE, bufsize=0) | |
| 401 # Immediately close the stdout pipe so that gsutil gets a broken pipe error. | |
| 402 p.stdout.close() | |
| 403 p.wait() | |
| 404 # Make sure it still exited cleanly. | |
| 405 self.assertEqual(p.returncode, 0) | |
| 406 | |
| 407 def test_recursive_list_trailing_slash(self): | |
| 408 """Tests listing an object with a trailing slash.""" | |
| 409 bucket_uri = self.CreateBucket() | |
| 410 self.CreateObject(bucket_uri=bucket_uri, object_name='/', contents='foo') | |
| 411 self.AssertNObjectsInBucket(bucket_uri, 1) | |
| 412 stdout = self.RunGsUtil(['ls', '-R', suri(bucket_uri)], return_stdout=True) | |
| 413 # Note: The suri function normalizes the URI, so the double slash gets | |
| 414 # removed. | |
| 415 self.assertIn(suri(bucket_uri) + '/', stdout) | |
| 416 | |
| 417 def test_recursive_list_trailing_two_slash(self): | |
| 418 """Tests listing an object with two trailing slashes.""" | |
| 419 bucket_uri = self.CreateBucket() | |
| 420 self.CreateObject(bucket_uri=bucket_uri, object_name='//', contents='foo') | |
| 421 self.AssertNObjectsInBucket(bucket_uri, 1) | |
| 422 stdout = self.RunGsUtil(['ls', '-R', suri(bucket_uri)], return_stdout=True) | |
| 423 # Note: The suri function normalizes the URI, so the double slash gets | |
| 424 # removed. | |
| 425 self.assertIn(suri(bucket_uri) + '//', stdout) | |
| 426 | |
| 427 @SkipForS3('S3 anonymous access is not supported.') | |
| 428 def test_get_object_without_list_bucket_permission(self): | |
| 429 # Bucket is not publicly readable by default. | |
| 430 bucket_uri = self.CreateBucket() | |
| 431 object_uri = self.CreateObject(bucket_uri=bucket_uri, | |
| 432 object_name='permitted', contents='foo') | |
| 433 # Set this object to be publicly readable. | |
| 434 self.RunGsUtil(['acl', 'set', 'public-read', suri(object_uri)]) | |
| 435 # Drop credentials. | |
| 436 with self.SetAnonymousBotoCreds(): | |
| 437 stdout = self.RunGsUtil(['ls', '-L', suri(object_uri)], | |
| 438 return_stdout=True) | |
| 439 self.assertIn(suri(object_uri), stdout) | |
| OLD | NEW |