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

How to use _bucket_url method in localstack

Best Python code snippet using localstack_python

aws_client.py

Source:aws_client.py

...33        log.debug(rsp.text)34        log.debug('\n')35    36    def create_bucket(self, bkt):37        rsp = self._ose_put(self._bucket_url(bkt))38        assert rsp.status_code == 20039        return rsp40    41    def head_bucket(self, bkt):42        rsp = self._ose_head(self._bucket_url(bkt))43        return rsp44    45    def delete_bucket(self, bkt):46        rsp = self._ose_delete(self._bucket_url(bkt))47        assert rsp.status_code == 20448        return rsp49    50    def get_bucket_versioning(self, bkt):51        rsp = self._ose_get(self._bucket_url(bkt), params={'versioning': ''})52        assert rsp.status_code == 20053        return rsp54    55    def enable_bucket_versioning(self, bkt):56        rsp = self._ose_put(self._bucket_url(bkt), params={'versioning': ''},57                            data=self._enable_bucket_versioning_body())58        assert rsp.status_code == 20059        return rsp60    61    def suspend_bucket_versioning(self, bkt):62        rsp = self._ose_put(self._bucket_url(bkt), params={'versioning': ''},63                            data=self._suspend_bucket_versioning_body())64        assert rsp.status_code == 20065        return rsp66    67    def create_object(self, bkt, *objects, data=None):68        rsp = self._ose_put(self._object_url(bkt, *objects), data=data)69        assert rsp.status_code == 20070        return rsp71    72    def get_object(self, bkt, *objects, params=None):73        rsp = self._ose_get(self._object_url(bkt, *objects), params=params)74        return rsp75    76    def head_object(self, bkt, *objects, params=None):77        rsp = self._ose_head(self._object_url(bkt, *objects), params=params)78        return rsp79    80    def delete_object(self, bkt, *objects, params=None):81        rsp = self._ose_delete(self._object_url(bkt, *objects), params=params)82        assert rsp.status_code == 20483        return rsp84    85    def delete_version(self, bkt, obj, version):86        rsp = self._ose_delete(self._object_url(bkt, obj), params={'versionId': version})87        assert rsp.status_code == 20488        return rsp89    90    def delete_all_versions(self, bkt, obj):91        rsp = self.list_versions(bkt)92        assert rsp.status_code == 20093        94        for version in rsp.json()['versions']:95            rsp = self.delete_version(bkt, obj, version['versionId'])96            assert rsp.status_code == 20497    98    def list_objects(self, bkt, params=None):99        rsp = self._ose_get(self._bucket_url(bkt), params=params)100        assert rsp.status_code == 200101        return rsp102    103    def list_versions(self, bucket, params=None):104        rsp = self._ose_get(self._versions_url(bucket), params=params)105        assert rsp.status_code == 200106        return rsp107    108    def _buckets_url(self):109        return self.aws_s3110    111    def _bucket_url(self, bkt):112        return '{}/{}'.format(self.aws_s3, bkt)113    114    def _object_url(self, bkt, *objects):115        url = '{}/{}/'.format(self.aws_s3, bkt)116        for obj in objects:117            url += obj118        return url119    120    def _versions_url(self, bkt):121        return '{}/{}?versions'.format(self.aws_s3, bkt)122    123    def _ose_get(self, url, params=None):124        pprint('REQUEST GET =====> {} {}'.format(url, params))125        rsp = requests.get(url, params=params, auth=self.auth, verify=False)...

delete_notification_configuration_task.py

Source:delete_notification_configuration_task.py

1# -*- coding: utf-8 -*- #2# Copyright 2022 Google LLC. All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#    http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Task for deleting a notification configuration."""16from __future__ import absolute_import17from __future__ import division18from __future__ import unicode_literals19from googlecloudsdk.api_lib.storage import api_factory20from googlecloudsdk.command_lib.storage import progress_callbacks21from googlecloudsdk.command_lib.storage.tasks import task22class DeleteNotificationConfigurationTask(task.Task):23  """Deletes a notification configuration."""24  def __init__(self, bucket_url, notification_id):25    """Initializes task.26    Args:27      bucket_url (storage_url.CloudUrl): URL of bucket that notification28        configuration exists on.29      notification_id (str): Name of the notification configuration (integer as30        string).31    """32    super(__class__, self).__init__()33    self._bucket_url = bucket_url34    self._notification_id = notification_id35    self.parallel_processing_key = bucket_url.url_string + '|' + notification_id36  def execute(self, task_status_queue=None):37    provider = self._bucket_url.scheme38    api_factory.get_api(provider).delete_notification_configuration(39        self._bucket_url, self._notification_id)40    if task_status_queue:41      progress_callbacks.increment_count_callback(task_status_queue)42  def __eq__(self, other):43    if not isinstance(other, DeleteNotificationConfigurationTask):44      return NotImplemented...

data_fetch.py

Source:data_fetch.py

1import logging2import os3from pathlib import Path4from typing import Dict5from urllib.request import HTTPError, urlretrieve6import pandas as pd7from dashbat.data.data_types import DatasetName8_BUCKET_URL = (9    "http://opendata.auth-6f31f706db6f4a24b55f42a6a79c5086.storage.sbg.cloud.ovh.net"10)11_DATA_FOLDER = Path(__file__).parent.parent.parent / "data"12URLS: Dict[DatasetName, str] = {13    "transition": f"{_BUCKET_URL}/2019-04-08/QUESTIONNAIRE_LA_TRANSITION_ECOLOGIQUE.csv",14    "fiscalite": f"{_BUCKET_URL}/2019-04-08/QUESTIONNAIRE_LA_FISCALITE_ET_LES_DEPENSES_PUBLIQUES.csv",15    "democratie": f"{_BUCKET_URL}/2019-04-08/QUESTIONNAIRE_DEMOCRATIE_ET_CITOYENNETE.csv",16    "organisation": f"{_BUCKET_URL}/2019-04-08/QUESTIONNAIRE_ORGANISATION_DE_LETAT_ET_DES_SERVICES_PUBLICS.csv",17}18def _download_file_if_doesnt_exist(source: str, destination: str) -> None:19    if os.path.exists(destination):20        return21    logging.info(f"Downloading {source} -> {destination}")22    try:23        urlretrieve(source, destination)24    except HTTPError as exc:25        print(source, exc)26def _local_filename(dataset: DatasetName) -> str:27    return str(_DATA_FOLDER / f"{dataset}.csv")28def _remote_filename(dataset: DatasetName) -> str:29    return URLS[dataset]30def _download_dataset(dataset: DatasetName) -> None:31    source = _remote_filename(dataset)32    destination = _local_filename(dataset)33    _download_file_if_doesnt_exist(source, destination)34def fetch_dataset(dataset: DatasetName) -> pd.DataFrame:35    _download_dataset(dataset)36    nb_rows = int(os.environ["FAST_MODE"]) if os.environ.get("FAST_MODE") else None...

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.