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

How to use update_time_to_live method in localstack

Best Python code snippet using localstack_python

test_ttl.py

Source:test_ttl.py

...47        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],48        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' } ]) as table:49        client = table.meta.client50        ttl_spec = {'AttributeName': 'expiration', 'Enabled': True}51        response = client.update_time_to_live(TableName=table.name,52            TimeToLiveSpecification=ttl_spec)53        assert 'TimeToLiveSpecification' in response54        assert response['TimeToLiveSpecification'] == ttl_spec55        # Verify that DescribeTimeToLive can recall this setting:56        response = client.describe_time_to_live(TableName=table.name)57        assert 'TimeToLiveDescription' in response58        assert response['TimeToLiveDescription'] == {59            'TimeToLiveStatus': 'ENABLED', 'AttributeName': 'expiration'}60        # Verify that UpdateTimeToLive cannot enable TTL is it is already61        # enabled. A user is not allowed to change the expiration attribute62        # without disabling TTL first, and it's an error even to try to63        # enable TTL with exactly the same attribute as already enabled.64        # (the error message uses slightly different wording in those two65        # cases)66        with pytest.raises(ClientError, match='ValidationException.*(active|enabled)'):67            client.update_time_to_live(TableName=table.name,68                TimeToLiveSpecification=ttl_spec)69        with pytest.raises(ClientError, match='ValidationException.*(active|enabled)'):70            new_ttl_spec = {'AttributeName': 'new', 'Enabled': True}71            client.update_time_to_live(TableName=table.name,72                TimeToLiveSpecification=new_ttl_spec)73# Test various *wrong* ways of disabling TTL. Although we test here various74# error cases of how to disable TTL incorrectly, we don't actually check in75# this test case the successful disabling case, because DynamoDB refuses to76# disable TTL if it was enabled in the last hour (according to DynamoDB's77# documentation). We have below a much longer test for the successful TTL78# disable case.79def test_ttl_disable_errors(dynamodb):80    with new_test_table(dynamodb,81        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],82        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' } ]) as table:83        client = table.meta.client84        # We can't disable TTL if it's not already enabled.85        with pytest.raises(ClientError, match='ValidationException.*disabled'):86            client.update_time_to_live(TableName=table.name,87                TimeToLiveSpecification={'AttributeName': 'expiration', 'Enabled': False})88        # So enable TTL, before disabling it:89        client.update_time_to_live(TableName=table.name,90            TimeToLiveSpecification={'AttributeName': 'expiration', 'Enabled': True})91        response = client.describe_time_to_live(TableName=table.name)92        assert response['TimeToLiveDescription'] == {93            'TimeToLiveStatus': 'ENABLED', 'AttributeName': 'expiration'}94        # To disable TTL, the user must specify the current expiration95        # attribute in the command - you can't not specify it, or specify96        # the wrong one!97        with pytest.raises(ClientError, match='ValidationException.*different'):98            client.update_time_to_live(TableName=table.name,99                TimeToLiveSpecification={'AttributeName': 'dog', 'Enabled': False})100        # Finally disable TTL the right way :-) On DynamoDB this fails101        # because you are not allowed to modify the TTL setting twice in102        # one hour, but in our implementation it can also pass quickly.103        with passes_or_raises(ClientError, match='ValidationException.*multiple times'):104            client.update_time_to_live(TableName=table.name,105                TimeToLiveSpecification={'AttributeName': 'expiration', 'Enabled': False})106# Test *sucessful* disabling of TTL. This is an extremely slow test, as107# DynamoDB refuses to disable TTL if it was enabled in the last half hour108# (the documentation suggests a full hour, but in practice it seems 30109# minutes).110@pytest.mark.veryslow111def test_ttl_disable(dynamodb):112    with new_test_table(dynamodb,113        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],114        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' } ]) as table:115        client = table.meta.client116        client.update_time_to_live(TableName=table.name,117            TimeToLiveSpecification={'AttributeName': 'expiration', 'Enabled': True})118        assert client.describe_time_to_live(TableName=table.name)['TimeToLiveDescription'] == {119            'TimeToLiveStatus': 'ENABLED', 'AttributeName': 'expiration'}120        start_time = time.time()121        while time.time() < start_time + 4000:122            try:123                client.update_time_to_live(TableName=table.name,124                    TimeToLiveSpecification={'AttributeName': 'expiration', 'Enabled': False})125                break126            except ClientError as error:127                # As long as we get "Time to live has been modified multiple128                # times within a fixed interval", we need to try again -129                # the documentation says for a full hour!130                assert 'times' in error.response['Error']['Message']131                print(time.time() - start_time)132                time.sleep(60)133                continue134        assert client.describe_time_to_live(TableName=table.name)['TimeToLiveDescription'] == {135            'TimeToLiveStatus': 'DISABLED'}136# Test various errors in the UpdateTimeToLive request.137def test_update_ttl_errors(dynamodb):138    client = dynamodb.meta.client139    # Can't set TTL on a non-existent table140    nonexistent_table = unique_table_name()141    with pytest.raises(ClientError, match='ResourceNotFoundException'):142        client.update_time_to_live(TableName=nonexistent_table,143            TimeToLiveSpecification={'AttributeName': 'expiration', 'Enabled': True})144    with new_test_table(dynamodb,145            KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],146            AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' } ]) as table:147        # AttributeName must be between 1 and 255 characters long.148        with pytest.raises(ClientError, match='ValidationException.*length'):149            client.update_time_to_live(TableName=table.name,150                TimeToLiveSpecification={'AttributeName': 'x'*256, 'Enabled': True})151        with pytest.raises(ClientError, match='ValidationException.*length'):152            client.update_time_to_live(TableName=table.name,153                TimeToLiveSpecification={'AttributeName': '', 'Enabled': True})154        # Missing mandatory UpdateTimeToLive parameters - AttributeName or Enabled155        with pytest.raises(ClientError, match='ValidationException.*[aA]ttributeName'):156            client.update_time_to_live(TableName=table.name,157                TimeToLiveSpecification={'Enabled': True})158        with pytest.raises(ClientError, match='ValidationException.*[eE]nabled'):159            client.update_time_to_live(TableName=table.name,160                TimeToLiveSpecification={'AttributeName': 'hello'})161        # Wrong types for these mandatory parameters (e.g., string for Enabled)162        # The error type is currently a bit different in Alternator163        # (ValidationException) and in DynamoDB (SerializationException).164        with pytest.raises(ClientError):165            client.update_time_to_live(TableName=table.name,166                TimeToLiveSpecification={'AttributeName': 'hello', 'Enabled': 'dog'})167        with pytest.raises(ClientError):168            client.update_time_to_live(TableName=table.name,169                TimeToLiveSpecification={'AttributeName': 3, 'Enabled': True})170# Basic test that expiration indeed expires items that should be expired,171# and doesn't expire items which shouldn't be expired.172# On AWS, this is an extremely slow test - DynamoDB documentation says that173# expiration may even be delayed for 48 hours. But in practice, at the time174# of this writing, for tiny tables we see delays of around 10 minutes.175# The following test is set to always run for "duration" seconds, currently176# 20 minutes on AWS. During this time, we expect to see the items which should177# have expired to be expired - and the items that should not have expired178# should still exist.179# When running against Scylla configured (for testing purposes) to expire180# items with very short delays, "duration" can be set much lower so this181# test will finish in a much more reasonable time.182@pytest.mark.veryslow183def test_ttl_expiration(dynamodb):184    duration = 1200 if is_aws(dynamodb) else 10185    # delta is a quarter of the test duration, but no less than one second,186    # and we use it to schedule some expirations a bit after the test starts,187    # not immediately.188    delta = math.ceil(duration / 4)189    assert delta >= 1190    with new_test_table(dynamodb,191        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],192        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' } ]) as table:193        # Insert one expiring item *before* enabling the TTL, to verify that194        # items that already exist when TTL is configured also get handled.195        p0 = random_string()196        table.put_item(Item={'p': p0, 'expiration': int(time.time())-60})197        # Enable TTL, using the attribute "expiration":198        client = table.meta.client199        ttl_spec = {'AttributeName': 'expiration', 'Enabled': True}200        response = client.update_time_to_live(TableName=table.name,201            TimeToLiveSpecification=ttl_spec)202        assert response['TimeToLiveSpecification'] == ttl_spec203        # This item should never expire, it is missing the "expiration"204        # attribute:205        p1 = random_string()206        table.put_item(Item={'p': p1})207        # This item should expire ASAP, as its "expiration" has already208        # passed, one minute ago:209        p2 = random_string()210        table.put_item(Item={'p': p2, 'expiration': int(time.time())-60})211        # This item has an expiration more than 5 years in the past (it is my212        # birth date...), so according to the DynamoDB documentation it should213        # be ignored and p3 should never expire:214        p3 = random_string()215        table.put_item(Item={'p': p3, 'expiration': 162777600})216        # This item has as its expiration delta into the future, which is a217        # small part of the test duration, so should expire by the time the218        # test ends:219        p4 = random_string()220        table.put_item(Item={'p': p4, 'expiration': int(time.time())+delta})221        # This item starts with expiration time delta into the future,222        # but before it expires we will move it again, so it will never expire:223        p5 = random_string()224        table.put_item(Item={'p': p5, 'expiration': int(time.time())+delta})225        # This item has an expiration time two durations into the future, so it226        # will not expire by the time the test ends:227        p6 = random_string()228        table.put_item(Item={'p': p6, 'expiration': int(time.time()+duration*2)})229        # Like p4, this item has an expiration time delta into the future,230        # here the expiration time is wrongly encoded as a string, not a231        # number, so the item should never expire:232        p7 = random_string()233        table.put_item(Item={'p': p7, 'expiration': str(int(time.time())+delta)})234        # Like p2, p8 and p9 also have an already-passed expiration time,235        # and should expire ASAP. However, whereas p2 had a straighforward236        # integer like 12345678 as the expiration time, p8 and p9 have237        # slightly more elaborate numbers: p8 has 1234567e1 and p9 has238        # 12345678.1234. Those formats should be fine, and this test verifies239        # the TTL code's number parsing doesn't get confused (in our original240        # implementation, it did).241        p8 = random_string()242        with client_no_transform(table.meta.client) as client:243            client.put_item(TableName=table.name, Item={'p': {'S': p8}, 'expiration': {'N': str((int(time.time())-60)//10) + "e1"}})244        # Similarly, floating point expiration time like 12345678.1 should245        # also be fine (note that Python's time.time() returns floating point).246        # This item should also be expired ASAP too.247        p9 = random_string()248        print(Decimal(str(time.time()-60)))249        table.put_item(Item={'p': p9, 'expiration': Decimal(str(time.time()-60))})250        def check_expired():251            # After the delay, p1,p3,p5,p6,p7 should be alive, p0,p2,p4 should not252            return not 'Item' in table.get_item(Key={'p': p0}) \253                and 'Item' in table.get_item(Key={'p': p1}) \254                and not 'Item' in table.get_item(Key={'p': p2}) \255                and 'Item' in table.get_item(Key={'p': p3}) \256                and not 'Item' in table.get_item(Key={'p': p4}) \257                and 'Item' in table.get_item(Key={'p': p5}) \258                and 'Item' in table.get_item(Key={'p': p6}) \259                and 'Item' in table.get_item(Key={'p': p7}) \260                and not 'Item' in table.get_item(Key={'p': p8}) \261                and not 'Item' in table.get_item(Key={'p': p9})262        # We could have just done time.sleep(duration) here, but in case a263        # user is watching this long test, let's output the status every264        # minute, and it also allows us to test what happens when an item265        # p5's expiration time is continuously pushed back into the future:266        start_time = time.time()267        while time.time() < start_time + duration:268            print(f"--- {int(time.time()-start_time)} seconds")269            if 'Item' in table.get_item(Key={'p': p0}):270                print("p0 alive")271            if 'Item' in table.get_item(Key={'p': p1}):272                print("p1 alive")273            if 'Item' in table.get_item(Key={'p': p2}):274                print("p2 alive")275            if 'Item' in table.get_item(Key={'p': p3}):276                print("p3 alive")277            if 'Item' in table.get_item(Key={'p': p4}):278                print("p4 alive")279            if 'Item' in table.get_item(Key={'p': p5}):280                print("p5 alive")281            if 'Item' in table.get_item(Key={'p': p6}):282                print("p6 alive")283            if 'Item' in table.get_item(Key={'p': p7}):284                print("p7 alive")285            if 'Item' in table.get_item(Key={'p': p8}):286                print("p8 alive")287            if 'Item' in table.get_item(Key={'p': p9}):288                print("p9 alive")289            # Always keep p5's expiration delta into the future290            table.update_item(Key={'p': p5},291                AttributeUpdates={'expiration': {'Value': int(time.time())+delta, 'Action': 'PUT'}})292            if check_expired():293                break294            time.sleep(duration/15.0)295        assert check_expired()296# test_ttl_expiration above used a table with just a hash key. Because the297# code to *delete* items in a table which also has a range key is subtly298# different than the code for just a hash key, we want to test the case of299# such a table as well. This test is simpler than test_ttl_expiration because300# all we want to check is that the deletion works - not the expiration time301# parsing and semantics.302@pytest.mark.veryslow303def test_ttl_expiration_with_rangekey(dynamodb):304    # Note that unlike test_ttl_expiration, this test doesn't have a fixed305    # duration - it finishes as soon as the item we expect to be expired306    # is expired.307    max_duration = 1200 if is_aws(dynamodb) else 10308    with new_test_table(dynamodb,309        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' },310                    { 'AttributeName': 'c', 'KeyType': 'RANGE' } ],311        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' },312                               { 'AttributeName': 'c', 'AttributeType': 'S' }]) as table:313        # Enable TTL, using the attribute "expiration":314        client = table.meta.client315        ttl_spec = {'AttributeName': 'expiration', 'Enabled': True}316        response = client.update_time_to_live(TableName=table.name,317            TimeToLiveSpecification=ttl_spec)318        assert response['TimeToLiveSpecification'] == ttl_spec319        # This item should never expire, it is missing the "expiration"320        # attribute:321        p1 = random_string()322        c1 = random_string()323        table.put_item(Item={'p': p1, 'c': c1})324        # This item should expire ASAP, as its "expiration" has already325        # passed, one minute ago:326        p2 = random_string()327        c2 = random_string()328        table.put_item(Item={'p': p2, 'c': c2, 'expiration': int(time.time())-60})329        start_time = time.time()330        while time.time() < start_time + max_duration:331            if ('Item' in table.get_item(Key={'p': p1, 'c': c1})) and not ('Item' in table.get_item(Key={'p': p2, 'c': c2})):332                # p2 expired, p1 didn't. We're done333                break334            time.sleep(max_duration/15.0)335        assert 'Item' in table.get_item(Key={'p': p1, 'c': c1})336        assert not 'Item' in table.get_item(Key={'p': p2, 'c': c2})337# While it probably makes little sense to do this, the designated338# expiration-time attribute *may* be the hash or range key attributes.339# So let's test that this indeed works and items indeed expire340# Just like test_ttl_expiration() above, these tests are extremely slow341# because DynamoDB delays expiration by around 10 minutes.342@pytest.mark.veryslow343def test_ttl_expiration_hash(dynamodb):344    duration = 1200 if is_aws(dynamodb) else 10345    with new_test_table(dynamodb,346        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],347        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'N' } ]) as table:348        ttl_spec = {'AttributeName': 'p', 'Enabled': True}349        table.meta.client.update_time_to_live(TableName=table.name,350            TimeToLiveSpecification=ttl_spec)351        # p1 is in the past, and should be expired. p2 is in the distant352        # future and should not be expired.353        p1 = int(time.time()) - 60354        p2 = int(time.time()) + 3600355        table.put_item(Item={'p': p1})356        table.put_item(Item={'p': p2})357        start_time = time.time()358        def check_expired():359            return not 'Item' in table.get_item(Key={'p': p1}) and 'Item' in table.get_item(Key={'p': p2})360        while time.time() < start_time + duration:361            print(f"--- {int(time.time()-start_time)} seconds")362            if 'Item' in table.get_item(Key={'p': p1}):363                print("p1 alive")364            if 'Item' in table.get_item(Key={'p': p2}):365                print("p2 alive")366            if check_expired():367                break368            time.sleep(duration/15)369        # After the delay, p2 should be alive, p1 should not370        assert check_expired()371@pytest.mark.veryslow372def test_ttl_expiration_range(dynamodb):373    duration = 1200 if is_aws(dynamodb) else 10374    with new_test_table(dynamodb,375        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, { 'AttributeName': 'c', 'KeyType': 'RANGE' } ],376        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' }, { 'AttributeName': 'c', 'AttributeType': 'N' } ]) as table:377        ttl_spec = {'AttributeName': 'c', 'Enabled': True}378        table.meta.client.update_time_to_live(TableName=table.name,379            TimeToLiveSpecification=ttl_spec)380        # c1 is in the past, and should be expired. c2 is in the distant381        # future and should not be expired.382        p = random_string()383        c1 = int(time.time()) - 60384        c2 = int(time.time()) + 3600385        table.put_item(Item={'p': p, 'c': c1})386        table.put_item(Item={'p': p, 'c': c2})387        start_time = time.time()388        def check_expired():389            return not 'Item' in table.get_item(Key={'p': p, 'c': c1}) and 'Item' in table.get_item(Key={'p': p, 'c': c2})390        while time.time() < start_time + duration:391            print(f"--- {int(time.time()-start_time)} seconds")392            if 'Item' in table.get_item(Key={'p': p, 'c': c1}):393                print("c1 alive")394            if 'Item' in table.get_item(Key={'p': p, 'c': c2}):395                print("c2 alive")396            if check_expired():397                break398            time.sleep(duration/15)399        # After the delay, c2 should be alive, c1 should not400        assert check_expired()401# In the above key-attribute tests, the key attribute we used for the402# expiration-time attribute had the expected numeric type. If the key403# attribute has a non-numeric type (e.g., string), it can never contain404# an expiration time, so nothing will ever expire - but although DynamoDB405# knows this, it doesn't refuse this setting anyway - although it could.406# This test demonstrates that:407@pytest.mark.veryslow408def test_ttl_expiration_hash_wrong_type(dynamodb):409    duration = 900 if is_aws(dynamodb) else 3410    with new_test_table(dynamodb,411        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' }, ],412        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'S' } ]) as table:413        ttl_spec = {'AttributeName': 'p', 'Enabled': True}414        table.meta.client.update_time_to_live(TableName=table.name,415            TimeToLiveSpecification=ttl_spec)416        # p1 is in the past, but it's a string, not the required numeric type,417        # so the item should never expire.418        p1 = str(int(time.time()) - 60)419        table.put_item(Item={'p': p1})420        start_time = time.time()421        while time.time() < start_time + duration:422            print(f"--- {int(time.time()-start_time)} seconds")423            if 'Item' in table.get_item(Key={'p': p1}):424                print("p1 alive")425            time.sleep(duration/15)426        # After the delay, p2 should be alive, p1 should not427        assert 'Item' in table.get_item(Key={'p': p1})428# Test how in a table with a GSI or LSI, expiring an item also removes it429# from the GSI and LSI.430# We already tested above the various reasons for an item expiring or not431# expiring, so we don't need to continue testing these various cases here,432# and can test just one expiring item. This also allows us to finish the433# test as soon as the item expires, instead of after a fixed "duration" as434# in the above tests.435@pytest.mark.veryslow436def test_ttl_expiration_gsi_lsi(dynamodb):437    # In our experiments we noticed that when a table has secondary indexes,438    # items are expired with significant delay. Whereas a 10 minute delay439    # for regular tables was typical, in the table we have here we saw440    # a typical delay of 30 minutes before items expired.441    max_duration = 3600 if is_aws(dynamodb) else 10442    with new_test_table(dynamodb,443        KeySchema=[444            { 'AttributeName': 'p', 'KeyType': 'HASH' },445            { 'AttributeName': 'c', 'KeyType': 'RANGE' },446        ],447        LocalSecondaryIndexes=[448            {   'IndexName': 'lsi',449                'KeySchema': [450                    { 'AttributeName': 'p', 'KeyType': 'HASH' },451                    { 'AttributeName': 'l', 'KeyType': 'RANGE' },452                ],453                'Projection': { 'ProjectionType': 'ALL' },454            },455        ],456        GlobalSecondaryIndexes=[457            {   'IndexName': 'gsi',458                'KeySchema': [459                    { 'AttributeName': 'g', 'KeyType': 'HASH' },460                ],461                'Projection': { 'ProjectionType': 'ALL' }462            },463        ],464        AttributeDefinitions=[465            { 'AttributeName': 'p', 'AttributeType': 'S' },466            { 'AttributeName': 'c', 'AttributeType': 'S' },467            { 'AttributeName': 'g', 'AttributeType': 'S' },468            { 'AttributeName': 'l', 'AttributeType': 'S' },469        ]) as table:470        ttl_spec = {'AttributeName': 'expiration', 'Enabled': True}471        response = table.meta.client.update_time_to_live(TableName=table.name,472            TimeToLiveSpecification=ttl_spec)473        assert 'TimeToLiveSpecification' in response474        assert response['TimeToLiveSpecification'] == ttl_spec475        p = random_string()476        c = random_string()477        g = random_string()478        l = random_string()479        # expiration one minute in the past, so item should expire ASAP.480        expiration = int(time.time()) - 60481        table.put_item(Item={'p': p, 'c': c, 'g': g, 'l': l, 'expiration': expiration})482        start_time = time.time()483        gsi_was_alive = False484        while time.time() < start_time + max_duration:485            print(f"--- {int(time.time()-start_time)} seconds")486            base_alive = 'Item' in table.get_item(Key={'p': p, 'c': c})487            gsi_alive = bool(full_query(table, IndexName='gsi',488                ConsistentRead=False,489                KeyConditionExpression="g=:g",490                ExpressionAttributeValues={':g': g}))491            lsi_alive = bool(full_query(table, IndexName='lsi',492                KeyConditionExpression="p=:p and l=:l",493                ExpressionAttributeValues={':p': p, ':l': l}))494            if base_alive:495                print("base alive")496            if gsi_alive:497                print("gsi alive")498                # gsi takes time to go up, so make sure it did499                gsi_was_alive = True500            if lsi_alive:501                print("lsi alive")502            # If the base item, gsi item and lsi item have all expired, the503            # test is done - and successful:504            if not base_alive and not gsi_alive and gsi_was_alive and not lsi_alive:505                return506            time.sleep(max_duration/15)507        pytest.fail('base, gsi, or lsi not expired')508# Above in test_ttl_expiration_hash() and test_ttl_expiration_range() we509# checked the case where TTL's expiration-time attribute is not a regular510# attribute but one of the two key attributes. Alternator encodes these511# attributes differently (as a real column in the schema - not a serialized512# JSON inside a map column), so needed to check that such an attribute is513# usable as well. LSI (and, currently, also GSI) *also* implement their keys514# differently (like the base key), so let's check that having the TTL column515# a GSI or LSI key still works. Even though it is unlikely that any user516# would want to have such a use case.517# The following test only tries LSI, which we're sure will always have this518# special case (for GSI, we are considering changing the implementation519# and make their keys regular attributes - to support adding GSIs after a520# table already exists - so after such a change GSIs will no longer be a521# special case.522@pytest.mark.veryslow523def test_ttl_expiration_lsi_key(dynamodb):524    # In my experiments, a 30-minute (1800 seconds) is the typical delay525    # for expiration in this test for DynamoDB526    max_duration = 3600 if is_aws(dynamodb) else 10527    with new_test_table(dynamodb,528        KeySchema=[529            { 'AttributeName': 'p', 'KeyType': 'HASH' },530            { 'AttributeName': 'c', 'KeyType': 'RANGE' },531        ],532        LocalSecondaryIndexes=[533            {   'IndexName': 'lsi',534                'KeySchema': [535                    { 'AttributeName': 'p', 'KeyType': 'HASH' },536                    { 'AttributeName': 'l', 'KeyType': 'RANGE' },537                ],538                'Projection': { 'ProjectionType': 'ALL' },539            },540        ],541        AttributeDefinitions=[542            { 'AttributeName': 'p', 'AttributeType': 'S' },543            { 'AttributeName': 'c', 'AttributeType': 'S' },544            { 'AttributeName': 'l', 'AttributeType': 'N' },545        ]) as table:546        # The expiration-time attribute is the LSI key "l":547        ttl_spec = {'AttributeName': 'l', 'Enabled': True}548        response = table.meta.client.update_time_to_live(TableName=table.name,549            TimeToLiveSpecification=ttl_spec)550        assert 'TimeToLiveSpecification' in response551        assert response['TimeToLiveSpecification'] == ttl_spec552        p = random_string()553        c = random_string()554        l = random_string()555        # expiration one minute in the past, so item should expire ASAP.556        expiration = int(time.time()) - 60557        table.put_item(Item={'p': p, 'c': c, 'l': expiration})558        start_time = time.time()559        gsi_was_alive = False560        while time.time() < start_time + max_duration:561            print(f"--- {int(time.time()-start_time)} seconds")562            if 'Item' in table.get_item(Key={'p': p, 'c': c}):563                print("base alive")564            else:565                # test is done - and successful:566                return567            time.sleep(max_duration/15)568        pytest.fail('base not expired')569# Check that in the DynamoDB Streams API, an event appears about an item570# becoming expired. This event should contain be a REMOVE event, contain571# the appropriate information about the expired item (its key and/or its572# content), and a special userIdentity flag saying that this is not a regular573# REMOVE but an expiration.574@pytest.mark.veryslow575@pytest.mark.xfail(reason="TTL expiration event in streams not yet marked")576def test_ttl_expiration_streams(dynamodb, dynamodbstreams):577    # In my experiments, a 30-minute (1800 seconds) is the typical578    # expiration delay in this test. If the test doesn't finish within579    # max_duration, we report a failure.580    max_duration = 3600 if is_aws(dynamodb) else 10581    with new_test_table(dynamodb,582        KeySchema=[583            { 'AttributeName': 'p', 'KeyType': 'HASH' },584            { 'AttributeName': 'c', 'KeyType': 'RANGE' },585        ],586        AttributeDefinitions=[587            { 'AttributeName': 'p', 'AttributeType': 'S' },588            { 'AttributeName': 'c', 'AttributeType': 'S' },589        ],590        StreamSpecification={591            'StreamEnabled': True,592            'StreamViewType':  'NEW_AND_OLD_IMAGES'}593        ) as table:594        ttl_spec = {'AttributeName': 'expiration', 'Enabled': True}595        table.meta.client.update_time_to_live(TableName=table.name,596            TimeToLiveSpecification=ttl_spec)597        # Before writing to the table, wait for the stream to become active598        # so we can be sure that the expiration - even if it miraculously599        # happens in a second (it usually takes 30 minutes) - is guaranteed600        # to reach the stream.601        stream_enabled = False602        start_time = time.time()603        while time.time() < start_time + 120:604            desc = table.meta.client.describe_table(TableName=table.name)['Table']605            if 'LatestStreamArn' in desc:606                arn = desc['LatestStreamArn']607                desc = dynamodbstreams.describe_stream(StreamArn=arn)608                if desc['StreamDescription']['StreamStatus'] == 'ENABLED':609                    stream_enabled = True610                    break611            time.sleep(10)612        assert stream_enabled613        # Write a single expiring item. Set its expiration one minute in the614        # past, so the item should expire ASAP.615        p = random_string()616        c = random_string()617        expiration = int(time.time()) - 60618        table.put_item(Item={'p': p, 'c': c, 'animal': 'dog', 'expiration': expiration})619        # Wait (up to max_duration) for the item to expire, and for the620        # expiration event to reach the stream:621        start_time = time.time()622        event_found = False623        while time.time() < start_time + max_duration:624            print(f"--- {int(time.time()-start_time)} seconds")625            item_expired = not 'Item' in table.get_item(Key={'p': p, 'c': c})626            for record in read_entire_stream(dynamodbstreams, table):627                # An expired item has a specific userIdentity as follows:628                if record.get('userIdentity') == { 'Type': 'Service', 'PrincipalId': 'dynamodb.amazonaws.com' }:629                    # The expired item should be a REMOVE, and because we630                    # asked for old images and both the key and the full631                    # content.632                    assert record['eventName'] == 'REMOVE'633                    assert record['dynamodb']['Keys'] == {'p': {'S': p}, 'c': {'S': c}}634                    assert record['dynamodb']['OldImage'] == {'p': {'S': p}, 'c': {'S': c}, 'animal': {'S': 'dog'}, 'expiration': {'N': str(expiration)} }635                    event_found = True636            print(f"item expired {item_expired} event {event_found}")637            if item_expired and event_found:638                return639            time.sleep(max_duration/15)640        pytest.fail('item did not expire or event did not reach stream')641# Utility function for reading the entire contents of a table's DynamoDB642# Streams. This function is only useful when we expect only a handful of643# items, and performance is not important, because nothing is cached between644# calls. So it's only used in "veryslow"-marked tests above.645def read_entire_stream(dynamodbstreams, table):646    # Look for the latest stream. If there is none, return nothing647    desc = table.meta.client.describe_table(TableName=table.name)['Table']648    if not 'LatestStreamArn' in desc:649        return []650    arn = desc['LatestStreamArn']651    # List all shards of the stream in an array "shards":652    response = dynamodbstreams.describe_stream(StreamArn=arn)['StreamDescription']653    shards = [x['ShardId'] for x in response['Shards']]654    while 'LastEvaluatedShardId' in response:655        response = dynamodbstreams.describe_stream(StreamArn=arn,656            ExclusiveStartShardId=response['LastEvaluatedShardId'])['StreamDescription']657        shards.extend([x['ShardId'] for x in response['Shards']])658    records = []659    for shard_id in shards:660        # Get an interator for everything (TRIM_HORIZON) in the shard661        iter = dynamodbstreams.get_shard_iterator(StreamArn=arn,662            ShardId=shard_id, ShardIteratorType='TRIM_HORIZON')['ShardIterator']663        while iter != None:664            response = dynamodbstreams.get_records(ShardIterator=iter)665            # DynamoDB will continue returning records until reaching the666            # current end, and only then we will get an empty response.667            if len(response['Records']) == 0:668                break669            records.extend(response['Records'])670            iter = response.get('NextShardIterator')671    return records672# Whereas tests above use tiny tables with just a few items, this one creates673# a table with significantly more items - one that will need to be scanned674# in more than one page - so verifies that the expiration scanner service675# does the paging properly.676@pytest.mark.veryslow677def test_ttl_expiration_long(dynamodb):678    # Write 100*N items to the table, 1/100th of them are expired. The test679    # completes when all of them are expired (or on timeout).680    # To compilicate matter for the paging that we're trying to test,681    # have N partitions with 100 items in each, so potentially the paging682    # might stop in the middle of a partition.683    # We need to pick a relatively big N to cause the 100*N items to exceed684    # the size of a single page (I tested this by stopping the scanner after685    # the first page and checking which N starts to generate incorrect results)686    N=400687    max_duration = 1200 if is_aws(dynamodb) else 15688    with new_test_table(dynamodb,689        KeySchema=[ { 'AttributeName': 'p', 'KeyType': 'HASH' },690                    { 'AttributeName': 'c', 'KeyType': 'RANGE' } ],691        AttributeDefinitions=[ { 'AttributeName': 'p', 'AttributeType': 'N' },692                               { 'AttributeName': 'c', 'AttributeType': 'N' }]) as table:693        ttl_spec = {'AttributeName': 'expiration', 'Enabled': True}694        response = table.meta.client.update_time_to_live(TableName=table.name,695            TimeToLiveSpecification=ttl_spec)696        with table.batch_writer() as batch:697            for p in range(N):698                for c in range(100):699                    # Only the first item (c==0) in each partition will expire700                    expiration = int(time.time())-60 if c==0 else int(time.time())+3600701                    batch.put_item(Item={702                        'p': p, 'c': c, 'expiration': expiration })703        start_time = time.time()704        while time.time() < start_time + max_duration:705            # We could have used Select=COUNT here, but Alternator doesn't706            # implement it yet (issue #5058).707            count = 0708            response = table.scan(ConsistentRead=True)...

