test: add unit test coverage for rm, signurl, rpo, util, credentials, and storage_url modules

gslib/tests/test_rm.py

@@ -24,6 +24,7 @@
 import sys
 from unittest import mock
 
+from gslib.exception import CommandException
 from gslib.exception import NO_URLS_MATCHED_PREFIX
 from gslib.exception import NO_URLS_MATCHED_TARGET
 import gslib.tests.testcase as testcase
@@ -751,6 +752,25 @@ def test_rm_multiple_nonexistent_objects(self):
       self.assertIn('2 files/objects could not be removed.', stderr)
 
 
+class TestRmUnitTests(testcase.GsUtilUnitTestCase):
+  """Unit tests for gsutil rm command."""
+
+  def test_rm_no_arguments_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'expects at least one URL\.'):
+      self.RunCommand('rm', [])
+
+  def test_rm_stdin_with_arguments_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'No arguments allowed with the -I flag\.'):
+      self.RunCommand('rm', ['-I', 'gs://bucket/obj'])
+
+  def test_rm_local_file_fails(self):
+    with self.assertRaisesRegex(
+        CommandException, r'does not support "file://" URLs'):
+      self.RunCommand('rm', ['file:///tmp/foo.txt'])
+
+
 class TestRmUnitTestsWithShim(testcase.ShimUnitTestBase):
   """Unit tests for gsutil rm with shim."""
 

gslib/tests/test_rpo.py

@@ -20,6 +20,7 @@
 import textwrap
 
 from gslib.commands.rpo import RpoCommand
+from gslib.cs_api_map import ApiSelector
 from gslib.exception import CommandException
 from gslib.gcs_json_api import GcsJsonApi
 from gslib.storage_url import StorageUrlFromString
@@ -86,6 +87,40 @@ def test_invalid_subcommand_raises_error(self):
         CommandException, 'Invalid subcommand "blah", use get|set instead'):
       self.RunCommand('rpo', ['blah', 'DEFAULT', 'gs://boo*'])
 
+  def test_rpo_xml_api_disabled_raises(self):
+    # Mock api selector to return XML for gs.
+    with mock.patch(
+        'gslib.cloud_api_delegator.CloudApiDelegator.GetApiSelector',
+        return_value=ApiSelector.XML):
+      with self.assertRaisesRegex(
+          CommandException,
+          r'command can only be with the Cloud Storage JSON API'):
+        self.RunCommand('rpo', ['get', 'gs://bucket'])
+
+  def test_rpo_missing_args_raises(self):
+    # Min args is 2 (action subcommand + URL).
+    with self.assertRaisesRegex(
+        CommandException, r'requires at least 2 arguments'):
+      self.RunCommand('rpo', [])
+
+    with self.assertRaisesRegex(
+        CommandException, r'requires at least 2 arguments'):
+      self.RunCommand('rpo', ['get'])
+
+    with self.assertRaisesRegex(
+        CommandException, r'requires at least 2 arguments'):
+      self.RunCommand('rpo', ['set', 'ASYNC_TURBO'])
+
+  def test_rpo_non_gcs_rejection(self):
+    s3_bucket = self.CreateBucket(provider='s3')
+    with self.assertRaisesRegex(
+        CommandException, r'command can only be used for GCS buckets'):
+      self.RunCommand('rpo', ['get', suri(s3_bucket)])
+
+    with self.assertRaisesRegex(
+        CommandException, r'command can only be used for GCS buckets'):
+      self.RunCommand('rpo', ['set', 'ASYNC_TURBO', suri(s3_bucket)])
+
 
 class TestRpoUnitWithShim(testcase.ShimUnitTestBase):
 

gslib/tests/test_setmeta.py

@@ -26,6 +26,7 @@
 import six
 
 from gslib.commands import setmeta
+from gslib.exception import CommandException
 from gslib.cs_api_map import ApiSelector
 import gslib.tests.testcase as testcase
 from gslib.tests.testcase.integration_testcase import SkipForS3
@@ -307,6 +308,66 @@ def test_setmeta_raises_error_if_not_provided_headers(self):
         ' -h flag. See "gsutil help setmeta" for more information.', stderr)
 
 
