Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use s3_create_bucket method in localstack

Best Python code snippet using localstack_python

test_s3events.py

Source:test_s3events.py

...20        assert len(parsed_lines) == 321        assert parsed_lines[2] == '\n'.join([json_lines[1], json_lines[3]])22    @mock_s323    def test_filter_by_logic_connector(self):24        s3_create_bucket(BUCKET_NAME)25        json_lines = ['{"age": 111, "str_field": "str_value1", "logic_connector": 1}',26                      '{"age": 222, "str_field": "str_value2", "logic_connector": 2}',27                      '{"age": 333, "str_field": "str_value3", "logic_connector": 3}',28                      '{"age": 22,  "str_field": "str_value22", "logic_connector": 2}']29        text_body = '\n'.join(json_lines)30        file_name = 'test_filet.txt'31        key_path = 'path/' + file_name32        save_file(BUCKET_NAME, key_path, text_body)33        s3_dict = {'s3': {'bucket': {'name': BUCKET_NAME},34                          'object': {'key': key_path}}}35        event = {'Records': [s3_dict, ]}36        context = {}37        parsed_lines = filter_by_logic_connector(event, context)38        assert len(parsed_lines) == 339        logic_connector_id = 140        assert len(parsed_lines[logic_connector_id].split('\n')) == 141        logic_connector_id = 242        assert len(parsed_lines[logic_connector_id].split('\n')) == 243        logic_connector_id = 344        assert len(parsed_lines[logic_connector_id].split('\n')) == 145    @mock_s346    def test_get_file_text(self):47        s3_create_bucket(BUCKET_NAME)48        saved_len = save_file(BUCKET_NAME, self.key_path, self.text_body)49        text = s3_get_file_text(BUCKET_NAME, self.key_path)50        assert text == self.text_body51    @mock_s352    def test_save_file(self):53        s3_create_bucket(BUCKET_NAME)54        save_file(BUCKET_NAME, self.key_path, self.text_body)55        text_body = s3_get_file_text(BUCKET_NAME, self.key_path)56        # check that file saved to s357        assert text_body == self.text_body58    def test_get_s3_key(self):59        file_name = 'file2'60        key_path = 'path/' + file_name61        logic_connector_id = 12362        key = get_s3_key(logic_connector_id, key_path)63        assert str(logic_connector_id) in key...

check_output_bucket.py

Source:check_output_bucket.py

1from abc import ABC2from video_streaming import settings3from video_streaming.celery import celery_app4from video_streaming.ffmpeg.constants import TASK_DECORATOR_KWARGS5from .base import BaseStreamingTask6from .mixins import CheckOutputBucketMixin7class CheckOutputBucketTask(8        CheckOutputBucketMixin,9        BaseStreamingTask,10        ABC11        ):12    # rewrite BaseCheckMixin.save_failed13    def save_failed(self, request_id):14        super().save_failed(request_id)15        # stop reason will only be set if there is no reason before.16        # set common reason for the task, it's can be connection error17        # after many retries or etc.18        self.save_job_stop_reason(19            self.stop_reason.FAILED_OUTPUT_BUCKET_CHECKING,20            request_id21        )22@celery_app.task(name="check_output_bucket",23                 base=CheckOutputBucketTask,24                 **TASK_DECORATOR_KWARGS)25def check_output_bucket(self,26                        *args,27                        s3_output_bucket: str = settings.S3_DEFAULT_OUTPUT_BUCKET_NAME,28                        s3_create_bucket: bool = settings.CREATE_OUTPUT_BUCKET,29                        request_id: str = None):30    """check output bucket or create if s3_create_bucket is True31       required parameters:32        - request_id33    """34    self.check_output_bucket_requirements(35        request_id=request_id,36        s3_output_bucket=s3_output_bucket37    )38    if self.is_forced_to_stop(request_id):39        raise self.raise_revoke(request_id)40    self.save_primary_status(41        self.primary_status.CHECKING,42        request_id)43    # check output bucket is exist44    # or create if s3_create_bucket is True45    bucket_details = self.get_output_bucket_details(s3_output_bucket)46    # bucket_details is None for 404 or 403 reason47    if not bucket_details:48        # check the task s3_create_bucket boolean param to create a49        #  output bucket when does not exist.50        if not s3_create_bucket:51            # ignore the task when s3_create_bucket is False/None and52            #  the output bucket does not exist.53            self.save_primary_status(54                self.primary_status.FAILED,55                request_id56            )57            self.save_job_stop_reason(58                self.stop_reason.OUTPUT_BUCKET_ON_S3_IS_404_OR_403,59                request_id60            )61            raise self.raise_ignore(62                message=self.error_messages.OUTPUT_BUCKET_404_OR_403,63                request_kwargs=self.request.kwargs)64        # try to create output bucket. (BucketAlreadyExist is handled)65        self.create_output_bucket(s3_output_bucket)...

create_buckets.py

Source:create_buckets.py

...9    def handle(self, *args, **options):10        """11        Script Execution.12        """13        s3_create_bucket(settings.BUCKET_OBJECTS + settings.S3_SEPARATOR + settings.S3_ID)14        s3_create_bucket(settings.BUCKET_PCAPS + settings.S3_SEPARATOR + settings.S3_ID)...

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.