KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use describe_time_to_live method in localstack

Best Python code snippet using localstack_python
test_dynamodb.py
Source:test_dynamodb.py
1# -*- coding: utf-8 -*-2from __future__ import absolute_import, unicode_literals3from decimal import Decimal4import pytest5from case import MagicMock, Mock, patch, sentinel, skip6from celery import states7from celery.backends import dynamodb as module8from celery.backends.dynamodb import DynamoDBBackend9from celery.exceptions import ImproperlyConfigured10from celery.five import string11@skip.unless_module('boto3')12class test_DynamoDBBackend:13    def setup(self):14        self._static_timestamp = Decimal(1483425566.52)  # noqa15        self.app.conf.result_backend = 'dynamodb://'16    @property17    def backend(self):18        """:rtype: DynamoDBBackend"""19        return self.app.backend20    def test_init_no_boto3(self):21        prev, module.boto3 = module.boto3, None22        try:23            with pytest.raises(ImproperlyConfigured):24                DynamoDBBackend(app=self.app)25        finally:26            module.boto3 = prev27    def test_init_aws_credentials(self):28        with pytest.raises(ImproperlyConfigured):29            DynamoDBBackend(30                app=self.app,31                url='dynamodb://a:@'32            )33    def test_init_invalid_ttl_seconds_raises(self):34        with pytest.raises(ValueError):35            DynamoDBBackend(36                app=self.app,37                url='dynamodb://@?ttl_seconds=1d'38            )39    def test_get_client_explicit_endpoint(self):40        table_creation_path = \41            'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'42        with patch('boto3.client') as mock_boto_client, \43                patch(table_creation_path):44            self.app.conf.dynamodb_endpoint_url = 'http://my.domain.com:666'45            backend = DynamoDBBackend(46                app=self.app,47                url='dynamodb://@us-east-1'48            )49            client = backend._get_client()50            assert backend.client is client51            mock_boto_client.assert_called_once_with(52                'dynamodb',53                endpoint_url='http://my.domain.com:666',54                region_name='us-east-1'55            )56            assert backend.endpoint_url == 'http://my.domain.com:666'57    def test_get_client_local(self):58        table_creation_path = \59            'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'60        with patch('boto3.client') as mock_boto_client, \61                patch(table_creation_path):62            backend = DynamoDBBackend(63                app=self.app,64                url='dynamodb://@localhost:8000'65            )66            client = backend._get_client()67            assert backend.client is client68            mock_boto_client.assert_called_once_with(69                'dynamodb',70                endpoint_url='http://localhost:8000',71                region_name='us-east-1'72            )73            assert backend.endpoint_url == 'http://localhost:8000'74    def test_get_client_credentials(self):75        table_creation_path = \76            'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'77        with patch('boto3.client') as mock_boto_client, \78                patch(table_creation_path):79            backend = DynamoDBBackend(80                app=self.app,81                url='dynamodb://key:secret@test'82            )83            client = backend._get_client()84            assert client is backend.client85            mock_boto_client.assert_called_once_with(86                'dynamodb',87                aws_access_key_id='key',88                aws_secret_access_key='secret',89                region_name='test'90            )91            assert backend.aws_region == 'test'92    @patch('boto3.client')93    @patch('celery.backends.dynamodb.DynamoDBBackend._get_or_create_table')94    @patch('celery.backends.dynamodb.DynamoDBBackend._validate_ttl_methods')95    @patch('celery.backends.dynamodb.DynamoDBBackend._set_table_ttl')96    def test_get_client_time_to_live_called(97            self,98            mock_set_table_ttl,99            mock_validate_ttl_methods,100            mock_get_or_create_table,101            mock_boto_client,102    ):103        backend = DynamoDBBackend(104            app=self.app,105            url='dynamodb://key:secret@test?ttl_seconds=30'106        )107        backend._get_client()108        mock_validate_ttl_methods.assert_called_once()109        mock_set_table_ttl.assert_called_once()110    def test_get_or_create_table_not_exists(self):111        self.backend._client = MagicMock()112        mock_create_table = self.backend._client.create_table = MagicMock()113        mock_describe_table = self.backend._client.describe_table = \114            MagicMock()115        mock_describe_table.return_value = {116            'Table': {117                'TableStatus': 'ACTIVE'118            }119        }120        self.backend._get_or_create_table()121        mock_create_table.assert_called_once_with(122            **self.backend._get_table_schema()123        )124    def test_get_or_create_table_already_exists(self):125        from botocore.exceptions import ClientError126        self.backend._client = MagicMock()127        mock_create_table = self.backend._client.create_table = MagicMock()128        client_error = ClientError(129            {130                'Error': {131                    'Code': 'ResourceInUseException',132                    'Message': 'Table already exists: {}'.format(133                        self.backend.table_name134                    )135                }136            },137            'CreateTable'138        )139        mock_create_table.side_effect = client_error140        mock_describe_table = self.backend._client.describe_table = \141            MagicMock()142        mock_describe_table.return_value = {143            'Table': {144                'TableStatus': 'ACTIVE'145            }146        }147        self.backend._get_or_create_table()148        mock_describe_table.assert_called_once_with(149            TableName=self.backend.table_name150        )151    def test_wait_for_table_status(self):152        self.backend._client = MagicMock()153        mock_describe_table = self.backend._client.describe_table = \154            MagicMock()155        mock_describe_table.side_effect = [156            {'Table': {157                'TableStatus': 'CREATING'158            }},159            {'Table': {160                'TableStatus': 'SOME_STATE'161            }}162        ]163        self.backend._wait_for_table_status(expected='SOME_STATE')164        assert mock_describe_table.call_count == 2165    def test_has_ttl_none_returns_none(self):166        self.backend.time_to_live_seconds = None167        assert self.backend._has_ttl() is None168    def test_has_ttl_lt_zero_returns_false(self):169        self.backend.time_to_live_seconds = -1170        assert self.backend._has_ttl() is False171    def test_has_ttl_gte_zero_returns_true(self):172        self.backend.time_to_live_seconds = 30173        assert self.backend._has_ttl() is True174    def test_validate_ttl_methods_present_returns_none(self):175        self.backend._client = MagicMock()176        assert self.backend._validate_ttl_methods() is None177    def test_validate_ttl_methods_missing_raise(self):178        self.backend._client = MagicMock()179        delattr(self.backend._client, 'describe_time_to_live')180        delattr(self.backend._client, 'update_time_to_live')181        with pytest.raises(AttributeError):182            self.backend._validate_ttl_methods()183        with pytest.raises(AttributeError):184            self.backend._validate_ttl_methods()185    def test_set_table_ttl_describe_time_to_live_fails_raises(self):186        from botocore.exceptions import ClientError187        self.backend.time_to_live_seconds = -1188        self.backend._client = MagicMock()189        mock_describe_time_to_live = \190            self.backend._client.describe_time_to_live = MagicMock()191        client_error = ClientError(192            {193                'Error': {194                    'Code': 'Foo',195                    'Message': 'Bar',196                }197            },198            'DescribeTimeToLive'199        )200        mock_describe_time_to_live.side_effect = client_error201        with pytest.raises(ClientError):202            self.backend._set_table_ttl()203    def test_set_table_ttl_enable_when_disabled_succeeds(self):204        self.backend.time_to_live_seconds = 30205        self.backend._client = MagicMock()206        mock_update_time_to_live = self.backend._client.update_time_to_live = \207            MagicMock()208        mock_describe_time_to_live = \209            self.backend._client.describe_time_to_live = MagicMock()210        mock_describe_time_to_live.return_value = {211            'TimeToLiveDescription': {212                'TimeToLiveStatus': 'DISABLED',213                'AttributeName': self.backend._ttl_field.name214            }215        }216        self.backend._set_table_ttl()217        mock_describe_time_to_live.assert_called_once_with(218            TableName=self.backend.table_name219        )220        mock_update_time_to_live.assert_called_once()221    def test_set_table_ttl_enable_when_enabled_with_correct_attr_succeeds(self):222        self.backend.time_to_live_seconds = 30223        self.backend._client = MagicMock()224        self.backend._client.update_time_to_live = MagicMock()225        mock_describe_time_to_live = \226            self.backend._client.describe_time_to_live = MagicMock()227        mock_describe_time_to_live.return_value = {228            'TimeToLiveDescription': {229                'TimeToLiveStatus': 'ENABLED',230                'AttributeName': self.backend._ttl_field.name231            }232        }233        self.backend._set_table_ttl()234        mock_describe_time_to_live.assert_called_once_with(235            TableName=self.backend.table_name236        )237    def test_set_table_ttl_enable_when_currently_disabling_raises(self):238        from botocore.exceptions import ClientError239        self.backend.time_to_live_seconds = 30240        self.backend._client = MagicMock()241        mock_update_time_to_live = self.backend._client.update_time_to_live = \242            MagicMock()243        client_error = ClientError(244            {245                'Error': {246                    'Code': 'ValidationException',247                    'Message': (248                        'Time to live has been modified multiple times '249                        'within a fixed interval'250                    )251                }252            },253            'UpdateTimeToLive'254        )255        mock_update_time_to_live.side_effect = client_error256        mock_describe_time_to_live = \257            self.backend._client.describe_time_to_live = MagicMock()258        mock_describe_time_to_live.return_value = {259            'TimeToLiveDescription': {260                'TimeToLiveStatus': 'DISABLING',261                'AttributeName': self.backend._ttl_field.name262            }263        }264        with pytest.raises(ClientError):265            self.backend._set_table_ttl()266    def test_set_table_ttl_enable_when_enabled_with_wrong_attr_raises(self):267        from botocore.exceptions import ClientError268        self.backend.time_to_live_seconds = 30269        self.backend._client = MagicMock()270        mock_update_time_to_live = self.backend._client.update_time_to_live = \271            MagicMock()272        wrong_attr_name = self.backend._ttl_field.name + 'x'273        client_error = ClientError(274            {275                'Error': {276                    'Code': 'ValidationException',277                    'Message': (278                        'TimeToLive is active on a different AttributeName: '279                        'current AttributeName is {}'280                    ).format(wrong_attr_name)281                }282            },283            'UpdateTimeToLive'284        )285        mock_update_time_to_live.side_effect = client_error286        mock_describe_time_to_live = \287            self.backend._client.describe_time_to_live = MagicMock()288        mock_describe_time_to_live.return_value = {289            'TimeToLiveDescription': {290                'TimeToLiveStatus': 'ENABLED',291                'AttributeName': self.backend._ttl_field.name + 'x'292            }293        }294        with pytest.raises(ClientError):295            self.backend._set_table_ttl()296    def test_set_table_ttl_disable_when_disabled_succeeds(self):297        self.backend.time_to_live_seconds = -1298        self.backend._client = MagicMock()299        self.backend._client.update_time_to_live = MagicMock()300        mock_describe_time_to_live = \301            self.backend._client.describe_time_to_live = MagicMock()302        mock_describe_time_to_live.return_value = {303            'TimeToLiveDescription': {304                'TimeToLiveStatus': 'DISABLED'305            }306        }307        self.backend._set_table_ttl()308        mock_describe_time_to_live.assert_called_once_with(309            TableName=self.backend.table_name310        )311    def test_set_table_ttl_disable_when_currently_enabling_raises(self):312        from botocore.exceptions import ClientError313        self.backend.time_to_live_seconds = -1314        self.backend._client = MagicMock()315        mock_update_time_to_live = self.backend._client.update_time_to_live = \316            MagicMock()317        client_error = ClientError(318            {319                'Error': {320                    'Code': 'ValidationException',321                    'Message': (322                        'Time to live has been modified multiple times '323                        'within a fixed interval'324                    )325                }326            },327            'UpdateTimeToLive'328        )329        mock_update_time_to_live.side_effect = client_error330        mock_describe_time_to_live = \331            self.backend._client.describe_time_to_live = MagicMock()332        mock_describe_time_to_live.return_value = {333            'TimeToLiveDescription': {334                'TimeToLiveStatus': 'ENABLING',335                'AttributeName': self.backend._ttl_field.name336            }337        }338        with pytest.raises(ClientError):339            self.backend._set_table_ttl()340    def test_prepare_get_request(self):341        expected = {342            'TableName': u'celery',343            'Key': {u'id': {u'S': u'abcdef'}}344        }345        assert self.backend._prepare_get_request('abcdef') == expected346    def test_prepare_put_request(self):347        expected = {348            'TableName': u'celery',349            'Item': {350                u'id': {u'S': u'abcdef'},351                u'result': {u'B': u'val'},352                u'timestamp': {353                    u'N': str(Decimal(self._static_timestamp))354                }355            }356        }357        with patch('celery.backends.dynamodb.time', self._mock_time):358            result = self.backend._prepare_put_request('abcdef', 'val')359        assert result == expected360    def test_prepare_put_request_with_ttl(self):361        ttl = self.backend.time_to_live_seconds = 30362        expected = {363            'TableName': u'celery',364            'Item': {365                u'id': {u'S': u'abcdef'},366                u'result': {u'B': u'val'},367                u'timestamp': {368                    u'N': str(Decimal(self._static_timestamp))369                },370                u'ttl': {371                    u'N': str(int(self._static_timestamp + ttl))372                }373            }374        }375        with patch('celery.backends.dynamodb.time', self._mock_time):376            result = self.backend._prepare_put_request('abcdef', 'val')377        assert result == expected378    def test_item_to_dict(self):379        boto_response = {380            'Item': {381                'id': {382                    'S': sentinel.key383                },384                'result': {385                    'B': sentinel.value386                },387                'timestamp': {388                    'N': Decimal(1)389                }390            }391        }392        converted = self.backend._item_to_dict(boto_response)393        assert converted == {394            'id': sentinel.key,395            'result': sentinel.value,396            'timestamp': Decimal(1)397        }398    def test_get(self):399        self.backend._client = Mock(name='_client')400        self.backend._client.get_item = MagicMock()401        assert self.backend.get('1f3fab') is None402        self.backend.client.get_item.assert_called_once_with(403            Key={u'id': {u'S': u'1f3fab'}},404            TableName='celery'405        )406    def _mock_time(self):407        return self._static_timestamp408    def test_set(self):409        self.backend._client = MagicMock()410        self.backend._client.put_item = MagicMock()411        # should return None412        with patch('celery.backends.dynamodb.time', self._mock_time):413            assert self.backend._set_with_state(sentinel.key, sentinel.value, states.SUCCESS) is None414        assert self.backend._client.put_item.call_count == 1415        _, call_kwargs = self.backend._client.put_item.call_args416        expected_kwargs = {417            'Item': {418                u'timestamp': {u'N': str(self._static_timestamp)},419                u'id': {u'S': string(sentinel.key)},420                u'result': {u'B': sentinel.value}421            },422            'TableName': 'celery'423        }424        assert call_kwargs['Item'] == expected_kwargs['Item']425        assert call_kwargs['TableName'] == 'celery'426    def test_set_with_ttl(self):427        ttl = self.backend.time_to_live_seconds = 30428        self.backend._client = MagicMock()429        self.backend._client.put_item = MagicMock()430        # should return None431        with patch('celery.backends.dynamodb.time', self._mock_time):432            assert self.backend._set_with_state(sentinel.key, sentinel.value, states.SUCCESS) is None433        assert self.backend._client.put_item.call_count == 1434        _, call_kwargs = self.backend._client.put_item.call_args435        expected_kwargs = {436            'Item': {437                u'timestamp': {u'N': str(self._static_timestamp)},438                u'id': {u'S': string(sentinel.key)},439                u'result': {u'B': sentinel.value},440                u'ttl': {u'N': str(int(self._static_timestamp + ttl))},441            },442            'TableName': 'celery'443        }444        assert call_kwargs['Item'] == expected_kwargs['Item']445        assert call_kwargs['TableName'] == 'celery'446    def test_delete(self):447        self.backend._client = Mock(name='_client')448        mocked_delete = self.backend._client.delete = Mock('client.delete')449        mocked_delete.return_value = None450        # should return None451        assert self.backend.delete('1f3fab') is None452        self.backend.client.delete_item.assert_called_once_with(453            Key={u'id': {u'S': u'1f3fab'}},454            TableName='celery'455        )456    def test_backend_by_url(self, url='dynamodb://'):457        from celery.app import backends458        from celery.backends.dynamodb import DynamoDBBackend459        backend, url_ = backends.by_url(url, self.app.loader)460        assert backend is DynamoDBBackend461        assert url_ == url462    def test_backend_params_by_url(self):463        self.app.conf.result_backend = (464            'dynamodb://@us-east-1/celery_results'465            '?read=10'466            '&write=20'467            '&ttl_seconds=600'468        )469        assert self.backend.aws_region == 'us-east-1'470        assert self.backend.table_name == 'celery_results'471        assert self.backend.read_capacity_units == 10472        assert self.backend.write_capacity_units == 20473        assert self.backend.time_to_live_seconds == 600...
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.