KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_bucket_policy method in localstack

Best Python code snippet using localstack_python

test_get_bucket_policy.py

Source:test_get_bucket_policy.py

...31        client_s3.create_bucket(**info_bucket)32        client_s3.put_bucket_policy(Bucket=bucket_name,33                                    Policy=expected_bucket_policy)34        # call function test35        actual_bucket_policy = S3Utils.get_bucket_policy(36            trace_id, client_s3, bucket_name, aws_account, region_name)37        # check result38        self.assertEqual(expected_bucket_policy,39                         actual_bucket_policy['Policy'])40    def test_get_bucket_policy_error_access_denied(self):41        # create mock throw error when called function get_bucket_policy42        expected_error_response = copy.deepcopy(DataCommon.ERROR_RESPONSE)43        expected_operation_name = copy.deepcopy(DataCommon.OPERATION_NAME)44        expected_error_response['Error']['Code'] = 'AccessDenied'45        with patch.object(client_s3, 'get_bucket_policy') as mock_method:46            mock_method.side_effect = ClientError(47                error_response=expected_error_response,48                operation_name=expected_operation_name)49            with patch.object(PmLogAdapter, 'warning',50                              return_value=None) as mock_method_warning:51                with self.assertRaises(PmError) as exception:52                    # call function test53                    S3Utils.get_bucket_policy(trace_id, client_s3, bucket_name,54                                              aws_account, region_name)55        # check error56        actual_cause_error = exception.exception.cause_error57        self.assertEqual(expected_error_response['Error'],58                         actual_cause_error.response['Error'])59        self.assertEqual(expected_operation_name,60                         actual_cause_error.operation_name)61        # check write log warning62        mock_method_warning.assert_any_call(63            '[%s/%s] æ¨©éã¨ã©ã¼ã«ããS3ãã±ããããªã·ã¼æå ±ã®åå¾ã«å¤±æãã¾ãããï¼%sï¼', aws_account,64            region_name, bucket_name)65    def test_get_bucket_policy_error_method_not_allowed(self):66        # create mock throw error when called function get_bucket_policy67        expected_error_response = copy.deepcopy(DataCommon.ERROR_RESPONSE)68        expected_operation_name = copy.deepcopy(DataCommon.OPERATION_NAME)69        expected_error_response['Error']['Code'] = 'MethodNotAllowed'70        with patch.object(client_s3, 'get_bucket_policy') as mock_method:71            mock_method.side_effect = ClientError(72                error_response=expected_error_response,73                operation_name=expected_operation_name)74            with patch.object(PmLogAdapter, 'warning',75                              return_value=None) as mock_method_warning:76                with self.assertRaises(PmError) as exception:77                    # call function test78                    S3Utils.get_bucket_policy(trace_id, client_s3, bucket_name,79                                              aws_account, region_name)80        # check error81        actual_cause_error = exception.exception.cause_error82        self.assertEqual(expected_error_response['Error'],83                         actual_cause_error.response['Error'])84        self.assertEqual(expected_operation_name,85                         actual_cause_error.operation_name)86        # check write log warning87        mock_method_warning.assert_any_call(88            '[%s/%s] æ¨©éã¨ã©ã¼ã«ããS3ãã±ããããªã·ã¼æå ±ã®åå¾ã«å¤±æãã¾ãããï¼%sï¼', aws_account,89            region_name, bucket_name)90    def test_get_bucket_policy_error_no_such_bucket_policy(self):91        # create mock throw error when called function get_bucket_policy92        expected_error_response = copy.deepcopy(DataCommon.ERROR_RESPONSE)93        expected_operation_name = copy.deepcopy(DataCommon.OPERATION_NAME)94        expected_error_response['Error']['Code'] = 'NoSuchBucketPolicy'95        with patch.object(client_s3, 'get_bucket_policy') as mock_method:96            mock_method.side_effect = ClientError(97                error_response=expected_error_response,98                operation_name=expected_operation_name)99            with patch.object(PmLogAdapter, 'info',100                              return_value=None) as mock_method_info:101                # call function test102                result_bucket_policy = S3Utils.get_bucket_policy(103                    trace_id, client_s3, bucket_name, aws_account, region_name)104        self.assertEqual(result_bucket_policy, None)105        # check write log info106        mock_method_info.assert_any_call('[%s/%s]S3ãã±ããããªã·ã¼ã¯æªè¨å®ã§ããï¼%sï¼',107                                         aws_account, region_name, bucket_name)108    def test_get_bucket_policy_error_other(self):109        # create mock throw error when called function get_bucket_policy110        expected_error_response = copy.deepcopy(DataCommon.ERROR_RESPONSE)111        expected_operation_name = copy.deepcopy(DataCommon.OPERATION_NAME)112        with patch.object(client_s3, 'get_bucket_policy') as mock_method:113            mock_method.side_effect = ClientError(114                error_response=expected_error_response,115                operation_name=expected_operation_name)116            with patch.object(PmLogAdapter, 'error',117                              return_value=None) as mock_method_error:118                with self.assertRaises(PmError) as exception:119                    # call function test120                    S3Utils.get_bucket_policy(trace_id, client_s3, bucket_name,121                                              aws_account, region_name)122        # check error123        actual_cause_error = exception.exception.cause_error124        self.assertEqual(expected_error_response['Error'],125                         actual_cause_error.response['Error'])126        self.assertEqual(expected_operation_name,127                         actual_cause_error.operation_name)128        # check write log error129        mock_method_error.assert_any_call('[%s/%s]S3ãã±ããããªã·ã¼æå ±ã®åå¾ã«å¤±æãã¾ãããï¼%sï¼',130                                          aws_account, region_name,...

get_bucket_policy.py

Source:get_bucket_policy.py

...8from botocore.exceptions import ClientError9class CONNECTION(object):10    def __init__(self, url=None):11        self.client = boto3.client('s3', endpoint_url=url)12    def get_bucket_policy(self, bucket_name=None):13        try:14            response = self.client.get_bucket_policy(Bucket=bucket_name)15            print(response['Policy'])16        except ClientError as e:17            logging.error(e)18            return False19        return response['Policy']20if __name__ == '__main__':21    url = "http://172.16.68.100:7480"22    # url = "http://10.255.20.121:7480"23    conn = CONNECTION(url)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.