Added Python 3.12 support

gslib/commands/hash.py

@@ -77,7 +77,7 @@
 _GCLOUD_FORMAT_STRING = (
     '--format=' + 'value[separator="",terminator=""](' + 'digest_format.sub("' +
     shim_util.get_format_flag_caret() + '", "Hashes ["),' + 'url.sub("' +
-    shim_util.get_format_flag_caret() + '", "] for ").sub("$", ":^\^n"),' +
+    shim_util.get_format_flag_caret() + '", "] for ").sub("$", ":^\\^n"),' +
     'md5_hash.yesno(yes="\tHash (md5):\t\t", no=""),' +
     'md5_hash.yesno(no=""),' + 'md5_hash.yesno(yes="NEWLINE", no="")' +
     '.sub("NEWLINE", "' + shim_util.get_format_flag_newline() + '"),' +

gslib/commands/rsync.py

@@ -493,19 +493,19 @@
                  path is always relative (similar to Unix rsync or tar exclude
                  options). For example, if you run the command:
 
-                   gsutil rsync -x "data.[/\\].*\.txt$" dir gs://my-bucket
+                   gsutil rsync -x "data.[/\\].*\\.txt$" dir gs://my-bucket
 
                  it skips the file dir/data1/a.txt.
 
                  You can use regex alternation to specify multiple exclusions,
                  for example:
 
-                   gsutil rsync -x ".*\.txt$|.*\.jpg$" dir gs://my-bucket
+                   gsutil rsync -x ".*\\.txt$|.*\\.jpg$" dir gs://my-bucket
 
                  skips all .txt and .jpg files in dir.
 
                  NOTE: When using the Windows cmd.exe command line interpreter,
-                 use ``^`` as an escape character instead of ``\`` and escape
+                 use ``^`` as an escape character instead of ``\\`` and escape
                  the ``|`` character. When using Windows PowerShell, use ``'``
                  instead of ``"`` and surround the ``|`` character with ``"``.
 
@@ -515,7 +515,7 @@
                  much faster, but won't work as intended with negative
                  lookahead patterns. For example, if you run the command:
 
-                   gsutil rsync -y "^(?!.*\.txt$).*" dir gs://my-bucket
+                   gsutil rsync -y "^(?!.*\\.txt$).*" dir gs://my-bucket
 
                  This would first exclude all subdirectories unless they end in
                  .txt before excluding all files except those ending in .txt.

gslib/tests/test_autoclass.py

@@ -30,22 +30,22 @@
 class TestAutoclassUnit(testcase.GsUtilUnitTestCase):
 
   def test_set_too_few_arguments_fails(self):
-    with self.assertRaisesRegexp(exception.CommandException,
+    with self.assertRaisesRegex(exception.CommandException,
                                  'command requires at least'):
       self.RunCommand('autoclass', ['set'])
 
   def test_get_too_few_arguments_fails(self):
-    with self.assertRaisesRegexp(exception.CommandException,
+    with self.assertRaisesRegex(exception.CommandException,
                                  'command requires at least'):
       self.RunCommand('autoclass', ['get'])
 
   def test_no_subcommand_fails(self):
-    with self.assertRaisesRegexp(exception.CommandException,
+    with self.assertRaisesRegex(exception.CommandException,
                                  'command requires at least'):
       self.RunCommand('autoclass', [])
 
   def test_invalid_subcommand_fails(self):
-    with self.assertRaisesRegexp(exception.CommandException,
+    with self.assertRaisesRegex(exception.CommandException,
                                  'Invalid subcommand'):
       self.RunCommand('autoclass', ['fakecommand', 'test'])
 

gslib/tests/test_command_runner.py

@@ -478,7 +478,7 @@ def test_valid_header_coding(self):
 
   def test_invalid_header_coding_fails(self):
     headers = {'content-type': 'bãr'}
-    with self.assertRaisesRegexp(CommandException, r'Invalid non-ASCII'):
+    with self.assertRaisesRegex(CommandException, r'Invalid non-ASCII'):
       HandleHeaderCoding(headers)
 
 

gslib/tests/test_copy_helper_funcs.py

@@ -108,7 +108,7 @@ def testGetPartitionInfo(self):
 
     # Test component_size > file_size (make sure we get at least two components.
     (num_components, component_size) = _GetPartitionInfo(100, 500, 51)
-    self.assertEquals(2, num_components)
+    self.assertEqual(2, num_components)
     self.assertEqual(50, component_size)
 
   def testFilterExistingComponentsNonVersioned(self):

gslib/tests/test_copy_objects_iterator.py

@@ -72,9 +72,9 @@ def test_iterator(self):
 
     for (src_string, dst_string) in src_dst_strings:
       copy_object_info = next(copy_objects_iterator)
-      self.assertEquals(src_string,
+      self.assertEqual(src_string,
                         copy_object_info.source_storage_url.object_name)
-      self.assertEquals(dst_string, copy_object_info.exp_dst_url.object_name)
+      self.assertEqual(dst_string, copy_object_info.exp_dst_url.object_name)
 
     iterator_ended = False
     try:
@@ -94,22 +94,22 @@ def test_iterator_metadata(self):
 
     self.assertFalse(copy_objects_iterator.has_cloud_src)
     self.assertFalse(copy_objects_iterator.has_file_src)
-    self.assertEquals(len(copy_objects_iterator.provider_types), 0)
+    self.assertEqual(len(copy_objects_iterator.provider_types), 0)
 
     next(copy_objects_iterator)
     self.assertTrue(copy_objects_iterator.has_cloud_src)
     self.assertFalse(copy_objects_iterator.has_file_src)
-    self.assertEquals(len(copy_objects_iterator.provider_types), 1)
+    self.assertEqual(len(copy_objects_iterator.provider_types), 1)
     self.assertTrue('gs' in copy_objects_iterator.provider_types)
 
     next(copy_objects_iterator)
     self.assertTrue(copy_objects_iterator.has_cloud_src)
     self.assertTrue(copy_objects_iterator.has_file_src)
-    self.assertEquals(len(copy_objects_iterator.provider_types), 2)
+    self.assertEqual(len(copy_objects_iterator.provider_types), 2)
     self.assertTrue('file' in copy_objects_iterator.provider_types)
     self.assertFalse(copy_objects_iterator.is_daisy_chain)
 
     next(copy_objects_iterator)
-    self.assertEquals(len(copy_objects_iterator.provider_types), 3)
+    self.assertEqual(len(copy_objects_iterator.provider_types), 3)
     self.assertTrue('s3' in copy_objects_iterator.provider_types)
     self.assertTrue(copy_objects_iterator.is_daisy_chain)

gslib/tests/test_cp.py

@@ -303,9 +303,9 @@ def TestCpMvPOSIXBucketToLocalErrors(cls, bucket_uri, obj, tmpdir, is_cp=True):
     listing1 = TailSet(suri(bucket_uri), cls.FlatListBucket(bucket_uri))
     listing2 = TailSet(tmpdir, cls.FlatListDir(tmpdir))
     # Bucket should have un-altered content.
-    cls.assertEquals(listing1, set(['/%s' % obj.object_name]))
+    cls.assertEqual(listing1, set(['/%s' % obj.object_name]))
     # Dir should have un-altered content.
-    cls.assertEquals(listing2, set(['']))
+    cls.assertEqual(listing2, set(['']))
 
 
 def TestCpMvPOSIXBucketToLocalNoErrors(cls, bucket_uri, tmpdir, is_cp=True):
@@ -376,7 +376,7 @@ def TestCpMvPOSIXBucketToLocalNoErrors(cls, bucket_uri, tmpdir, is_cp=True):
         ['cp' if is_cp else 'mv', '-P',
          suri(bucket_uri, obj_name), tmpdir])
   listing = TailSet(tmpdir, cls.FlatListDir(tmpdir))
-  cls.assertEquals(
+  cls.assertEqual(
       listing,
       set([
           '/obj1', '/obj2', '/obj3', '/obj4', '/obj5', '/obj6', '/obj7',
@@ -1356,24 +1356,24 @@ def _Check2():
       listing2 = self.RunGsUtil(['ls', '-la', suri(bucket2_uri)],
                                 return_stdout=True).split('\n')
       # 2 lines of listing output, 1 summary line, 1 empty line from \n split.
-      self.assertEquals(len(listing1), 4)
-      self.assertEquals(len(listing2), 4)
+      self.assertEqual(len(listing1), 4)
+      self.assertEqual(len(listing2), 4)
 
       # First object in each bucket should match in size and version-less name.
       size1, _, uri_str1, _ = listing1[0].split()
-      self.assertEquals(size1, str(len('data0')))
-      self.assertEquals(storage_uri(uri_str1).object_name, 'k')
+      self.assertEqual(size1, str(len('data0')))
+      self.assertEqual(storage_uri(uri_str1).object_name, 'k')
       size2, _, uri_str2, _ = listing2[0].split()
-      self.assertEquals(size2, str(len('data0')))
-      self.assertEquals(storage_uri(uri_str2).object_name, 'k')
+      self.assertEqual(size2, str(len('data0')))
+      self.assertEqual(storage_uri(uri_str2).object_name, 'k')
 
       # Similarly for second object in each bucket.
       size1, _, uri_str1, _ = listing1[1].split()
-      self.assertEquals(size1, str(len('longer_data1')))
-      self.assertEquals(storage_uri(uri_str1).object_name, 'k')
+      self.assertEqual(size1, str(len('longer_data1')))
+      self.assertEqual(storage_uri(uri_str1).object_name, 'k')
       size2, _, uri_str2, _ = listing2[1].split()
-      self.assertEquals(size2, str(len('longer_data1')))
-      self.assertEquals(storage_uri(uri_str2).object_name, 'k')
+      self.assertEqual(size2, str(len('longer_data1')))
+      self.assertEqual(storage_uri(uri_str2).object_name, 'k')
 
     _Check2()
 
@@ -1390,14 +1390,14 @@ def _Check3():
       listing2 = self.RunGsUtil(['ls', '-la', suri(bucket3_uri)],
                                 return_stdout=True).split('\n')
       # 2 lines of listing output, 1 summary line, 1 empty line from \n split.
-      self.assertEquals(len(listing1), 4)
+      self.assertEqual(len(listing1), 4)
       # 1 lines of listing output, 1 summary line, 1 empty line from \n split.
-      self.assertEquals(len(listing2), 3)
+      self.assertEqual(len(listing2), 3)
 
       # Live (second) object in bucket 1 should match the single live object.
       size1, _, uri_str1, _ = listing2[0].split()
-      self.assertEquals(size1, str(len('longer_data1')))
-      self.assertEquals(storage_uri(uri_str1).object_name, 'k')
+      self.assertEqual(size1, str(len('longer_data1')))
+      self.assertEqual(storage_uri(uri_str1).object_name, 'k')
 
     _Check3()
 
@@ -2264,14 +2264,14 @@ def test_copy_unicode_non_ascii_filename(self):
       stdout = self.RunGsUtil(['cat', suri(key_uri)],
                               return_stdout=True,
                               force_gsutil=True)
-      self.assertEquals(stdout.encode('ascii'), file_contents)
+      self.assertEqual(stdout.encode('ascii'), file_contents)
     with SetBotoConfigForTest([('GSUtil', 'resumable_threshold',
                                 str(START_CALLBACK_PER_BYTES * 3))]):
       self.RunGsUtil(['cp', fpath, suri(key_uri)], return_stderr=True)
       stdout = self.RunGsUtil(['cat', suri(key_uri)],
                               return_stdout=True,
                               force_gsutil=True)
-      self.assertEquals(stdout.encode('ascii'), file_contents)
+      self.assertEqual(stdout.encode('ascii'), file_contents)
 
   # Note: We originally one time implemented a test
   # (test_copy_invalid_unicode_filename) that invalid unicode filenames were
@@ -2654,7 +2654,7 @@ def test_cp_object_ending_with_slash(self):
     # Check that files in the subdir got copied even though subdir object
     # download was skipped.
     with open(os.path.join(tmpdir, bucket_uri.bucket_name, 'abc', 'def')) as f:
-      self.assertEquals('def', '\n'.join(f.readlines()))
+      self.assertEqual('def', '\n'.join(f.readlines()))
 
   def test_cp_without_read_access(self):
     """Tests that cp fails without read access to the object."""
@@ -3083,15 +3083,15 @@ def test_cp_progress_callbacks(self):
     with SetBotoConfigForTest([boto_config_for_test]):
       stderr = self.RunGsUtil(['cp', fpath, suri(bucket_uri)],
                               return_stderr=True)
-      self.assertEquals(1, stderr.count(final_progress_callback))
+      self.assertEqual(1, stderr.count(final_progress_callback))
     boto_config_for_test = ('GSUtil', 'resumable_threshold', str(2 * ONE_MIB))
     with SetBotoConfigForTest([boto_config_for_test]):
       stderr = self.RunGsUtil(['cp', fpath, suri(bucket_uri)],
                               return_stderr=True)
-      self.assertEquals(1, stderr.count(final_progress_callback))
+      self.assertEqual(1, stderr.count(final_progress_callback))
     stderr = self.RunGsUtil(['cp', suri(bucket_uri, 'foo'), fpath],
                             return_stderr=True)
-    self.assertEquals(1, stderr.count(final_progress_callback))
+    self.assertEqual(1, stderr.count(final_progress_callback))
 
   @SkipForS3('No resumable upload support for S3.')
   def test_cp_resumable_upload(self):
@@ -4913,7 +4913,7 @@ def testDownloadWithNoHashAvailable(self):
     log_handler = self.RunCommand('cp', [suri(object_uri), dst_dir],
                                   return_log_handler=True)
     warning_messages = log_handler.messages['warning']
-    self.assertEquals(2, len(warning_messages))
+    self.assertEqual(2, len(warning_messages))
     self.assertRegex(
         warning_messages[0], r'Non-MD5 etag \(12345\) present for key .*, '
         r'data integrity checks are not possible')
@@ -4952,7 +4952,7 @@ def test_cp_upload_respects_no_hashes(self):
       log_handler = self.RunCommand('cp', [fpath, suri(bucket_uri)],
                                     return_log_handler=True)
     warning_messages = log_handler.messages['warning']
-    self.assertEquals(1, len(warning_messages))
+    self.assertEqual(1, len(warning_messages))
     self.assertIn('Found no hashes to validate object upload',
                   warning_messages[0])
 

gslib/tests/test_daisy_chain_wrapper.py

@@ -120,7 +120,7 @@ def testDownloadSingleChunk(self):
       self._WriteFromWrapperToFile(daisy_chain_wrapper, upload_file)
       # Since the chunk size is >= the file size, only a single GetObjectMedia
       # call should be made.
-      self.assertEquals(mock_api.get_calls, 1)
+      self.assertEqual(mock_api.get_calls, 1)
       with open(upload_file, 'rb') as upload_stream:
         with open(self.test_data_file, 'rb') as download_stream:
           self.assertEqual(upload_stream.read(), download_stream.read())
@@ -202,7 +202,7 @@ def testDownloadWithZeroWrites(self):
         mock_api,
         download_chunk_size=self.test_data_file_len)
     self._WriteFromWrapperToFile(daisy_chain_wrapper, upload_file)
-    self.assertEquals(mock_api.get_calls, 1)
+    self.assertEqual(mock_api.get_calls, 1)
     with open(upload_file, 'rb') as upload_stream:
       with open(self.test_data_file, 'rb') as download_stream:
         self.assertEqual(upload_stream.read(), download_stream.read())
@@ -280,7 +280,7 @@ def testSeekAndReturn(self):
         if not data:
           break
         upload_stream.write(data)
-    self.assertEquals(mock_api.get_calls, 1)
+    self.assertEqual(mock_api.get_calls, 1)
     with open(upload_file, 'rb') as upload_stream:
       with open(self.test_data_file, 'rb') as download_stream:
         self.assertEqual(upload_stream.read(), download_stream.read())
@@ -305,7 +305,7 @@ def testRestartDownloadThread(self):
     daisy_chain_wrapper.read(TRANSFER_BUFFER_SIZE)
     daisy_chain_wrapper.seek(0)
     self._WriteFromWrapperToFile(daisy_chain_wrapper, upload_file)
-    self.assertEquals(mock_api.get_calls, 2)
+    self.assertEqual(mock_api.get_calls, 2)
     with open(upload_file, 'rb') as upload_stream:
       with open(self.test_data_file, 'rb') as download_stream:
         self.assertEqual(upload_stream.read(), download_stream.read())

gslib/tests/test_defacl.py

@@ -179,7 +179,7 @@ def testEmptyDefAcl(self):
     self.RunGsUtil(self._defacl_set_prefix + ['private', suri(bucket)])
     stdout = self.RunGsUtil(self._defacl_get_prefix + [suri(bucket)],
                             return_stdout=True)
-    self.assertEquals(stdout.rstrip(), '[]')
+    self.assertEqual(stdout.rstrip(), '[]')
     self.RunGsUtil(self._defacl_ch_prefix +
                    ['-u', self.USER_TEST_ADDRESS +
                     ':fc', suri(bucket)])

gslib/tests/test_execution_util.py

@@ -96,5 +96,5 @@ def testExternalCommandRaisesFormattedStderr(self, mock_Popen):
     mock_command_process.communicate.return_value = (None, b'error.\n')
     mock_Popen.return_value = mock_command_process
 
-    with self.assertRaisesRegexp(exception.ExternalBinaryError, 'error'):
+    with self.assertRaisesRegex(exception.ExternalBinaryError, 'error'):
       execution_util.ExecuteExternalCommand(['fake-command'])

gslib/tests/test_gcs_json_credentials.py

@@ -138,7 +138,7 @@ def set_store(store):
       self.assertTrue(gcs_json_credentials._HasGceCreds())
       client = gcs_json_api.GcsJsonApi(None, None, None, None)
       self.assertIsInstance(client.credentials, GceAssertionCredentials)
-      self.assertEquals(client.credentials.refresh_token, "rEfrEshtOkEn")
+      self.assertEqual(client.credentials.refresh_token, "rEfrEshtOkEn")
       self.assertIs(client.credentials.client_id, None)
 
   @mock.patch.object(GceAssertionCredentials,

gslib/tests/test_hash.py

@@ -77,7 +77,7 @@ def testHashWildcard(self):
                              return_stdout=True)
     # One summary line and two hash lines per file.
     num_expected_lines = num_test_files * (1 + 2)
-    self.assertEquals(len(stdout.splitlines()), num_expected_lines)
+    self.assertEqual(len(stdout.splitlines()), num_expected_lines)
 
   def testHashSelectAlg(self):
     tmp_file = self.CreateTempFile(contents=_TEST_FILE_CONTENTS)

gslib/tests/test_hmac.py

@@ -84,13 +84,13 @@ def AssertKeyMetadataMatches(self,
                                state='ACTIVE',
                                service_account='.*',
                                project='.*'):
-    self.assertRegexpMatches(output_string, r'Access ID %s:' % access_id)
-    self.assertRegexpMatches(output_string, r'\sState:\s+%s' % state)
-    self.assertRegexpMatches(output_string,
+    self.assertRegex(output_string, r'Access ID %s:' % access_id)
+    self.assertRegex(output_string, r'\sState:\s+%s' % state)
+    self.assertRegex(output_string,
                              r'\s+Service Account:\s+%s\n' % service_account)
-    self.assertRegexpMatches(output_string, r'\s+Project:\s+%s' % project)
-    self.assertRegexpMatches(output_string, r'\s+Time Created:\s+.*')
-    self.assertRegexpMatches(output_string, r'\s+Time Last Updated:\s+.*')
+    self.assertRegex(output_string, r'\s+Project:\s+%s' % project)
+    self.assertRegex(output_string, r'\s+Time Created:\s+.*')
+    self.assertRegex(output_string, r'\s+Time Last Updated:\s+.*')
 
   def CleanupHelper(self, access_id):
     # Set the key to inactive if it isn't already.
@@ -175,8 +175,8 @@ def test_create(self):
     stdout = self.RunGsUtil(['hmac', 'create', SERVICE_ACCOUNT],
                             return_stdout=True)
     try:
-      self.assertRegexpMatches(stdout, r'Access ID:\s+\S+')
-      self.assertRegexpMatches(stdout, r'Secret:\s+\S+')
+      self.assertRegex(stdout, r'Access ID:\s+\S+')
+      self.assertRegex(stdout, r'Secret:\s+\S+')
     finally:
       access_id = self.ExtractAccessId(stdout)
       self.CleanupHelper(access_id)