Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_bucket method in localstack

Best Python code snippet using localstack_python

test_flags.py

Source:test_flags.py

...33        bucket_patch = patch('experiments.flags.stable_bucketing_hash_group', return_value=1)34        self.addCleanup(bucket_patch.stop)35        bucket_patch.start()36        self.addCleanup(RequestCache.clear_all_namespaces)37    def get_bucket(self, track=False, active=True):38        # Does not use ExperimentWaffleFlag.override, since that shortcuts get_bucket and we want to test internals39        with override_waffle_flag(self.flag, active):40            with override_waffle_flag(self.flag.bucket_flags[1], True):41                return self.flag.get_bucket(course_key=self.key, track=track)42    def test_basic_happy_path(self):43        self.assertEqual(self.get_bucket(), 1)44    def test_no_request(self):45        set_current_request(None)46        self.assertEqual(self.get_bucket(), 0)47    def test_not_enabled(self):48        self.assertEqual(self.get_bucket(active=False), 0)49    @ddt.data(50        ('2012-01-06', None, 1),  # no enrollment, but start is in past (we allow normal bucketing in this case)51        ('9999-01-06', None, 0),  # no enrollment, but start is in future (we give bucket 0 in that case)52        ('2012-01-06', '2012-01-05', 0),  # enrolled before experiment start53        ('2012-01-06', '2012-01-07', 1),  # enrolled after experiment start54        (None, '2012-01-07', 1),  # no experiment date55        ('not-a-date', '2012-01-07', 0),  # bad experiment date56    )57    @ddt.unpack58    def test_enrollment_start(self, experiment_start, enrollment_created, expected_bucket):59        if enrollment_created:60            enrollment = CourseEnrollmentFactory(user=self.user, course_id='a/b/c')61            enrollment.created = parser.parse(enrollment_created).replace(tzinfo=pytz.UTC)62            enrollment.save()63        if experiment_start:64            ExperimentKeyValueFactory(experiment_id=0, key='enrollment_start', value=experiment_start)65        self.assertEqual(self.get_bucket(), expected_bucket)66    @ddt.data(67        ('2012-01-06', None, 0),  # no enrollment, but end is in past (we give bucket 0 in that case)68        ('9999-01-06', None, 1),  # no enrollment, but end is in future (we allow normal bucketing in this case)69        ('2012-01-06', '2012-01-05', 1),  # enrolled before experiment end70        ('2012-01-06', '2012-01-07', 0),  # enrolled after experiment end71        (None, '2012-01-07', 1),  # no experiment date72        ('not-a-date', '2012-01-07', 0),  # bad experiment date73    )74    @ddt.unpack75    def test_enrollment_end(self, experiment_end, enrollment_created, expected_bucket):76        if enrollment_created:77            enrollment = CourseEnrollmentFactory(user=self.user, course_id='a/b/c')78            enrollment.created = parser.parse(enrollment_created).replace(tzinfo=pytz.UTC)79            enrollment.save()80        if experiment_end:81            ExperimentKeyValueFactory(experiment_id=0, key='enrollment_end', value=experiment_end)82        self.assertEqual(self.get_bucket(), expected_bucket)83    @ddt.data(84        (True, 0),85        (False, 1),86    )87    @ddt.unpack88    def test_forcing_bucket(self, active, expected_bucket):89        bucket_flag = CourseWaffleFlag('experiments', 'test.0')90        with bucket_flag.override(active=active):91            self.assertEqual(self.get_bucket(), expected_bucket)92    def test_tracking(self):93        # Run twice, with same request94        with patch('experiments.flags.segment') as segment_mock:95            self.assertEqual(self.get_bucket(track=True), 1)96            RequestCache.clear_all_namespaces()  # we want to force get_bucket to check session, not early exit97            self.assertEqual(self.get_bucket(track=True), 1)98        # Now test that we only sent the signal once, and with the correct properties99        self.assertEqual(segment_mock.track.call_count, 1)100        self.assertEqual(segment_mock.track.call_args, ((), {101            'user_id': self.user.id,102            'event_name': 'edx.bi.experiment.user.bucketed',103            'properties': {104                'site': self.request.site.domain,105                'app_label': 'experiments',106                'experiment': 'test',107                'bucket': 1,108                'course_id': 'a/b/c',109                'is_staff': self.user.is_staff,110                'nonInteraction': 1,111            },112        }))113    def test_caching(self):114        self.assertEqual(self.get_bucket(active=True), 1)115        self.assertEqual(self.get_bucket(active=False), 1)  # still returns 1!116    def test_is_enabled(self):117        with patch('experiments.flags.ExperimentWaffleFlag.get_bucket', return_value=1):118            self.assertEqual(self.flag.is_enabled_without_course_context(), True)119            self.assertEqual(self.flag.is_enabled(self.key), True)120            self.assertEqual(self.flag.is_enabled(), True)121        with patch('experiments.flags.ExperimentWaffleFlag.get_bucket', return_value=0):122            self.assertEqual(self.flag.is_enabled_without_course_context(), False)123            self.assertEqual(self.flag.is_enabled(self.key), False)124            self.assertEqual(self.flag.is_enabled(), False)125    @ddt.data(126        (True, 1, 1),127        (True, 0, 0),128        (False, 1, 0),  # bucket is always 0 if the experiment is off129        (False, 0, 0),130    )131    @ddt.unpack132    # Test the override method133    def test_override_method(self, active, bucket_override, expected_bucket):134        with self.flag.override(active=active, bucket=bucket_override):135            self.assertEqual(self.flag.get_bucket(), expected_bucket)...

