KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