+class TestSetMetaUnit(testcase.GsUtilUnitTestCase):
+  """Unit tests for gsutil setmeta command."""
+
+  def test_setmeta_no_headers_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'requires one or more headers to be provided'):
+      self.RunCommand('setmeta', ['gs://bucket/obj'])
+
+  def test_setmeta_bucket_url_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'must name an object'):
+      self.RunCommand(
+          'setmeta', ['-h', 'Content-Type:text/html', 'gs://bucket'])
+
+  def test_setmeta_canned_acl_disallowed(self):
+    with self.assertRaisesRegex(
+        CommandException, r'no longer allows canned ACLs'):
+      self.RunCommand(
+          'setmeta', ['-h', 'x-goog-acl:public-read', 'gs://bucket/obj'])
+
+    with self.assertRaisesRegex(
+        CommandException, r'no longer allows canned ACLs'):
+      self.RunCommand(
+          'setmeta', ['-h', 'x-amz-acl:public-read', 'gs://bucket/obj'])
+
+  def test_setmeta_duplicate_headers_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'Each header must appear at most once\.'):
+      self.RunCommand(
+          'setmeta',
+          ['-h', 'Content-Type:text/html', '-h', 'content-type:text/plain',
+           'gs://bucket/obj'])
+
+    with self.assertRaisesRegex(
+        CommandException, r'Each header must appear at most once\.'):
+      self.RunCommand(
+          'setmeta',
+          ['-h', 'Content-Type:text/html', '-h', 'Content-Type',
+           'gs://bucket/obj'])
+
+  def test_setmeta_invalid_header_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'Invalid or disallowed header'):
+      self.RunCommand(
+          'setmeta', ['-h', 'Invalid-Header:val', 'gs://bucket/obj'])
+
+  def test_setmeta_non_ascii_header_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'Invalid non-ASCII'):
+      self.RunCommand(
+          'setmeta', ['-h', 'x-goog-meta-héader:val', 'gs://bucket/obj'])
+
+  def test_setmeta_non_ascii_standard_value_raises(self):
+    with self.assertRaisesRegex(
+        CommandException, r'Invalid non-ASCII'):
+      self.RunCommand(
+          'setmeta',
+          ['-h', 'Cache-Control:public, max-age=3600ã', 'gs://bucket/obj'])
+
+
 class TestSetMetaShim(testcase.ShimUnitTestBase):
 
   @mock.patch.object(setmeta.SetMetaCommand, 'RunCommand', new=mock.Mock())

gslib/tests/test_signurl.py

@@ -277,6 +277,23 @@ def testSignUrlMissingUrlArgs(self):
     self.assertIn('The command requires a key file argument and one or more '
                   'URL arguments', stderr)
 
+  def testSignUrlNonGsUrlRaises(self):
+    """Tests signurl with a non-gs:// URL."""
+    stderr = self.RunGsUtil(
+        ['signurl', self._GetJSONKsFile(), 's3://bucket/obj'],
+        return_stderr=True,
+        expected_status=1)
+    self.assertIn('Can only create signed urls from gs:// urls', stderr)
+
+  def testSignUrlResumableBucketRaises(self):
+    """Tests signurl with RESUMABLE method on a bucket URL."""
+    stderr = self.RunGsUtil(
+        ['signurl', '-m', 'RESUMABLE', '-r', 'us-east1',
+         self._GetJSONKsFile(), 'gs://bucket'],
+        return_stderr=True,
+        expected_status=1)
+    self.assertIn('Resumable signed URLs require an object name.', stderr)
+
 
 @unittest.skipUnless(HAVE_CRYPTO, 'signurl requires cryptography library.')
 class UnitTestSignUrl(testcase.GsUtilUnitTestCase):

gslib/tests/test_storage_url.py

@@ -20,6 +20,7 @@
 from __future__ import unicode_literals
 
 import os
+import stat
 import sys
 
 from gslib.exception import CommandException
@@ -171,3 +172,65 @@ def test_not_raising_error_if_multiple_objects_with_recursion(self):
     urls = list(map(storage_url.StorageUrlFromString, ['gs://b/o', 'gs://b/p']))
     storage_url.RaiseErrorIfUrlsAreMixOfBucketsAndObjects(
         urls, recursion_requested=True)
+
+  def test_urls_are_for_single_provider(self):
+    self.assertTrue(
+        storage_url.UrlsAreForSingleProvider(['gs://b1', 'gs://b2/o']))
+    self.assertFalse(
+        storage_url.UrlsAreForSingleProvider(['gs://b1', 's3://b2']))
+    self.assertFalse(
+        storage_url.UrlsAreForSingleProvider([]))
+
+  def test_storage_url_with_generation_or_version(self):
+    url = storage_url.StorageUrlFromString('gs://bucket/obj#12345')
+    self.assertTrue(url.IsCloudUrl())
+    self.assertEqual(url.generation, '12345')
+    self.assertEqual(url.object_name, 'obj')
+
+    url = storage_url.StorageUrlFromString('s3://bucket/obj#versionid')
+    self.assertTrue(url.IsCloudUrl())
+    self.assertEqual(url.generation, 'versionid')
+    self.assertEqual(url.object_name, 'obj')
+
+  def test_raises_error_for_invalid_root_level_object_name(self):
+    with self.assertRaises(InvalidUrlError):
+      storage_url.StorageUrlFromString('gs://bucket/.')
+    with self.assertRaises(InvalidUrlError):
+      storage_url.StorageUrlFromString('gs://bucket/..')
+
+  def test_raises_error_for_unrecognized_scheme(self):
+    with self.assertRaisesRegex(
+        InvalidUrlError, r'Unrecognized scheme "http"'):
+      storage_url.StorageUrlFromString('http://example.com/file')
+
+  def test_storage_url_equality_and_hashing(self):
+    url1 = storage_url.StorageUrlFromString('gs://bucket/obj')
+    url2 = storage_url.StorageUrlFromString('gs://bucket/obj')
+    url3 = storage_url.StorageUrlFromString('gs://bucket/other')
+
+    self.assertEqual(url1, url2)
+    self.assertNotEqual(url1, url3)
+    self.assertEqual(hash(url1), hash(url2))
+    self.assertNotEqual(hash(url1), hash(url3))
+
+  def test_file_url_stream(self):
+    url = storage_url.StorageUrlFromString('-')
+    self.assertTrue(url.IsStream())
+
+    url2 = storage_url.StorageUrlFromString('file://-')
+    self.assertTrue(url2.IsStream())
+
+    url3 = storage_url.StorageUrlFromString('gs://bucket/obj')
+    with self.assertRaises(NotImplementedError):
+      url3.IsStream()
+
+  @mock.patch('os.stat')
+  def test_file_url_fifo(self, mock_os_stat):
+    mock_os_stat.return_value.st_mode = stat.S_IFIFO
+    url = storage_url.StorageUrlFromString('file:///tmp/fake-fifo')
+    self.assertTrue(url.IsFifo())
+
+    # CloudUrl does not support IsFifo
+    url2 = storage_url.StorageUrlFromString('gs://bucket/obj')
+    with self.assertRaises(NotImplementedError):
+      url2.IsFifo()

