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) 356
