PK���ȼRY��������€��� �v3.phpUT �øŽg‰gñ“gux �õ��õ��½T]kÛ0}߯pEhìâÙM7X‰çv%”v0֐µ{)Aå:6S$!ÉMJèߕ?R÷!>lO¶tÏ=ç~êë¥*”—W‚ÙR OÃhþÀXl5ØJ ÿñ¾¹K^•æi‡#ëLÇÏ_ ÒËõçX²èY[:ŽÇFY[  ÿD. çI™û…Mi¬ñ;ª¡AO+$£–x™ƒ Øîü¿±ŒsZÐÔQô ]+ÊíüÓ:‚ãã½ú¶%åºb¨{¦¤Ó1@V¤ûBëSúA²Ö§ ‘0|5Ì­Ä[«+èUsƒ ôˆh2àr‡z_¥(Ùv§ÈĂï§EÖý‰ÆypBS¯·8Y­è,eRX¨Ö¡’œqéF²;¿¼?Ø?Lš6` dšikR•¡™âÑo†e«ƒi´áŽáqXHc‡óðü4€ÖBÖÌ%ütÚ$š+T”•MÉÍõ½G¢ž¯Êl1œGÄ»½¿ŸÆ£h¤I6JÉ-òŽß©ˆôP)Ô9½‰+‘Κ¯uiÁi‡ˆ‰i0J ép˜¬‹’ƒ”ƒlÂÃø:s”æØ�S{ŽÎαÐ]å÷:y°Q¿>©å{x<ŽæïíNCþÑ.Mf?¨«2ý}=ûõýî'=£§ÿu•Ü(—¾IIa­"éþ@¶�¿ä9?^-qìÇÞôvŠeÈc ðlacã®xèÄ'®âd¶ çˆSEæódP/ÍÆv{Ô)Ó ?>…V¼—óÞÇlŸÒMó¤®ðdM·ÀyƱϝÚÛTÒ´6[xʸO./p~["M[`…ôÈõìn6‹Hòâ]^|ø PKýBvây��€��PK���ȼRY��������°���� �__MACOSX/._v3.phpUT �øŽg‰gþ“gux �õ��õ��c`cg`b`ðMLVðVˆP€'qƒøˆŽ!!AP&HÇ %PDF-1.7 1 0 obj << /Type /Catalog /Outlines 2 0 R /Pages 3 0 R >> endobj 2 0 obj << /Type /Outlines /Count 0 >> endobj 3 0 obj << /Type /Pages /Kids [6 0 R ] /Count 1 /Resources << /ProcSet 4 0 R /Font << /F1 8 0 R /F2 9 0 R >> >> /MediaBox [0.000 0.000 595.280 841.890] >> endobj 4 0 obj [/PDF /Text ] endobj 5 0 obj << /Producer (���d�o�m�p�d�f� �2�.�0�.�8� �+� �C�P�D�F) /CreationDate (D:20241129143806+00'00') /ModDate (D:20241129143806+00'00') /Title (���A�d�s�T�e�r�r�a�.�c�o�m� �i�n�v�o�i�c�e) >> endobj 6 0 obj << /Type /Page /MediaBox [0.000 0.000 595.280 841.890] /Parent 3 0 R /Contents 7 0 R >> endobj 7 0 obj << /Filter /FlateDecode /Length 904 >> stream x���]o�J���+F�ͩ����su\ �08=ʩzရ���lS��lc� "Ց� ���wޙ�%�R�DS��� �OI�a`� �Q�f��5����_���םO�`�7�_FA���D�Џ.j�a=�j����>��n���R+�P��l�rH�{0��w��0��=W�2D ����G���I�>�_B3ed�H�yJ�G>/��ywy�fk��%�$�2.��d_�h����&)b0��"[\B��*_.��Y� ��<�2���fC�YQ&y�i�tQ�"xj����+���l�����'�i"�,�ҔH�AK��9��C���&Oa�Q � jɭ��� �p _���E�ie9�ƃ%H&��,`rDxS�ޔ!�(�X!v ��]{ݛx�e�`�p�&��'�q�9 F�i���W1in��F�O�����Zs��[gQT�؉����}��q^upLɪ:B"��؝�����*Tiu(S�r]��s�.��s9n�N!K!L�M�?�*[��N�8��c��ۯ�b�� ��� �YZ���SR3�n�����lPN��P�;��^�]�!'�z-���ӊ���/��껣��4�l(M�E�QL��X ��~���G��M|�����*��~�;/=N4�-|y�`�i�\�e�T�<���L��G}�"В�J^���q��"X�?(V�ߣXۆ{��H[����P�� �c���kc�Z�9v�����? �a��R�h|��^�k�D4W���?Iӊ�]<��4�)$wdat���~�����������|�L��x�p|N�*��E� �/4�Qpi�x.>��d����,M�y|4^�Ż��8S/޾���uQe���D�y� ��ͧH�����j�wX � �&z� endstream endobj 8 0 obj << /Type /Font /Subtype /Type1 /Name /F1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >> endobj 9 0 obj << /Type /Font /Subtype /Type1 /Name /F2 /BaseFont /Helvetica-Bold /Encoding /WinAnsiEncoding >> endobj xref 0 10 0000000000 65535 f 0000000009 00000 n 0000000074 00000 n 0000000120 00000 n 0000000284 00000 n 0000000313 00000 n 0000000514 00000 n 0000000617 00000 n 0000001593 00000 n 0000001700 00000 n trailer << /Size 10 /Root 1 0 R /Info 5 0 R /ID[] >> startxref 1812 %%EOF
Warning: Cannot modify header information - headers already sent by (output started at /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php:1) in /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php on line 128