gslib/tests/test_tabcomplete.py

@@ -20,11 +20,18 @@
 from __future__ import unicode_literals
 
 import os
-import time
 import sys
+import time
+from unittest import mock
 
 from gslib.command import CreateOrGetGsutilLogger
 from gslib.tab_complete import CloudObjectCompleter
+from gslib.tab_complete import CloudOrLocalObjectCompleter
+from gslib.tab_complete import CompleterType
+from gslib.tab_complete import LocalObjectCompleter
+from gslib.tab_complete import LocalObjectOrCannedACLCompleter
+from gslib.tab_complete import MakeCompleter
+from gslib.tab_complete import NoOpCompleter
 from gslib.tab_complete import TAB_COMPLETE_CACHE_TTL
 from gslib.tab_complete import TabCompletionCache
 import gslib.tests.testcase as testcase
@@ -366,3 +373,63 @@ def test_prefix_caching_partial_results(self):
       results = completer(request)
 
       self.assertEqual([str(object_uri)], results)
+
+  def test_make_completer(self):
+    api = self.MakeGsUtilApi()
+    self.assertIsInstance(
+        MakeCompleter(CompleterType.NO_OP, api), NoOpCompleter)
+    self.assertIsInstance(
+        MakeCompleter(CompleterType.LOCAL_OBJECT, api), LocalObjectCompleter)
+    self.assertIsInstance(
+        MakeCompleter(CompleterType.LOCAL_OBJECT_OR_CANNED_ACL, api),
+        LocalObjectOrCannedACLCompleter)
+    self.assertIsInstance(
+        MakeCompleter(CompleterType.CLOUD_OR_LOCAL_OBJECT, api),
+        CloudOrLocalObjectCompleter)
+
+    with self.assertRaisesRegex(RuntimeError, 'Unknown completer'):
+      MakeCompleter('invalid_completer', api)
+
+  def test_no_op_completer(self):
+    completer = NoOpCompleter()
+    self.assertEqual([], completer('any_prefix'))
+
+  @mock.patch.object(LocalObjectCompleter, '__call__', autospec=True)
+  def test_local_object_or_canned_acl_completer(self, mock_local_completer):
+    mock_local_completer.return_value = ['/tmp/pubfile']
+    completer = LocalObjectOrCannedACLCompleter()
+    results = completer('pub')
+    # Should contain local file and canned ACLs starting with 'pub':
+    self.assertIn('/tmp/pubfile', results)
+    self.assertIn('public-read', results)
+    self.assertIn('public-read-write', results)
+
+  @mock.patch.object(LocalObjectCompleter, '__call__', autospec=True)
+  @mock.patch.object(CloudObjectCompleter, '__call__', autospec=True)
+  def test_cloud_or_local_object_completer(self, mock_cloud_completer,
+                                           mock_local_completer):
+    completer = CloudOrLocalObjectCompleter(self.MakeGsUtilApi())
+
+    # Cloud URL prefix -> routes to CloudObjectCompleter
+    completer('gs://')
+    mock_cloud_completer.assert_called_once()
+    mock_local_completer.assert_not_called()
+
+    # Local file path prefix -> routes to LocalObjectCompleter
+    mock_cloud_completer.reset_mock()
+    completer('/tmp/')
+    mock_local_completer.assert_called_once()
+    mock_cloud_completer.assert_not_called()
+
+  @mock.patch('gslib.tab_complete.CloudListingRequestThread', autospec=True)
+  @mock.patch('argcomplete.warn')
+  def test_cloud_object_completer_timeout(self, mock_warn, mock_thread_class):
+    mock_thread_class.return_value.is_alive.return_value = True
+
+    completer = CloudObjectCompleter(self.MakeGsUtilApi())
+    with SetBotoConfigForTest([('GSUtil', 'tab_completion_timeout', '1'),
+                               ('GSUtil', 'state_dir', self.CreateTempDir())]):
+      results = completer('gs://prefix')
+
+    self.assertEqual([], results)
+    mock_warn.assert_called_once()