tools/telemetry/third_party/gsutilz/gslib/tests/test_tabcomplete.py

   1 # -*- coding: utf-8 -*-
   2 # Copyright 2014 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 """Integration tests for tab completion."""
  16
  17 from __future__ import absolute_import
  18
  19 import os
  20 import time
  21
  22 from gslib.command import CreateGsutilLogger
  23 from gslib.tab_complete import CloudObjectCompleter
  24 from gslib.tab_complete import TAB_COMPLETE_CACHE_TTL
  25 from gslib.tab_complete import TabCompletionCache
  26 import gslib.tests.testcase as testcase
  27 from gslib.tests.util import ARGCOMPLETE_AVAILABLE
  28 from gslib.tests.util import SetBotoConfigForTest
  29 from gslib.tests.util import unittest
  30 from gslib.tests.util import WorkingDirectory
  31 from gslib.util import GetTabCompletionCacheFilename
  32
  33
  34 @unittest.skipUnless(ARGCOMPLETE_AVAILABLE,
  35                      'Tab completion requires argcomplete')
  36 class TestTabComplete(testcase.GsUtilIntegrationTestCase):
  37   """Integration tests for tab completion."""
  38
  39   def setUp(self):
  40     super(TestTabComplete, self).setUp()
  41     self.logger = CreateGsutilLogger('tab_complete')
  42
  43   def test_single_bucket(self):
  44     """Tests tab completion matching a single bucket."""
  45
  46     bucket_base_name = self.MakeTempName('bucket')
  47     bucket_name = bucket_base_name + '-suffix'
  48     self.CreateBucket(bucket_name)
  49
  50     request = '%s://%s' % (self.default_provider, bucket_base_name)
  51     expected_result = '//%s/' % bucket_name
  52
  53     self.RunGsUtilTabCompletion(['ls', request],
  54                                 expected_results=[expected_result])
  55
  56   def test_bucket_only_single_bucket(self):
  57     """Tests bucket-only tab completion matching a single bucket."""
  58
  59     bucket_base_name = self.MakeTempName('bucket')
  60     bucket_name = bucket_base_name + '-s'
  61     self.CreateBucket(bucket_name)
  62
  63     request = '%s://%s' % (self.default_provider, bucket_base_name)
  64     expected_result = '//%s ' % bucket_name
  65
  66     self.RunGsUtilTabCompletion(['rb', request],
  67                                 expected_results=[expected_result])
  68
  69   def test_bucket_only_no_objects(self):
  70     """Tests that bucket-only tab completion doesn't match objects."""
  71
  72     object_base_name = self.MakeTempName('obj')
  73     object_name = object_base_name + '-suffix'
  74     object_uri = self.CreateObject(object_name=object_name, contents='data')
  75
  76     request = '%s://%s/%s' % (
  77         self.default_provider, object_uri.bucket_name, object_base_name)
  78
  79     self.RunGsUtilTabCompletion(['rb', request], expected_results=[])
  80
  81   def test_single_subdirectory(self):
  82     """Tests tab completion matching a single subdirectory."""
  83
  84     object_base_name = self.MakeTempName('obj')
  85     object_name = object_base_name + '/subobj'
  86     object_uri = self.CreateObject(object_name=object_name, contents='data')
  87
  88     request = '%s://%s/' % (self.default_provider, object_uri.bucket_name)
  89     expected_result = '//%s/%s/' % (object_uri.bucket_name, object_base_name)
  90
  91     self.RunGsUtilTabCompletion(['ls', request],
  92                                 expected_results=[expected_result])
  93
  94   def test_multiple_buckets(self):
  95     """Tests tab completion matching multiple buckets."""
  96
  97     bucket_base_name = self.MakeTempName('bucket')
  98     bucket1_name = bucket_base_name + '-suffix1'
  99     self.CreateBucket(bucket1_name)
 100     bucket2_name = bucket_base_name + '-suffix2'
 101     self.CreateBucket(bucket2_name)
 102
 103     request = '%s://%s' % (self.default_provider, bucket_base_name)
 104     expected_result1 = '//%s/' % bucket1_name
 105     expected_result2 = '//%s/' % bucket2_name
 106
 107     self.RunGsUtilTabCompletion(['ls', request], expected_results=[
 108         expected_result1, expected_result2])
 109
 110   def test_single_object(self):
 111     """Tests tab completion matching a single object."""
 112
 113     object_base_name = self.MakeTempName('obj')
 114     object_name = object_base_name + '-suffix'
 115     object_uri = self.CreateObject(object_name=object_name, contents='data')
 116
 117     request = '%s://%s/%s' % (
 118         self.default_provider, object_uri.bucket_name, object_base_name)
 119     expected_result = '//%s/%s ' % (object_uri.bucket_name, object_name)
 120
 121     self.RunGsUtilTabCompletion(['ls', request],
 122                                 expected_results=[expected_result])
 123
 124   def test_multiple_objects(self):
 125     """Tests tab completion matching multiple objects."""
 126
 127     bucket_uri = self.CreateBucket()
 128
 129     object_base_name = self.MakeTempName('obj')
 130     object1_name = object_base_name + '-suffix1'
 131     self.CreateObject(
 132         bucket_uri=bucket_uri, object_name=object1_name, contents='data')
 133     object2_name = object_base_name + '-suffix2'
 134     self.CreateObject(
 135         bucket_uri=bucket_uri, object_name=object2_name, contents='data')
 136
 137     request = '%s://%s/%s' % (
 138         self.default_provider, bucket_uri.bucket_name, object_base_name)
 139     expected_result1 = '//%s/%s' % (bucket_uri.bucket_name, object1_name)
 140     expected_result2 = '//%s/%s' % (bucket_uri.bucket_name, object2_name)
 141
 142     self.RunGsUtilTabCompletion(['ls', request], expected_results=[
 143         expected_result1, expected_result2])
 144
 145   def test_subcommands(self):
 146     """Tests tab completion for commands with subcommands."""
 147
 148     bucket_base_name = self.MakeTempName('bucket')
 149     bucket_name = bucket_base_name + '-suffix'
 150     self.CreateBucket(bucket_name)
 151
 152     bucket_request = '%s://%s' % (self.default_provider, bucket_base_name)
 153     expected_bucket_result = '//%s ' % bucket_name
 154
 155     local_file = 'a_local_file'
 156     local_dir = self.CreateTempDir(test_files=[local_file])
 157
 158     local_file_request = '%s%s' % (local_dir, os.sep)
 159     expected_local_file_result = '%s ' % os.path.join(local_dir, local_file)
 160
 161     # Should invoke Cloud bucket URL completer.
 162     self.RunGsUtilTabCompletion(['cors', 'get', bucket_request],
 163                                 expected_results=[expected_bucket_result])
 164
 165     # Should invoke File URL completer which should match the local file.
 166     self.RunGsUtilTabCompletion(['cors', 'set', local_file_request],
 167                                 expected_results=[expected_local_file_result])
 168
 169     # Should invoke Cloud bucket URL completer.
 170     self.RunGsUtilTabCompletion(['cors', 'set', 'some_file', bucket_request],
 171                                 expected_results=[expected_bucket_result])
 172
 173   def test_invalid_partial_bucket_name(self):
 174     """Tests tab completion with a partial URL that by itself is not valid.
 175
 176     The bucket name in a Cloud URL cannot end in a dash, but a partial URL
 177     during tab completion may end in a dash and completion should still work.
 178     """
 179
 180     bucket_base_name = self.MakeTempName('bucket')
 181     bucket_name = bucket_base_name + '-s'
 182     self.CreateBucket(bucket_name)
 183
 184     request = '%s://%s-' % (self.default_provider, bucket_base_name)
 185     expected_result = '//%s/' % bucket_name
 186
 187     self.RunGsUtilTabCompletion(['ls', request],
 188                                 expected_results=[expected_result])
 189
 190   def test_acl_argument(self):
 191     """Tests tab completion for ACL arguments."""
 192
 193     local_file = 'a_local_file'
 194     local_dir = self.CreateTempDir(test_files=[local_file])
 195
 196     local_file_request = '%s%s' % (local_dir, os.sep)
 197     expected_local_file_result = '%s ' % os.path.join(local_dir, local_file)
 198
 199     # Should invoke File URL completer which should match the local file.
 200     self.RunGsUtilTabCompletion(['acl', 'set', local_file_request],
 201                                 expected_results=[expected_local_file_result])
 202
 203     # Should match canned ACL name.
 204     self.RunGsUtilTabCompletion(['acl', 'set', 'priv'],
 205                                 expected_results=['private '])
 206
 207     local_file = 'priv_file'
 208     local_dir = self.CreateTempDir(test_files=[local_file])
 209     with WorkingDirectory(local_dir):
 210       # Should match both a file and a canned ACL since argument takes
 211       # either one.
 212       self.RunGsUtilTabCompletion(['acl', 'set', 'priv'],
 213                                   expected_results=[local_file, 'private'])
 214
 215
 216 def _WriteTabCompletionCache(prefix, results, timestamp=None,
 217                              partial_results=False):
 218   if timestamp is None:
 219     timestamp = time.time()
 220   cache = TabCompletionCache(prefix, results, timestamp, partial_results)
 221   cache.WriteToFile(GetTabCompletionCacheFilename())
 222
 223
 224 @unittest.skipUnless(ARGCOMPLETE_AVAILABLE,
 225                      'Tab completion requires argcomplete')
 226 class TestTabCompleteUnitTests(testcase.unit_testcase.GsUtilUnitTestCase):
 227   """Unit tests for tab completion."""
 228
 229   def test_cached_results(self):
 230     """Tests tab completion results returned from cache."""
 231
 232     with SetBotoConfigForTest([('GSUtil', 'state_dir', self.CreateTempDir())]):
 233       request = 'gs://prefix'
 234       cached_results = ['gs://prefix1', 'gs://prefix2']
 235
 236       _WriteTabCompletionCache(request, cached_results)
 237
 238       completer = CloudObjectCompleter(self.MakeGsUtilApi())
 239       results = completer(request)
 240
 241       self.assertEqual(cached_results, results)
 242
 243   def test_expired_cached_results(self):
 244     """Tests tab completion results not returned from cache when too old."""
 245
 246     with SetBotoConfigForTest([('GSUtil', 'state_dir', self.CreateTempDir())]):
 247       bucket_base_name = self.MakeTempName('bucket')
 248       bucket_name = bucket_base_name + '-suffix'
 249       self.CreateBucket(bucket_name)
 250
 251       request = '%s://%s' % (self.default_provider, bucket_base_name)
 252       expected_result = '%s://%s/' % (self.default_provider, bucket_name)
 253
 254       cached_results = ['//%s1' % bucket_name, '//%s2' % bucket_name]
 255
 256       _WriteTabCompletionCache(request, cached_results,
 257                                time.time() - TAB_COMPLETE_CACHE_TTL)
 258
 259       completer = CloudObjectCompleter(self.MakeGsUtilApi())
 260       results = completer(request)
 261
 262       self.assertEqual([expected_result], results)
 263
 264   def test_prefix_caching(self):
 265     """Tests tab completion results returned from cache with prefix match.
 266
 267     If the tab completion prefix is an extension of the cached prefix, tab
 268     completion should return results from the cache that start with the prefix.
 269     """
 270
 271     with SetBotoConfigForTest([('GSUtil', 'state_dir', self.CreateTempDir())]):
 272       cached_prefix = 'gs://prefix'
 273       cached_results = ['gs://prefix-first', 'gs://prefix-second']
 274       _WriteTabCompletionCache(cached_prefix, cached_results)
 275
 276       request = 'gs://prefix-f'
 277       completer = CloudObjectCompleter(self.MakeGsUtilApi())
 278       results = completer(request)
 279
 280       self.assertEqual(['gs://prefix-first'], results)
 281
 282   def test_prefix_caching_boundary(self):
 283     """Tests tab completion prefix caching not spanning directory boundaries.
 284
 285     If the tab completion prefix is an extension of the cached prefix, but is
 286     not within the same bucket/sub-directory then the cached results should not
 287     be used.
 288     """
 289
 290     with SetBotoConfigForTest([('GSUtil', 'state_dir', self.CreateTempDir())]):
 291       object_uri = self.CreateObject(
 292           object_name='subdir/subobj', contents='test data')
 293
 294       cached_prefix = '%s://%s/' % (
 295           self.default_provider, object_uri.bucket_name)
 296       cached_results = ['%s://%s/subdir' % (
 297           self.default_provider, object_uri.bucket_name)]
 298       _WriteTabCompletionCache(cached_prefix, cached_results)
 299
 300       request = '%s://%s/subdir/' % (
 301           self.default_provider, object_uri.bucket_name)
 302       expected_result = '%s://%s/subdir/subobj' % (
 303           self.default_provider, object_uri.bucket_name)
 304
 305       completer = CloudObjectCompleter(self.MakeGsUtilApi())
 306       results = completer(request)
 307
 308       self.assertEqual([expected_result], results)
 309
 310   def test_prefix_caching_no_results(self):
 311     """Tests tab completion returning empty result set using cached prefix.
 312
 313     If the tab completion prefix is an extension of the cached prefix, but does
 314     not match any of the cached results then no remote request should be made
 315     and an empty result set should be returned.
 316     """
 317
 318     with SetBotoConfigForTest([('GSUtil', 'state_dir', self.CreateTempDir())]):
 319       object_uri = self.CreateObject(object_name='obj', contents='test data')
 320
 321       cached_prefix = '%s://%s/' % (
 322           self.default_provider, object_uri.bucket_name)
 323       cached_results = []
 324       _WriteTabCompletionCache(cached_prefix, cached_results)
 325
 326       request = '%s://%s/o' % (self.default_provider, object_uri.bucket_name)
 327
 328       completer = CloudObjectCompleter(self.MakeGsUtilApi())
 329       results = completer(request)
 330
 331       self.assertEqual([], results)
 332
 333   def test_prefix_caching_partial_results(self):
 334     """Tests tab completion prefix matching ignoring partial cached results.
 335
 336     If the tab completion prefix is an extension of the cached prefix, but the
 337     cached result set is partial, the cached results should not be used because
 338     the matching results for the prefix may be incomplete.
 339     """
 340
 341     with SetBotoConfigForTest([('GSUtil', 'state_dir', self.CreateTempDir())]):
 342       object_uri = self.CreateObject(object_name='obj', contents='test data')
 343
 344       cached_prefix = '%s://%s/' % (
 345           self.default_provider, object_uri.bucket_name)
 346       cached_results = []
 347       _WriteTabCompletionCache(cached_prefix, cached_results,
 348                                partial_results=True)
 349
 350       request = '%s://%s/o' % (self.default_provider, object_uri.bucket_name)
 351
 352       completer = CloudObjectCompleter(self.MakeGsUtilApi())
 353       results = completer(request)
 354
 355       self.assertEqual([str(object_uri)], results)