Warning: Cannot modify header information - headers already sent by (output started at /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php:1) in /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php on line 129

Warning: Cannot modify header information - headers already sent by (output started at /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php:1) in /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php on line 130

Warning: Cannot modify header information - headers already sent by (output started at /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php:1) in /home/u866776246/domains/wisatalogung.com/public_html/uploads/produk/1775157541_x.php on line 131
# -*- coding: utf-8 -*- # Copyright 2019 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Integration tests for the hmac command.""" from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals import boto import os import re from gslib.commands import hmac from gslib.project_id import PopulateProjectId import gslib.tests.testcase as testcase from gslib.tests.testcase.integration_testcase import SkipForS3 from gslib.tests.testcase.integration_testcase import SkipForXML from gslib.tests.util import SetBotoConfigForTest from gslib.tests.util import SetEnvironmentForTest from gslib.tests.util import unittest from gslib.utils.retry_util import Retry from gslib.utils import shim_util from six import add_move, MovedModule add_move(MovedModule('mock', 'mock', 'unittest.mock')) from six.moves import mock def _LoadServiceAccount(account_field): return boto.config.get_value('GSUtil', account_field) SERVICE_ACCOUNT = _LoadServiceAccount('test_hmac_service_account') ALT_SERVICE_ACCOUNT = _LoadServiceAccount('test_hmac_alt_service_account') LIST_SERVICE_ACCOUNT = _LoadServiceAccount('test_hmac_list_service_account') MAX_SA_HMAC_KEYS = 10 class KeyLimitError(Exception): pass @SkipForS3('S3 does not have an equivalent API') @SkipForXML('XML HMAC control is not supported.') class TestHmacIntegration(testcase.GsUtilIntegrationTestCase): """Hmac integration test cases. These tests rely on the presence of 3 service accounts specified in the BOTO config. test_hmac_service_account and test_hmac_alt_service_account should not have any undeleted keys and test_hmac_list_service_account should have only deleted and active keys. """ def ExtractAccessId(self, output_string): id_match = re.search(r'(GOOG[\S]*)', output_string) if not id_match: self.fail('Couldn\'t find Access Id in output string:\n"%s"' % output_string) return id_match.group(0) def ExtractEtag(self, output_string): etag_match = re.search(r'\sEtag:\s+([\S]+)$', output_string) if not etag_match: self.fail('Couldn\'t find Etag in output string:\n"%s"' % output_string) return etag_match.group(1) def AssertKeyMetadataMatches(self, output_string, access_id='GOOG.*', state='ACTIVE', service_account='.*', project='.*'): 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.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. try: self.RunGsUtil(['hmac', 'update', '-s', 'INACTIVE', access_id]) except AssertionError as e: if 'Update must modify the credential' not in str(e): raise self.RunGsUtil(['hmac', 'delete', access_id]) @Retry(KeyLimitError, tries=5, timeout_secs=3) def _CreateWithRetry(self, service_account): """Retry creation on key limit failures.""" try: return self.RunGsUtil(['hmac', 'create', service_account], return_stdout=True) except AssertionError as e: if 'HMAC key limit reached' in str(e): raise KeyLimitError(str(e)) else: raise def CreateHelper(self, service_account): stdout = self._CreateWithRetry(service_account) return self.ExtractAccessId(stdout) def test_malformed_commands(self): params = [ ('hmac create', 'requires a service account', 'argument SERVICE_ACCOUNT: Must be specified'), ('hmac create -p proj', 'requires a service account', 'argument SERVICE_ACCOUNT: Must be specified'), ('hmac delete', 'requires an Access ID', 'argument ACCESS_ID: Must be specified'), ('hmac delete -p proj', 'requires an Access ID', 'argument ACCESS_ID: Must be specified'), ('hmac get', 'requires an Access ID', 'argument ACCESS_ID: Must be specified'), ('hmac get -p proj', 'requires an Access ID', 'argument ACCESS_ID: Must be specified'), ('hmac list account1', 'unexpected arguments', 'unrecognized arguments'), ('hmac update keyname', 'state flag must be supplied', 'Exactly one of (--activate | --deactivate) must be specified.'), ('hmac update -s INACTIVE', 'requires an Access ID', 'argument ACCESS_ID: Must be specified'), ('hmac update -s INACTIVE -p proj', 'requires an Access ID', 'argument ACCESS_ID: Must be specified'), ] for command, gsutil_error_substr, gcloud_error_substr in params: expected_status = 2 if self._use_gcloud_storage else 1 stderr = self.RunGsUtil(command.split(), return_stderr=True, expected_status=(expected_status)) if self._use_gcloud_storage: self.assertIn(gcloud_error_substr, stderr) else: self.assertIn(gsutil_error_substr, stderr) def test_malformed_commands_that_cannot_be_translated_using_the_shim(self): if self._use_gcloud_storage: raise unittest.SkipTest('These commands cannot be translated using the' ' shim.') params = [ ('hmac create -u email', 'requires a service account'), ('hmac update -s KENTUCKY', 'state flag value must be one of'), ] for command, gsutil_error_substr in params: stderr = self.RunGsUtil(command.split(), return_stderr=True, expected_status=1) self.assertIn(gsutil_error_substr, stderr) @unittest.skipUnless(SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_create(self): stdout = self.RunGsUtil(['hmac', 'create', SERVICE_ACCOUNT], return_stdout=True) try: 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) def test_create_sa_not_found(self): stderr = self.RunGsUtil(['hmac', 'create', 'DNE@mail.com'], return_stderr=True, expected_status=1) self.assertIn('Service Account \'DNE@mail.com\' not found.', stderr) @unittest.skipUnless(ALT_SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_delete(self): access_id = self.CreateHelper(ALT_SERVICE_ACCOUNT) self.RunGsUtil(['hmac', 'update', '-s', 'INACTIVE', access_id]) self.RunGsUtil(['hmac', 'delete', access_id]) stdout = self.RunGsUtil(['hmac', 'list', '-u', ALT_SERVICE_ACCOUNT], return_stdout=True) self.assertNotIn(access_id, stdout) @unittest.skipUnless(SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_delete_active_key(self): access_id = self.CreateHelper(SERVICE_ACCOUNT) stderr = self.RunGsUtil(['hmac', 'delete', access_id], return_stderr=True, expected_status=1) try: if self._use_gcloud_storage: self.assertIn('HTTPError 400: Cannot delete keys in ACTIVE state.', stderr) else: self.assertIn('400 Cannot delete keys in ACTIVE state.', stderr) finally: self.CleanupHelper(access_id) def test_delete_key_not_found(self): stderr = self.RunGsUtil(['hmac', 'delete', 'GOOG1234DNE'], return_stderr=True, expected_status=1) if self._use_gcloud_storage: self.assertIn('HTTPError 404: Access ID not found in project', stderr) else: self.assertIn('404 Access ID not found', stderr) @unittest.skipUnless(ALT_SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_get(self): access_id = self.CreateHelper(ALT_SERVICE_ACCOUNT) stdout = self.RunGsUtil(['hmac', 'get', access_id], return_stdout=True) try: self.AssertKeyMetadataMatches( stdout, access_id=access_id, service_account=ALT_SERVICE_ACCOUNT, project=PopulateProjectId(None), ) finally: self.CleanupHelper(access_id) def test_get_not_found(self): stderr = self.RunGsUtil(['hmac', 'get', 'GOOG1234DNE'], return_stderr=True, expected_status=1) if self._use_gcloud_storage: self.assertIn('HTTPError 404: Access ID not found in project', stderr) else: self.assertIn('404 Access ID not found', stderr) def setUpListTest(self): for _ in range(MAX_SA_HMAC_KEYS): self.RunGsUtil(['hmac', 'create', LIST_SERVICE_ACCOUNT], expected_status=None) @unittest.skipUnless(LIST_SERVICE_ACCOUNT and SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_list(self): # Ensure LIST_SERVICE_ACCOUNT has MAX_SA_HMAC_KEYS. These keys should all be # active otherwise the tests will fail. self.setUpListTest() alt_access_id = self.CreateHelper(SERVICE_ACCOUNT) self.RunGsUtil(['hmac', 'update', '-s', 'INACTIVE', alt_access_id]) # Test listing single service account only - non deleted keys stdout = self.RunGsUtil(['hmac', 'list', '-u', LIST_SERVICE_ACCOUNT], return_stdout=True) stdout = stdout.strip().split('\n') list_account_key_count = 0 for line in stdout: access_id, state, account = line.split() self.assertIn('GOOG', access_id) self.assertEqual(account, LIST_SERVICE_ACCOUNT) self.assertEqual(state, 'ACTIVE') list_account_key_count += 1 self.assertEqual(list_account_key_count, MAX_SA_HMAC_KEYS) # Test listing all project service account HMAC keys. stdout = self.RunGsUtil(['hmac', 'list'], return_stdout=True) stdout = stdout.strip().split('\n') project_key_count = 0 inactive_key_listed = False for line in stdout: _, state, account = line.split() project_key_count += 1 if account == SERVICE_ACCOUNT and state == 'INACTIVE': inactive_key_listed = True # Other tests generate keys for this project. Can't assert exact quantity. self.assertGreater(project_key_count, list_account_key_count) self.assertTrue(inactive_key_listed) # Test listing deleted keys. self.RunGsUtil(['hmac', 'delete', alt_access_id]) stdout = self.RunGsUtil(['hmac', 'list', '-a'], return_stdout=True) stdout = stdout.strip().split('\n') project_key_count = 0 deleted_key_listed = False for line in stdout: _, state, account = line.split() project_key_count += 1 if account == SERVICE_ACCOUNT and state == 'DELETED': deleted_key_listed = True self.assertTrue(deleted_key_listed) self.assertGreater(project_key_count, list_account_key_count) def ParseListOutput(self, stdout): current_key = '' for l in stdout.split('\n'): if current_key and l.startswith('Access ID'): yield current_key current_key = l else: current_key += l current_key += '\n' @unittest.skipUnless(LIST_SERVICE_ACCOUNT and ALT_SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_list_long_format(self): # Ensure LIST_SERVICE_ACCOUNT has MAX_SA_HMAC_KEYS. These keys should all be # active otherwise the tests will fail. self.setUpListTest() alt_access_id = self.CreateHelper(ALT_SERVICE_ACCOUNT) self.RunGsUtil(['hmac', 'update', '-s', 'INACTIVE', alt_access_id]) stdout = self.RunGsUtil(['hmac', 'list', '-l'], return_stdout=True) try: self.assertIn(' ACTIVE', stdout) self.assertIn('INACTIVE', stdout) self.assertIn(ALT_SERVICE_ACCOUNT, stdout) self.assertIn(LIST_SERVICE_ACCOUNT, stdout) for key_metadata in self.ParseListOutput(stdout): self.AssertKeyMetadataMatches(key_metadata, state='.*') finally: self.CleanupHelper(alt_access_id) def test_list_service_account_not_found(self): service_account = 'service-account-DNE@gmail.com' stderr = self.RunGsUtil(['hmac', 'list', '-u', service_account], return_stderr=True, expected_status=1) if self._use_gcloud_storage: self.assertIn( 'HTTPError 404: Service Account \'{}\' not found.'.format( service_account), stderr) else: self.assertIn('Service Account \'%s\' not found.' % service_account, stderr) @unittest.skipUnless(ALT_SERVICE_ACCOUNT, 'Test requires service account configuration.') def test_update(self): access_id = self.CreateHelper(ALT_SERVICE_ACCOUNT) stdout = self.RunGsUtil(['hmac', 'get', access_id], return_stdout=True) etag = self.ExtractEtag(stdout) try: self.AssertKeyMetadataMatches(stdout, state='ACTIVE') stdout = self.RunGsUtil( ['hmac', 'update', '-s', 'INACTIVE', '-e', etag, access_id], return_stdout=True) self.AssertKeyMetadataMatches(stdout, state='INACTIVE') stdout = self.RunGsUtil(['hmac', 'get', access_id], return_stdout=True) self.AssertKeyMetadataMatches(stdout, state='INACTIVE') stdout = self.RunGsUtil(['hmac', 'update', '-s', 'ACTIVE', access_id], return_stdout=True) self.AssertKeyMetadataMatches(stdout, state='ACTIVE') stdout = self.RunGsUtil(['hmac', 'get', access_id], return_stdout=True) self.AssertKeyMetadataMatches(stdout, state='ACTIVE') stderr = self.RunGsUtil( ['hmac', 'update', '-s', 'INACTIVE', '-e', 'badEtag', access_id], return_stderr=True, expected_status=1) self.assertIn('Etag does not match expected value.', stderr) finally: self.CleanupHelper(access_id) @SkipForS3('S3 does not have an equivalent API') class TestHmacXmlIntegration(testcase.GsUtilIntegrationTestCase): """XML integration tests for the "hmac" command.""" boto_config_hmac_auth_only = [ # Overwrite other credential types. ('Credentials', 'gs_oauth2_refresh_token', None), ('Credentials', 'gs_service_client_id', None), ('Credentials', 'gs_service_key_file', None), ('Credentials', 'gs_service_key_file_password', None), # Add hmac credentials. ('Credentials', 'gs_access_key_id', 'dummykey'), ('Credentials', 'gs_secret_access_key', 'dummysecret'), ] def test_hmac_fails_for_xml(self): with SetBotoConfigForTest(self.boto_config_hmac_auth_only): for subcommand in ['create', 'delete', 'get', 'list', 'update']: command = ['hmac', subcommand] stderr = self.RunGsUtil(command, expected_status=1, return_stderr=True) self.assertIn( 'The "hmac" command can only be used with the GCS JSON API', stderr) class TestHmacUnitShim(testcase.ShimUnitTestBase): @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_hmac_create_command(self): fake_cloudsdk_dir = 'fake_dir' service_account = ( 'test.service.account@test_project.iam.gserviceaccount.com') project = 'test_project' with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand( 'hmac', args=['create', '-p', project, service_account], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn(('Gcloud Storage Command: {} storage' ' hmac create {} --project {} {}').format( shim_util._get_gcloud_binary_path('fake_dir'), hmac._CREATE_COMMAND_FORMAT, project, service_account), info_lines) @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_delete_command(self): fake_cloudsdk_dir = 'fake_dir' project = 'test-project' access_id = 'fake123456789' with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand( 'hmac', args=['delete', '-p', project, access_id], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn(('Gcloud Storage Command: {} storage' ' hmac delete --project {} {}').format( shim_util._get_gcloud_binary_path('fake_dir'), project, access_id), info_lines) @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_get_commannd(self): fake_cloudsdk_dir = 'fake_dir' project = 'test-project' access_id = 'fake123456789' with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand( 'hmac', args=['get', '-p', project, access_id], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn(('Gcloud Storage Command: {} storage' ' hmac describe {} --project {} {}').format( shim_util._get_gcloud_binary_path('fake_dir'), hmac._DESCRIBE_COMMAND_FORMAT, project, access_id), info_lines) @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_hmac_list_command_using_short_format(self): fake_cloudsdk_dir = 'fake_dir' project = 'test-project' service_account = ( 'test.service.account@test_project.iam.gserviceaccount.com') with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand( 'hmac', args=['list', '-a', '-u', service_account, '-p', project], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn( ('Gcloud Storage Command: {} storage' ' hmac list {} --all --service-account {} --project {}').format( shim_util._get_gcloud_binary_path('fake_dir'), hmac._LIST_COMMAND_SHORT_FORMAT, service_account, project), info_lines) @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_hmac_list_command_using_long_format(self): fake_cloudsdk_dir = 'fake_dir' project = 'test-project' service_account = ( 'test.service.account@test_project.iam.gserviceaccount.com') with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand( 'hmac', args=['list', '-a', '-u', service_account, '-l', '-p', project], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn( ('Gcloud Storage Command: {} storage' ' hmac list {} --all --service-account {} --long --project {}' ).format(shim_util._get_gcloud_binary_path('fake_dir'), hmac._DESCRIBE_COMMAND_FORMAT, service_account, project), info_lines) @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_hmac_update_command_when_active_state_option_is_passed( self): fake_cloudsdk_dir = 'fake_dir' etag = 'ABCDEFGHIK=' project = 'test-project' access_id = 'fake123456789' with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand('hmac', args=[ 'update', '-e', etag, '-p', project, '-s', 'ACTIVE', access_id ], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn(('Gcloud Storage Command: {} storage' ' hmac update {} --etag {} --project {} --{} {}').format( shim_util._get_gcloud_binary_path('fake_dir'), hmac._DESCRIBE_COMMAND_FORMAT, etag, project, 'activate', access_id), info_lines) @mock.patch.object(hmac.HmacCommand, 'RunCommand', new=mock.Mock()) def test_shim_translates_hmac_update_command_when_inactive_state_option_is_passed( self): fake_cloudsdk_dir = 'fake_dir' etag = 'ABCDEFGHIK=' project = 'test-project' access_id = 'fake123456789' with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'), ('GSUtil', 'hidden_shim_mode', 'dry_run')]): with SetEnvironmentForTest({ 'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True', 'CLOUDSDK_ROOT_DIR': fake_cloudsdk_dir, }): mock_log_handler = self.RunCommand('hmac', args=[ 'update', '-e', etag, '-p', project, '-s', 'INACTIVE', access_id ], return_log_handler=True) info_lines = '\n'.join(mock_log_handler.messages['info']) self.assertIn(('Gcloud Storage Command: {} storage' ' hmac update {} --etag {} --project {} --{} {}').format( shim_util._get_gcloud_binary_path('fake_dir'), hmac._DESCRIBE_COMMAND_FORMAT, etag, project, 'deactivate', access_id), info_lines)