test_decorator.py

Source:test_decorator.py

...75    a_handle.when.called_with().should.throw(IOError,76        'The file foobar.csv cannot be found on S3.')77@patch("keeper.core.settings")78@patch("keeper.core.boto.connect_s3")79def test_get_bucket(connect_s3, settings):80    connect_s3.return_value.get_bucket.return_value = "foobar"81    get_bucket.when.called_with().should.return_value("foobar")82@patch('keeper.core.open', mock_open(read_data='local contents\nand some more'), create=True)83@patch("keeper.core.get_bucket")84def test_local_result(get_bucket):85    get_bucket.return_value.get_key.return_value = None86    @use_file('foobar.csv')87    def a_handle(keeper_file, *args, **options):88        lines = []89        for line in keeper_file:90            lines.append(line)91        return lines...

gcs_storage_client_helper_test.py

Source:gcs_storage_client_helper_test.py

...6class StorageClientHelperTestCase(TestCase):7    @patch('google.cloud.storage.Client.get_bucket')8    def test_get_bucket_should_return_bucket(self, get_bucket):9        storage_client = StorageClientHelper('test_project')10        bucket = storage_client.get_bucket('my_bucket')11        self.assertIsNotNone(bucket)12        get_bucket.assert_called_once()13    @patch('google.cloud.storage.Client.get_bucket')14    def test_get_bucket_on_exception_should_not_leak_error(self, get_bucket):15        get_bucket.side_effect = exceptions.NotFound('error on retrieving bucket')16        storage_client = StorageClientHelper('test_project')17        bucket = storage_client.get_bucket('my_bucket')18        self.assertIsNone(bucket)19        get_bucket.assert_called_once()20    @patch('google.cloud.storage.Client.list_buckets')21    def test_list_buckets_should_return_buckets(self, list_buckets):22        results_iterator = MockedObject()23        results_iterator.pages = [{}]24        list_buckets.return_value = results_iterator25        storage_client = StorageClientHelper('test_project')26        buckets = storage_client.list_buckets()27        self.assertIsNotNone(buckets)28        list_buckets.assert_called_once()29    @patch('google.cloud.storage.Client.list_blobs')30    def test_list_blobs_should_return_blobs(self, list_blobs):31        results_iterator = MockedObject()...

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.