test_dynamodb.py

Source:test_dynamodb.py

1from decimal import Decimal2from unittest.mock import MagicMock, Mock, patch, sentinel3import pytest4from celery import states5from celery.backends import dynamodb as module6from celery.backends.dynamodb import DynamoDBBackend7from celery.exceptions import ImproperlyConfigured8pytest.importorskip('boto3')9class test_DynamoDBBackend:10    def setup(self):11        self._static_timestamp = Decimal(1483425566.52)  # noqa12        self.app.conf.result_backend = 'dynamodb://'13    @property14    def backend(self):15        """:rtype: DynamoDBBackend"""16        return self.app.backend17    def test_init_no_boto3(self):18        prev, module.boto3 = module.boto3, None19        try:20            with pytest.raises(ImproperlyConfigured):21                DynamoDBBackend(app=self.app)22        finally:23            module.boto3 = prev24    def test_init_aws_credentials(self):25        with pytest.raises(ImproperlyConfigured):26            DynamoDBBackend(27                app=self.app,28                url='dynamodb://a:@'29            )30    def test_init_invalid_ttl_seconds_raises(self):31        with pytest.raises(ValueError):32            DynamoDBBackend(33                app=self.app,34                url='dynamodb://@?ttl_seconds=1d'35            )36    def test_get_client_explicit_endpoint(self):37        table_creation_path = \38            'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'39        with patch('boto3.client') as mock_boto_client, \40                patch(table_creation_path):41            self.app.conf.dynamodb_endpoint_url = 'http://my.domain.com:666'42            backend = DynamoDBBackend(43                app=self.app,44                url='dynamodb://@us-east-1'45            )46            client = backend._get_client()47            assert backend.client is client48            mock_boto_client.assert_called_once_with(49                'dynamodb',50                endpoint_url='http://my.domain.com:666',51                region_name='us-east-1'52            )53            assert backend.endpoint_url == 'http://my.domain.com:666'54    def test_get_client_local(self):55        table_creation_path = \56            'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'57        with patch('boto3.client') as mock_boto_client, \58                patch(table_creation_path):59            backend = DynamoDBBackend(60                app=self.app,61                url='dynamodb://@localhost:8000'62            )63            client = backend._get_client()64            assert backend.client is client65            mock_boto_client.assert_called_once_with(66                'dynamodb',67                endpoint_url='http://localhost:8000',68                region_name='us-east-1'69            )70            assert backend.endpoint_url == 'http://localhost:8000'71    def test_get_client_credentials(self):72        table_creation_path = \73            'celery.backends.dynamodb.DynamoDBBackend._get_or_create_table'74        with patch('boto3.client') as mock_boto_client, \75                patch(table_creation_path):76            backend = DynamoDBBackend(77                app=self.app,78                url='dynamodb://key:secret@test'79            )80            client = backend._get_client()81            assert client is backend.client82            mock_boto_client.assert_called_once_with(83                'dynamodb',84                aws_access_key_id='key',85                aws_secret_access_key='secret',86                region_name='test'87            )88            assert backend.aws_region == 'test'89    @patch('boto3.client')90    @patch('celery.backends.dynamodb.DynamoDBBackend._get_or_create_table')91    @patch('celery.backends.dynamodb.DynamoDBBackend._validate_ttl_methods')92    @patch('celery.backends.dynamodb.DynamoDBBackend._set_table_ttl')93    def test_get_client_time_to_live_called(94            self,95            mock_set_table_ttl,96            mock_validate_ttl_methods,97            mock_get_or_create_table,98            mock_boto_client,99    ):100        backend = DynamoDBBackend(101            app=self.app,102            url='dynamodb://key:secret@test?ttl_seconds=30'103        )104        backend._get_client()105        mock_validate_ttl_methods.assert_called_once()106        mock_set_table_ttl.assert_called_once()107    def test_get_or_create_table_not_exists(self):108        self.backend._client = MagicMock()109        mock_create_table = self.backend._client.create_table = MagicMock()110        mock_describe_table = self.backend._client.describe_table = \111            MagicMock()112        mock_describe_table.return_value = {113            'Table': {114                'TableStatus': 'ACTIVE'115            }116        }117        self.backend._get_or_create_table()118        mock_create_table.assert_called_once_with(119            **self.backend._get_table_schema()120        )121    def test_get_or_create_table_already_exists(self):122        from botocore.exceptions import ClientError123        self.backend._client = MagicMock()124        mock_create_table = self.backend._client.create_table = MagicMock()125        client_error = ClientError(126            {127                'Error': {128                    'Code': 'ResourceInUseException',129                    'Message': 'Table already exists: {}'.format(130                        self.backend.table_name131                    )132                }133            },134            'CreateTable'135        )136        mock_create_table.side_effect = client_error137        mock_describe_table = self.backend._client.describe_table = \138            MagicMock()139        mock_describe_table.return_value = {140            'Table': {141                'TableStatus': 'ACTIVE'142            }143        }144        self.backend._get_or_create_table()145        mock_describe_table.assert_called_once_with(146            TableName=self.backend.table_name147        )148    def test_wait_for_table_status(self):149        self.backend._client = MagicMock()150        mock_describe_table = self.backend._client.describe_table = \151            MagicMock()152        mock_describe_table.side_effect = [153            {'Table': {154                'TableStatus': 'CREATING'155            }},156            {'Table': {157                'TableStatus': 'SOME_STATE'158            }}159        ]160        self.backend._wait_for_table_status(expected='SOME_STATE')161        assert mock_describe_table.call_count == 2162    def test_has_ttl_none_returns_none(self):163        self.backend.time_to_live_seconds = None164        assert self.backend._has_ttl() is None165    def test_has_ttl_lt_zero_returns_false(self):166        self.backend.time_to_live_seconds = -1167        assert self.backend._has_ttl() is False168    def test_has_ttl_gte_zero_returns_true(self):169        self.backend.time_to_live_seconds = 30170        assert self.backend._has_ttl() is True171    def test_validate_ttl_methods_present_returns_none(self):172        self.backend._client = MagicMock()173        assert self.backend._validate_ttl_methods() is None174    def test_validate_ttl_methods_missing_raise(self):175        self.backend._client = MagicMock()176        delattr(self.backend._client, 'describe_time_to_live')177        delattr(self.backend._client, 'update_time_to_live')178        with pytest.raises(AttributeError):179            self.backend._validate_ttl_methods()180        with pytest.raises(AttributeError):181            self.backend._validate_ttl_methods()182    def test_set_table_ttl_describe_time_to_live_fails_raises(self):183        from botocore.exceptions import ClientError184        self.backend.time_to_live_seconds = -1185        self.backend._client = MagicMock()186        mock_describe_time_to_live = \187            self.backend._client.describe_time_to_live = MagicMock()188        client_error = ClientError(189            {190                'Error': {191                    'Code': 'Foo',192                    'Message': 'Bar',193                }194            },195            'DescribeTimeToLive'196        )197        mock_describe_time_to_live.side_effect = client_error198        with pytest.raises(ClientError):199            self.backend._set_table_ttl()200    def test_set_table_ttl_enable_when_disabled_succeeds(self):201        self.backend.time_to_live_seconds = 30202        self.backend._client = MagicMock()203        mock_update_time_to_live = self.backend._client.update_time_to_live = \204            MagicMock()205        mock_describe_time_to_live = \206            self.backend._client.describe_time_to_live = MagicMock()207        mock_describe_time_to_live.return_value = {208            'TimeToLiveDescription': {209                'TimeToLiveStatus': 'DISABLED',210                'AttributeName': self.backend._ttl_field.name211            }212        }213        self.backend._set_table_ttl()214        mock_describe_time_to_live.assert_called_once_with(215            TableName=self.backend.table_name216        )217        mock_update_time_to_live.assert_called_once()218    def test_set_table_ttl_enable_when_enabled_with_correct_attr_succeeds(self):219        self.backend.time_to_live_seconds = 30220        self.backend._client = MagicMock()221        self.backend._client.update_time_to_live = MagicMock()222        mock_describe_time_to_live = \223            self.backend._client.describe_time_to_live = MagicMock()224        mock_describe_time_to_live.return_value = {225            'TimeToLiveDescription': {226                'TimeToLiveStatus': 'ENABLED',227                'AttributeName': self.backend._ttl_field.name228            }229        }230        self.backend._set_table_ttl()231        mock_describe_time_to_live.assert_called_once_with(232            TableName=self.backend.table_name233        )234    def test_set_table_ttl_enable_when_currently_disabling_raises(self):235        from botocore.exceptions import ClientError236        self.backend.time_to_live_seconds = 30237        self.backend._client = MagicMock()238        mock_update_time_to_live = self.backend._client.update_time_to_live = \239            MagicMock()240        client_error = ClientError(241            {242                'Error': {243                    'Code': 'ValidationException',244                    'Message': (245                        'Time to live has been modified multiple times '246                        'within a fixed interval'247                    )248                }249            },250            'UpdateTimeToLive'251        )252        mock_update_time_to_live.side_effect = client_error253        mock_describe_time_to_live = \254            self.backend._client.describe_time_to_live = MagicMock()255        mock_describe_time_to_live.return_value = {256            'TimeToLiveDescription': {257                'TimeToLiveStatus': 'DISABLING',258                'AttributeName': self.backend._ttl_field.name259            }260        }261        with pytest.raises(ClientError):262            self.backend._set_table_ttl()263    def test_set_table_ttl_enable_when_enabled_with_wrong_attr_raises(self):264        from botocore.exceptions import ClientError265        self.backend.time_to_live_seconds = 30266        self.backend._client = MagicMock()267        mock_update_time_to_live = self.backend._client.update_time_to_live = \268            MagicMock()269        wrong_attr_name = self.backend._ttl_field.name + 'x'270        client_error = ClientError(271            {272                'Error': {273                    'Code': 'ValidationException',274                    'Message': (275                        'TimeToLive is active on a different AttributeName: '276                        'current AttributeName is {}'277                    ).format(wrong_attr_name)278                }279            },280            'UpdateTimeToLive'281        )282        mock_update_time_to_live.side_effect = client_error283        mock_describe_time_to_live = \284            self.backend._client.describe_time_to_live = MagicMock()285        mock_describe_time_to_live.return_value = {286            'TimeToLiveDescription': {287                'TimeToLiveStatus': 'ENABLED',288                'AttributeName': self.backend._ttl_field.name + 'x'289            }290        }291        with pytest.raises(ClientError):292            self.backend._set_table_ttl()293    def test_set_table_ttl_disable_when_disabled_succeeds(self):294        self.backend.time_to_live_seconds = -1295        self.backend._client = MagicMock()296        self.backend._client.update_time_to_live = MagicMock()297        mock_describe_time_to_live = \298            self.backend._client.describe_time_to_live = MagicMock()299        mock_describe_time_to_live.return_value = {300            'TimeToLiveDescription': {301                'TimeToLiveStatus': 'DISABLED'302            }303        }304        self.backend._set_table_ttl()305        mock_describe_time_to_live.assert_called_once_with(306            TableName=self.backend.table_name307        )308    def test_set_table_ttl_disable_when_currently_enabling_raises(self):309        from botocore.exceptions import ClientError310        self.backend.time_to_live_seconds = -1311        self.backend._client = MagicMock()312        mock_update_time_to_live = self.backend._client.update_time_to_live = \313            MagicMock()314        client_error = ClientError(315            {316                'Error': {317                    'Code': 'ValidationException',318                    'Message': (319                        'Time to live has been modified multiple times '320                        'within a fixed interval'321                    )322                }323            },324            'UpdateTimeToLive'325        )326        mock_update_time_to_live.side_effect = client_error327        mock_describe_time_to_live = \328            self.backend._client.describe_time_to_live = MagicMock()329        mock_describe_time_to_live.return_value = {330            'TimeToLiveDescription': {331                'TimeToLiveStatus': 'ENABLING',332                'AttributeName': self.backend._ttl_field.name333            }334        }335        with pytest.raises(ClientError):336            self.backend._set_table_ttl()337    def test_prepare_get_request(self):338        expected = {339            'TableName': 'celery',340            'Key': {'id': {'S': 'abcdef'}}341        }342        assert self.backend._prepare_get_request('abcdef') == expected343    def test_prepare_put_request(self):344        expected = {345            'TableName': 'celery',346            'Item': {347                'id': {'S': 'abcdef'},348                'result': {'B': 'val'},349                'timestamp': {350                    'N': str(Decimal(self._static_timestamp))351                }352            }353        }354        with patch('celery.backends.dynamodb.time', self._mock_time):355            result = self.backend._prepare_put_request('abcdef', 'val')356        assert result == expected357    def test_prepare_put_request_with_ttl(self):358        ttl = self.backend.time_to_live_seconds = 30359        expected = {360            'TableName': 'celery',361            'Item': {362                'id': {'S': 'abcdef'},363                'result': {'B': 'val'},364                'timestamp': {365                    'N': str(Decimal(self._static_timestamp))366                },367                'ttl': {368                    'N': str(int(self._static_timestamp + ttl))369                }370            }371        }372        with patch('celery.backends.dynamodb.time', self._mock_time):373            result = self.backend._prepare_put_request('abcdef', 'val')374        assert result == expected375    def test_item_to_dict(self):376        boto_response = {377            'Item': {378                'id': {379                    'S': sentinel.key380                },381                'result': {382                    'B': sentinel.value383                },384                'timestamp': {385                    'N': Decimal(1)386                }387            }388        }389        converted = self.backend._item_to_dict(boto_response)390        assert converted == {391            'id': sentinel.key,392            'result': sentinel.value,393            'timestamp': Decimal(1)394        }395    def test_get(self):396        self.backend._client = Mock(name='_client')397        self.backend._client.get_item = MagicMock()398        assert self.backend.get('1f3fab') is None399        self.backend.client.get_item.assert_called_once_with(400            Key={'id': {'S': '1f3fab'}},401            TableName='celery'402        )403    def _mock_time(self):404        return self._static_timestamp405    def test_set(self):406        self.backend._client = MagicMock()407        self.backend._client.put_item = MagicMock()408        # should return None409        with patch('celery.backends.dynamodb.time', self._mock_time):410            assert self.backend._set_with_state(sentinel.key, sentinel.value, states.SUCCESS) is None411        assert self.backend._client.put_item.call_count == 1412        _, call_kwargs = self.backend._client.put_item.call_args413        expected_kwargs = {414            'Item': {415                'timestamp': {'N': str(self._static_timestamp)},416                'id': {'S': str(sentinel.key)},417                'result': {'B': sentinel.value}418            },419            'TableName': 'celery'420        }421        assert call_kwargs['Item'] == expected_kwargs['Item']422        assert call_kwargs['TableName'] == 'celery'423    def test_set_with_ttl(self):424        ttl = self.backend.time_to_live_seconds = 30425        self.backend._client = MagicMock()426        self.backend._client.put_item = MagicMock()427        # should return None428        with patch('celery.backends.dynamodb.time', self._mock_time):429            assert self.backend._set_with_state(sentinel.key, sentinel.value, states.SUCCESS) is None430        assert self.backend._client.put_item.call_count == 1431        _, call_kwargs = self.backend._client.put_item.call_args432        expected_kwargs = {433            'Item': {434                'timestamp': {'N': str(self._static_timestamp)},435                'id': {'S': str(sentinel.key)},436                'result': {'B': sentinel.value},437                'ttl': {'N': str(int(self._static_timestamp + ttl))},438            },439            'TableName': 'celery'440        }441        assert call_kwargs['Item'] == expected_kwargs['Item']442        assert call_kwargs['TableName'] == 'celery'443    def test_delete(self):444        self.backend._client = Mock(name='_client')445        mocked_delete = self.backend._client.delete = Mock('client.delete')446        mocked_delete.return_value = None447        # should return None448        assert self.backend.delete('1f3fab') is None449        self.backend.client.delete_item.assert_called_once_with(450            Key={'id': {'S': '1f3fab'}},451            TableName='celery'452        )453    def test_backend_by_url(self, url='dynamodb://'):454        from celery.app import backends455        from celery.backends.dynamodb import DynamoDBBackend456        backend, url_ = backends.by_url(url, self.app.loader)457        assert backend is DynamoDBBackend458        assert url_ == url459    def test_backend_params_by_url(self):460        self.app.conf.result_backend = (461            'dynamodb://@us-east-1/celery_results'462            '?read=10'463            '&write=20'464            '&ttl_seconds=600'465        )466        assert self.backend.aws_region == 'us-east-1'467        assert self.backend.table_name == 'celery_results'468        assert self.backend.read_capacity_units == 10469        assert self.backend.write_capacity_units == 20470        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.