KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use update_visibility_timeout method in localstack
Best Python code snippet using localstack_python
test_resolution_hierarchy.py
Source:test_resolution_hierarchy.py 
1# Copyright 2020 The Johns Hopkins University Applied Physics Laboratory2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14# lambdafcns contains symbolic links to lambda functions in boss-tools/lambda.15# Since lambda is a reserved word, this allows importing from that folder 16# without updating scripts responsible for deploying the lambda code.17"""18Note that the module under test (resolution_hierarchy.py) is imported inside19of the various test classes. This must be done to ensure that none of the20boto3 clients are instantiated before the moto mocks are put in place. Also,21environment variables for the AWS keys should be set to a bogus value such as22`testing` to ensure that no real AWS resources are accidentally used.23https://github.com/spulec/moto#how-do-i-avoid-tests-from-mutating-my-real-infrastructure24"""25import os26import sys27import importlib28import json29import types30import unittest31from unittest.mock import patch, MagicMock, call32from datetime import datetime, timedelta33from bossutils.multidimensional import ceildiv34from ._test_case_with_patch_object import TestCaseWithPatchObject35import boto336from moto import mock_sqs37#import logging.config38#logging.config.dictConfig({39# 'version': 1,40# 'disable_existing_loggers': False,41# 'formatters': { 'detailed': { 'format': '%(message)s' } },42# 'handlers': { 'console': { 'class': 'logging.StreamHandler',43# 'formatter': 'detailed' } },44# 'root': { 'handlers': ['console'],45# 'level': 'DEBUG' }46#})47####################################################48################## CUSTOM MOCKING ##################49# This is a fake id as another safe guard against utilizing real AWS resources.50ACCOUNT_ID = '123456789012'51REGION = 'us-east-1'52SESSION = boto3.session.Session(region_name = REGION)53#SQS_URL = 'https://sqs.{}.amazonaws.com/{}/'.format(REGION, ACCOUNT_ID)54SQS_URL = 'https://queue.amazonaws.com/{}/'.format(ACCOUNT_ID)55DOWNSAMPLE_QUEUE_URL = SQS_URL + 'downsample_queue'56RECEIPT_HANDLE = '987654321'57# Stub XYZMorton implementation that returns a unique number when called58def XYZMorton(xyz):59 x,y,z = xyz60 morton = (x << 40) | (y << 20) | (z)61 return morton62# Stub ndtype.CUBOIDSIZE63class CUBOIDSIZE(object):64 def __getitem__(self, key):65 return (512, 512, 16)66class DateTime(datetime):67 @staticmethod68 def now():69 return datetime(year=1970, month=1, day=1)70class MultiprocessingPool(object):71 def __init__(self, size):72 self.size = size73 def __enter__(self):74 return self75 def __exit__(self, t, v, tb):76 return False77 def starmap(self, func, args):78 rtn = []79 for arg in args:80 rtn.append(func(*arg))81 return rtn82def make_check_queue(**kwargs):83 """kwargs:84 arn: values85 """86 def wrap(values):87 if type(values) == list:88 for value in values:89 yield value90 while True:91 yield 092 else:93 while True:94 yield values95 arns = {96 arn: wrap(kwargs[arn])97 for arn in kwargs98 }99 def check_queue(arn):100 return next(arns[arn])101 return check_queue102# Mockup import for resolution_hierarchy and bossutils.multidimensional103sys.modules['bossutils'] = MagicMock()104sys.modules['bossutils'].aws.get_session.return_value = SESSION105sys.modules['numpy'] = MagicMock()106sys.modules['spdb.c_lib'] = MagicMock()107sys.modules['spdb.c_lib'].ndlib.XYZMorton = XYZMorton108sys.modules['spdb.c_lib.ndtype'] = MagicMock()109sys.modules['spdb.c_lib.ndtype'].CUBOIDSIZE = CUBOIDSIZE()110# Import the current verions of bossutils.multidimensional so that all of the111# 3D math works correctly112cur_dir = os.path.dirname(os.path.realpath(__file__))113bossutils_dir = os.path.normpath(os.path.join(cur_dir, '..', '..', 'bossutils'))114sys.path.insert(0, bossutils_dir)115md = importlib.import_module('multidimensional')116sys.modules['bossutils.multidimensional'] = md117sys.path.pop(0)118################## END CUSTOM MOCKING ##################119########################################################120"""121Test isotropic normal122Test anisotropic normal123Test anisotropic at the isotropic split124Test anisotropic after the isotropic split125Test non even frame size (frame size is not an even multiple of cube size)126Test non zero frame start127Test resolution / resolution_max logic so it doesn't change128Test passing in downsample_id129Always create / delete the queue or create it when creating the downsample_id and delete when res_lt_max is False?130"""131class TestDownsampleChannel(TestCaseWithPatchObject):132 @classmethod133 def setUpClass(cls):134 # Import the code to be tested 135 import resolution_hierarchy as rh136 rh.POOL_SIZE = 2137 rh.MAX_LAMBDA_TIME = rh.timedelta()138 rh.datetime = DateTime139 rh.Pool = MultiprocessingPool140 cls.rh = rh141 def setUp(self):142 self.mock_del_queue = self.patch_object(self.rh, 'delete_queue')143 self.mock_launch_lambdas = self.patch_object(self.rh, 'launch_lambdas')144 self.mock_populate_cubes = self.patch_object(self.rh, 'populate_cubes', return_value=1)145 self.mock_create_queue = self.patch_object(self.rh, 'create_queue', side_effect=lambda name: SQS_URL + name)146 mock_rand = self.patch_object(self.rh.random, 'random', return_value=0.1234)147 def get_args(self, scale=2, **kwargs):148 """Create args dict for input to the step function.149 Args:150 scale (int):151 kwargs (dict): Keywords replace key-values in the 'msg' nested dict152 Returns:153 (dict)154 """155 cube_size = md.XYZ(*CUBOIDSIZE()[0])156 frame_stop = cube_size * scale157 args = {158 # Only including keys that the Activity uses, the others are passed159 # to the downsample_volume lambda without inspection160 'msg': {161 'downsample_volume_lambda': 'lambda-arn',162 'x_start': 0,163 'y_start': 0,164 'z_start': 0,165 'x_stop': int(frame_stop.x), # int() to handle float scale multiplication results166 'y_stop': int(frame_stop.y),167 'z_stop': int(frame_stop.z),168 'resolution': 0,169 'resolution_max': 3,170 'res_lt_max': True,171 'type': 'isotropic',172 'iso_resolution': 3,173 'lookup_key': 'fake_lookup_key',174 },175 'queue_url': DOWNSAMPLE_QUEUE_URL,176 'job_receipt_handle': RECEIPT_HANDLE,177 }178 args['msg'].update(kwargs)179 return args180 def test_downsample_channel_iso(self):181 args1 = self.get_args(type='isotropic')182 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2183 expected = [call('downsample-dlq-1234'),184 call('downsample-cubes-1234')]185 self.assertEqual(self.mock_create_queue.mock_calls, expected)186 expected = call(SQS_URL + 'downsample-cubes-1234',187 md.XYZ(0,0,0),188 md.XYZ(2,2,2),189 md.XYZ(2,2,2))190 self.assertEqual(self.mock_populate_cubes.mock_calls, [expected])191 args = {192 'bucket_size': self.rh.BUCKET_SIZE,193 'args': self.get_args(type='isotropic')['msg'], # Need the original arguments194 'step': md.XYZ(2,2,2),195 'dim': md.XYZ(512, 512, 16),196 'use_iso_flag': False,197 'dlq_arn': SQS_URL + 'downsample-dlq-1234',198 'cubes_arn': SQS_URL + 'downsample-cubes-1234'199 }200 expected = call(ceildiv(self.mock_populate_cubes.return_value, self.rh.BUCKET_SIZE) + self.rh.EXTRA_LAMBDAS,201 args1['msg']['downsample_volume_lambda'],202 json.dumps(args).encode('UTF8'),203 SQS_URL + 'downsample-dlq-1234',204 SQS_URL + 'downsample-cubes-1234',205 DOWNSAMPLE_QUEUE_URL,206 RECEIPT_HANDLE)207 self.assertEqual(self.mock_launch_lambdas.mock_calls, [expected])208 self.assertEqual(args2['msg']['x_stop'], 512)209 self.assertEqual(args2['msg']['y_stop'], 512)210 self.assertEqual(args2['msg']['z_stop'], 16)211 self.assertEqual(args2['msg']['resolution'], 1)212 self.assertTrue(args2['msg']['res_lt_max'])213 expected = [call(SQS_URL + 'downsample-dlq-1234'),214 call(SQS_URL + 'downsample-cubes-1234')]215 self.assertEqual(self.mock_del_queue.mock_calls, expected)216 def test_downsample_channel_aniso(self):217 args1 = self.get_args(type='anisotropic')218 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2219 expected = [call('downsample-dlq-1234'),220 call('downsample-cubes-1234')]221 self.assertEqual(self.mock_create_queue.mock_calls, expected)222 expected = call(SQS_URL + 'downsample-cubes-1234',223 md.XYZ(0,0,0),224 md.XYZ(2,2,2),225 md.XYZ(2,2,1))226 self.assertEqual(self.mock_populate_cubes.mock_calls, [expected])227 args = {228 'bucket_size': self.rh.BUCKET_SIZE,229 'args': self.get_args(type='anisotropic')['msg'], # Need the original arguments230 'step': md.XYZ(2,2,1),231 'dim': md.XYZ(512, 512, 16),232 'use_iso_flag': False,233 'dlq_arn': SQS_URL + 'downsample-dlq-1234',234 'cubes_arn': SQS_URL + 'downsample-cubes-1234'235 }236 expected = call(ceildiv(self.mock_populate_cubes.return_value, self.rh.BUCKET_SIZE) + self.rh.EXTRA_LAMBDAS,237 args1['msg']['downsample_volume_lambda'],238 json.dumps(args).encode('UTF8'),239 SQS_URL + 'downsample-dlq-1234',240 SQS_URL + 'downsample-cubes-1234',241 DOWNSAMPLE_QUEUE_URL,242 RECEIPT_HANDLE)243 self.assertEqual(self.mock_launch_lambdas.mock_calls, [expected])244 self.assertEqual(args2['msg']['x_stop'], 512)245 self.assertEqual(args2['msg']['y_stop'], 512)246 self.assertEqual(args2['msg']['z_stop'], 32)247 self.assertNotIn('iso_x_start', args2['msg'])248 self.assertEqual(args2['msg']['resolution'], 1)249 self.assertTrue(args2['msg']['res_lt_max'])250 expected = [call(SQS_URL + 'downsample-dlq-1234'),251 call(SQS_URL + 'downsample-cubes-1234')]252 self.assertEqual(self.mock_del_queue.mock_calls, expected)253 def test_downsample_channel_aniso_split(self):254 args1 = self.get_args(type='anisotropic', iso_resolution=1, resolution=1)255 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2256 self.assertIn('iso_x_start', args2['msg'])257 self.assertEqual(args2['msg']['iso_x_stop'], 512)258 self.assertEqual(args2['msg']['iso_y_stop'], 512)259 self.assertEqual(args2['msg']['iso_z_stop'], 16)260 def test_downsample_channel_aniso_post_split(self):261 args1 = self.get_args(type='anisotropic',262 iso_resolution=0,263 iso_x_start = 0, iso_y_start = 0, iso_z_start = 0,264 iso_x_stop = 1024, iso_y_stop = 1024, iso_z_stop = 32)265 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2266 expected = call(SQS_URL + 'downsample-cubes-1234',267 md.XYZ(0,0,0),268 md.XYZ(2,2,2),269 md.XYZ(2,2,1))270 expected1 = call(SQS_URL + 'downsample-cubes-1234',271 md.XYZ(0,0,0),272 md.XYZ(2,2,2),273 md.XYZ(2,2,2))274 self.assertEqual(self.mock_populate_cubes.mock_calls, [expected, expected1])275 self.assertEqual(args2['msg']['x_stop'], 512)276 self.assertEqual(args2['msg']['y_stop'], 512)277 self.assertEqual(args2['msg']['z_stop'], 32)278 self.assertEqual(args2['msg']['iso_x_stop'], 512)279 self.assertEqual(args2['msg']['iso_y_stop'], 512)280 self.assertEqual(args2['msg']['iso_z_stop'], 16)281 def test_downsample_channel_not_even(self):282 args1 = self.get_args(type='isotropic', scale=2.5)283 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2284 expected = call(SQS_URL + 'downsample-cubes-1234',285 md.XYZ(0,0,0),286 md.XYZ(3,3,3), # Scaled up from 2.5 to 3 cubes that will be downsampled287 md.XYZ(2,2,2))288 self.assertEqual(self.mock_populate_cubes.mock_calls, [expected])289 self.assertEqual(args2['msg']['x_stop'], 640) # 640 = 512 * 2.5 / 2 (scaled up volume and then downsampled)290 self.assertEqual(args2['msg']['y_stop'], 640)291 self.assertEqual(args2['msg']['z_stop'], 20) # 20 = 16 * 2.5 / 2292 def test_downsample_channel_non_zero_start(self):293 ###### NOTES ######294 # Derek: I believe that downsampling a non-zero start frame actually downsamples the data correctly295 # the problem that we run into is that the frame start shrinks towards zero. This presents an296 # issue when the Channel's Frame starts at the initial offset and downsampling shrinks the frame297 # for lower resolutions beyond the lower extent of the Channel's Frame, makng that data unavailable.298 # Derek: The start / stop values were taken from a non-zero start downsample that showed there was a problem299 # when the start cube was not even, and therefore didn't align with a full downsample that started300 # at zero301 args1 = self.get_args(type='isotropic', resolution=4,302 x_start=4992, y_start=4992, z_start=1232,303 x_stop=6016, y_stop=6016, z_stop=1264)304 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2305 expected = call(SQS_URL + 'downsample-cubes-1234',306 md.XYZ(8,8,76),307 md.XYZ(12,12,79),308 md.XYZ(2,2,2))309 self.assertEqual(self.mock_populate_cubes.mock_calls, [expected])310 self.assertEqual(args2['msg']['x_start'], 2496)311 self.assertEqual(args2['msg']['y_start'], 2496)312 self.assertEqual(args2['msg']['z_start'], 616)313 self.assertEqual(args2['msg']['x_stop'], 3008)314 self.assertEqual(args2['msg']['y_stop'], 3008)315 self.assertEqual(args2['msg']['z_stop'], 632)316 def test_downsample_channel_before_stop(self):317 args1 = self.get_args(type='isotropic',318 resolution = 0, resolution_max = 3)319 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2320 self.assertEqual(args2['msg']['resolution'], 1)321 self.assertTrue(args2['msg']['res_lt_max'])322 def test_downsample_channel_at_stop(self):323 args1 = self.get_args(type='isotropic',324 resolution = 1, resolution_max = 3)325 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2326 self.assertEqual(args2['msg']['resolution'], 2)327 self.assertFalse(args2['msg']['res_lt_max'])328 def test_downsample_channel_after_stop(self):329 ###### NOTES ######330 # Derek: Not a real world call, as the input arguments are no longer valid331 args1 = self.get_args(type='isotropic',332 resolution = 2, resolution_max = 3)333 args2 = self.rh.downsample_channel(args1) # warning, will mutate args1 === args2334 self.assertEqual(args2['msg']['resolution'], 3)335 self.assertFalse(args2['msg']['res_lt_max'])336class TestChunk(unittest.TestCase):337 @classmethod338 def setUpClass(cls):339 # Import the code to be tested 340 import resolution_hierarchy as rh341 rh.POOL_SIZE = 2342 rh.MAX_LAMBDA_TIME = rh.timedelta()343 rh.datetime = DateTime344 rh.Pool = MultiprocessingPool345 cls.rh = rh346 def test_exact(self):347 args = [1,2,3,4]348 gen = self.rh.chunk(args, 2) # buckets of size 2349 self.assertEqual(type(gen), types.GeneratorType)350 chunks = list(gen)351 expected = [[1,2], [3,4]]352 self.assertEqual(len(chunks), 2)353 self.assertEqual(chunks, expected)354 def test_not_exist(self):355 args = (i for i in [1,2,3,4,5])356 gen = self.rh.chunk(args, 2) # buckets of size 2357 self.assertEqual(type(gen), types.GeneratorType)358 chunks = list(gen)359 expected = [[1,2], [3,4], [5]]360 self.assertEqual(len(chunks), 3)361 self.assertEqual(chunks, expected)362 def test_empty(self):363 args = []364 gen = self.rh.chunk(args, 2) # buckets of size 2365 self.assertEqual(type(gen), types.GeneratorType)366 chunks = list(gen)367 expected = []368 self.assertEqual(len(chunks), 0)369 self.assertEqual(chunks, expected)370class TestCubes(unittest.TestCase):371 @classmethod372 def setUpClass(cls):373 # Import the code to be tested 374 import resolution_hierarchy as rh375 rh.POOL_SIZE = 2376 rh.MAX_LAMBDA_TIME = rh.timedelta()377 rh.datetime = DateTime378 rh.Pool = MultiprocessingPool379 cls.rh = rh380 cls.start = md.XYZ(0,0,0)381 cls.stop = md.XYZ(2,2,2)382 cls.step = md.XYZ(1,1,1)383 def test_num_cubes(self):384 num = self.rh.num_cubes(self.start, self.stop, self.step)385 self.assertEqual(num, 8)386 def test_make_cubes(self):387 gen = self.rh.make_cubes(self.start, self.stop, self.step)388 self.assertEqual(type(gen), types.GeneratorType)389 cubes = list(gen)390 expected = [md.XYZ(0,0,0),391 md.XYZ(0,0,1),392 md.XYZ(0,1,0),393 md.XYZ(0,1,1),394 md.XYZ(1,0,0),395 md.XYZ(1,0,1),396 md.XYZ(1,1,0),397 md.XYZ(1,1,1)]398 self.assertEqual(len(cubes), 8)399 self.assertEqual(cubes, expected)400class TestEnqueue(TestCaseWithPatchObject):401 @classmethod402 def setUpClass(cls):403 # Import the code to be tested 404 import resolution_hierarchy as rh405 rh.POOL_SIZE = 2406 rh.MAX_LAMBDA_TIME = rh.timedelta()407 rh.datetime = DateTime408 rh.Pool = MultiprocessingPool409 cls.rh = rh410 cls.name = 'downsample-1234'411 cls.arn = SQS_URL + cls.name412 cls.start = md.XYZ(0,0,0)413 cls.stop = md.XYZ(2,2,2)414 cls.step = md.XYZ(1,1,1)415 cls.chunk = [md.XYZ(0,0,0),416 md.XYZ(1,1,1)]417 def test_populate(self):418 mock_enqueue_cubes = self.patch_object(self.rh, 'enqueue_cubes')419 count = self.rh.populate_cubes(self.arn, self.start, self.stop, self.step)420 self.assertEqual(count, 8)421 expected = [call(self.arn, [md.XYZ(0,0,0),422 md.XYZ(0,0,1),423 md.XYZ(0,1,0),424 md.XYZ(0,1,1)]),425 call(self.arn, [md.XYZ(1,0,0),426 md.XYZ(1,0,1),427 md.XYZ(1,1,0),428 md.XYZ(1,1,1)])]429 self.assertEqual(mock_enqueue_cubes.mock_calls, expected)430 @patch.object(SESSION, 'resource')431 def test_enqueue(self, mResource):432 self.rh.enqueue_cubes(self.arn, self.chunk)433 func = mResource.return_value.Queue.return_value.send_messages434 expected = [call(Entries=[{'Id': str(id(self.chunk[0])),435 'MessageBody': '[0, 0, 0]'},436 {'Id': str(id(self.chunk[1])),437 'MessageBody': '[1, 1, 1]'}])]438 self.assertEqual(func.mock_calls, expected)439 @patch.object(SESSION, 'resource')440 def test_enqueue_multi_batch(self, mResource):441 chunk = range(15)442 self.rh.enqueue_cubes(self.arn, chunk)443 func = mResource.return_value.Queue.return_value.send_messages444 self.assertEqual(len(func.mock_calls), 2)445 # index 2 of a call is the kwargs for the call446 self.assertEqual(len(func.mock_calls[0][2]['Entries']), 10)447 self.assertEqual(len(func.mock_calls[1][2]['Entries']), 5)448 @patch.object(SESSION, 'resource')449 def test_enqueue_error(self, mResource):450 func = mResource.return_value.Queue.return_value.send_messages451 func.side_effect = Exception("Could not enqueue data")452 with self.assertRaises(self.rh.ResolutionHierarchyError):453 self.rh.enqueue_cubes(self.arn, self.chunk)454class TestInvoke(TestCaseWithPatchObject):455 @classmethod456 def setUpClass(cls):457 # Import the code to be tested 458 import resolution_hierarchy as rh459 rh.POOL_SIZE = 2460 rh.MAX_LAMBDA_TIME = rh.timedelta()461 rh.datetime = DateTime462 rh.Pool = MultiprocessingPool463 cls.rh = rh464 def test_launch(self):465 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 466 mock_check_queue = self.patch_object(self.rh, 'check_queue',467 side_effect=make_check_queue(dlq_arn=0, cubes_arn=[5,4,3,2,1]))468 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)469 self.patch_object(self.rh, 'update_visibility_timeout')470 count = 10471 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',472 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)473 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE)474 self.assertEqual(len(mock_check_queue.mock_calls), (5 + self.rh.ZERO_COUNT) * 2) # (5+) for cubes_arn, (*2) for both queues475 def test_launch_empty(self):476 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 477 mock_check_queue = self.patch_object(self.rh, 'check_queue',478 side_effect=make_check_queue(dlq_arn=0, cubes_arn=0))479 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)480 self.patch_object(self.rh, 'update_visibility_timeout')481 count = 10482 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',483 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)484 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE)485 self.assertEqual(len(mock_check_queue.mock_calls), self.rh.ZERO_COUNT * 2) # * 2, one for each queue486 def test_launch_dlq(self):487 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 488 mock_check_queue = self.patch_object(self.rh, 'check_queue',489 side_effect=make_check_queue(dlq_arn=1, cubes_arn=0))490 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)491 self.patch_object(self.rh, 'update_visibility_timeout')492 count = 10493 with self.assertRaises(self.rh.FailedLambdaError):494 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',495 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)496 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE)497 self.assertEqual(len(mock_check_queue.mock_calls), 1) # single dlq check_queue call498 def test_launch_relaunch(self):499 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 500 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)501 self.patch_object(self.rh, 'update_visibility_timeout')502 cubes_count = [5] * self.rh.UNCHANGING_LAUNCH + [4,3,2,1]503 mock_check_queue = self.patch_object(self.rh, 'check_queue',504 side_effect=make_check_queue(dlq_arn=0, cubes_arn=cubes_count))505 count = 10506 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',507 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)508 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE + 1) # +1 relaunch509 self.assertEqual(len(mock_check_queue.mock_calls), (len(cubes_count) + self.rh.ZERO_COUNT) * 2) # (*2) for both queues510 def test_launch_relaunch_multi(self):511 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 512 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)513 self.patch_object(self.rh, 'update_visibility_timeout')514 cubes_count = [5] * self.rh.UNCHANGING_LAUNCH + [4] * self.rh.UNCHANGING_LAUNCH + [3,2,1]515 mock_check_queue = self.patch_object(self.rh, 'check_queue',516 side_effect=make_check_queue(dlq_arn=0, cubes_arn=cubes_count))517 count = 10518 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',519 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)520 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE + 2) # +2 relaunches521 self.assertEqual(len(mock_check_queue.mock_calls), (len(cubes_count) + self.rh.ZERO_COUNT) * 2) # (*2) for both queues522 def test_launch_throttle(self):523 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 524 self.patch_object(self.rh, 'update_visibility_timeout')525 cubes_count = [5] * self.rh.UNCHANGING_THROTTLE + [4,3,2,1]526 throttle_count = [2] * self.rh.UNCHANGING_THROTTLE + [2,2,2,1,0,0,0,0,0,0,0,0,0]527 mock_check_queue = self.patch_object(self.rh, 'check_queue',528 side_effect=make_check_queue(dlq_arn=0, cubes_arn=cubes_count))529 self.patch_object(self.rh, 'lambda_throttle_count', side_effect=throttle_count)530 count = 10531 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',532 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)533 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE + 1) # +1 relaunch534 self.assertEqual(len(mock_check_queue.mock_calls), ((len(cubes_count) + self.rh.ZERO_COUNT) * 2) + 2) # (+2) for 2 extra dlq polls in throttling loop, (*2) for both queues535 def test_launch_no_throttle(self):536 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 537 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)538 self.patch_object(self.rh, 'update_visibility_timeout')539 cubes_count = [5] * self.rh.UNCHANGING_THROTTLE + [4,3,2,1]540 mock_check_queue = self.patch_object(self.rh, 'check_queue',541 side_effect=make_check_queue(dlq_arn=0, cubes_arn=cubes_count))542 count = 10543 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',544 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)545 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE + 1) # +1 relaunch546 self.assertEqual(len(mock_check_queue.mock_calls), (len(cubes_count) + self.rh.ZERO_COUNT) * 2) # (*2) for both queues547 def test_launch_unchanging_max(self):548 mock_invoke_lambdas = self.patch_object(self.rh, 'invoke_lambdas', return_value=0) 549 self.patch_object(self.rh, 'lambda_throttle_count', return_value=0)550 self.patch_object(self.rh, 'update_visibility_timeout')551 cubes_count = [5] * self.rh.UNCHANGING_MAX552 mock_check_queue = self.patch_object(self.rh, 'check_queue',553 side_effect=make_check_queue(dlq_arn=0, cubes_arn=cubes_count))554 with self.assertRaises(self.rh.ResolutionHierarchyError):555 count = 10556 self.rh.launch_lambdas(count, 'lambda_arn', {}, 'dlq_arn', 'cubes_arn',557 DOWNSAMPLE_QUEUE_URL, RECEIPT_HANDLE)558 self.assertEqual(len(mock_invoke_lambdas.mock_calls), self.rh.POOL_SIZE + 1) # +1 relaunch559 self.assertEqual(len(mock_check_queue.mock_calls), len(cubes_count) * 2) # (*2) for both queues560 @patch.object(SESSION, 'client', autospec=True)561 def test_invoke(self, mClient):562 invoke = mClient.return_value.invoke563 count = 2564 self.rh.invoke_lambdas(count, 'lambda_arn', '{}', 'dlq_arn')565 expected = call(FunctionName = 'lambda_arn',566 InvocationType = 'Event',567 Payload = '{}')568 self.assertEqual(len(invoke.mock_calls), count)569 self.assertEqual(invoke.mock_calls[0], expected)570 @patch.object(SESSION, 'client', autospec=True)571 def test_invoke_error(self, mClient):572 invoke = mClient.return_value.invoke573 invoke.side_effect = Exception("Could not invoke lambda")574 with self.assertRaises(self.rh.ResolutionHierarchyError):575 count = 2576 self.rh.invoke_lambdas(count, 'lambda_arn', '{}', 'dlq_arn')577 @patch.object(SESSION, 'client', autospec=True)578 def test_invoke_dlq(self, mClient):579 invoke = mClient.return_value.invoke580 mock_check_queue = self.patch_object(self.rh, 'check_queue', return_value=1)581 with self.assertRaises(self.rh.ResolutionHierarchyError):582 count = 12583 self.rh.invoke_lambdas(count, 'lambda_arn', '{}', 'dlq_arn')584 self.assertEqual(len(invoke.mock_calls), 9)585class TestQueue(TestCaseWithPatchObject):586 """587 This class does not use SESSION defined at the top of this file because 588 that session was created before moto mocks are put in place.589 """590 @classmethod591 def setUpClass(cls):592 cls.downsample_id = '1234'593 cls.name = 'downsample-{}'.format(cls.downsample_id)594 cls.url = SQS_URL + cls.name595 def configure(self):596 """597 Import module under test. Done in here so that moto mocks can be598 established first.599 Returns:600 (Session): Mocked boto3 Session.601 """602 # Import the code to be tested 603 import resolution_hierarchy as rh604 rh.POOL_SIZE = 2605 rh.MAX_LAMBDA_TIME = rh.timedelta()606 rh.datetime = DateTime607 rh.Pool = MultiprocessingPool608 self.rh = rh609 session = boto3.session.Session(region_name=REGION)610 mock_aws = self.patch_object(self.rh, 'aws')611 mock_aws.get_session.return_value = session612 return session613 def assertSuccess(self):614 self.assertTrue(True)615 @mock_sqs616 def test_create_queue(self):617 self.configure()618 actual = self.rh.create_queue(self.name)619 self.assertEqual(self.url, actual)620 @mock_sqs621 def test_create_queue_already_exist(self):622 self.configure()623 # SQS create_queue will not error if the same queue tries to be created624 first = self.rh.create_queue(self.downsample_id)625 second = self.rh.create_queue(self.downsample_id)626 self.assertEqual(first, second)627 @mock_sqs628 def test_delete_queue(self):629 session = self.configure()630 session.client('sqs').create_queue(QueueName = self.name)631 self.rh.delete_queue(self.url)632 self.assertSuccess()633 @mock_sqs634 def test_delete_queue_doesnt_exist(self):635 self.configure()636 # delete_queue logs the error637 self.rh.delete_queue(self.url)638 self.assertSuccess()639 @mock_sqs640 def test_check_queue_doesnt_exist(self):641 self.configure()642 count = self.rh.check_queue(self.url)643 self.assertEqual(count, 0)644 @mock_sqs645 def test_check_queue_empty(self):646 self.configure()647 arn = self.rh.create_queue(self.downsample_id)648 count = self.rh.check_queue(arn)649 self.assertEqual(count, 0)650 @mock_sqs651 def test_check_queue_non_empty(self):652 session = self.configure()653 arn = self.rh.create_queue(self.downsample_id)654 session.client('sqs').send_message(QueueUrl = arn,655 MessageBody = 'message')656 session.client('sqs').send_message(QueueUrl = arn,657 MessageBody = 'message')658 count = self.rh.check_queue(arn)659 self.assertEqual(count, 2)660@patch.object(SESSION, 'client', autospec=True)661class TestThrottle(unittest.TestCase):662 @classmethod663 def setUpClass(cls):664 # Import the code to be tested 665 import resolution_hierarchy as rh666 rh.POOL_SIZE = 2667 rh.MAX_LAMBDA_TIME = rh.timedelta()668 rh.datetime = DateTime669 rh.Pool = MultiprocessingPool670 cls.rh = rh671 cls.statistics = {672 'Datapoints': [{673 'SampleCount': 123674 }]675 }676 def mock_client(self, client, return_data=True):677 gms = client.return_value.get_metric_statistics678 if return_data is True:679 gms.return_value = self.statistics680 elif return_data is False:681 gms.return_value = {}682 elif return_data is None:683 gms.side_effect = Exception("Could not get metrics")684 def test_throttle_metric(self, mClient):685 self.mock_client(mClient, return_data=True)686 arn = "arn:aws:lambda:region:account:function:name"687 count = self.rh.lambda_throttle_count(arn)688 self.assertEqual(count, 123)689 actual = mClient.return_value.get_metric_statistics.mock_calls690 expected = [call(Namespace = 'AWS/Lambda',691 MetricName = 'Throttles',692 Dimensions = [{'Name': 'FunctionName', 'Value': 'name'}],693 StartTime = DateTime.now() - timedelta(minutes=1),694 EndTime = DateTime.now(),695 Period = 60,696 Unit = 'Count',697 Statistics = ['SampleCount'])]698 self.assertEqual(actual, expected)699 def test_throttle_no_metric(self, mClient):700 self.mock_client(mClient, return_data=False)701 count = self.rh.lambda_throttle_count('lambda_arn')702 self.assertEqual(count, 0)703 def test_throttle_error(self, mClient):704 self.mock_client(mClient, return_data=None)705 count = self.rh.lambda_throttle_count('lambda_arn')...
test_downsample_queue.py
Source:test_downsample_queue.py
...129 Configure an individual test.130 Doesn't use setUp() because this has to run inside the test method for131 proper mocking of AWS resources.132 Member variables set:133 sut (update_visibility_timeout()): Function being tested.134 url (str): Downsample queue URL.135 session (Session): Boto3 session.136 """137 import resolution_hierarchy as rh138 # System (function) under test.139 self.sut = rh.update_visibility_timeout140 self.session = boto3.session.Session(region_name=REGION)141 mock_aws = self.patch_object(rh, 'aws')142 mock_aws.get_session.return_value = self.session143 resp = self.session.client('sqs').create_queue(QueueName='fake-downsample-queue')144 self.url = resp['QueueUrl']145 @mock_sqs146 def test_update_timeout(self):147 self.configure()...
simplequeue.py
Source:simplequeue.py
...61 ret = db_queue.dequeue(name, 'system', visibility_timeout=visibility_timeout, session=dbsession)62 return (ret)63 except Exception as err:64 raise err65def update_visibility_timeout(name, receipt_handle, visibility_timeout):66 """67 Reset the visibility timeout of the message with given receipt handle to the given visibility timeout value68 :param name:69 :param receipt_handle:70 :param visibility_timeout:71 :return:72 """73 ret = {}74 queue = get_queue(name)75 if queue is None:76 return (None)77 try:78 with db.session_scope() as dbsession:79 ret = db_queue.update_visibility_by_handle(name, 'system', receipt_handle, visibility_timeout, session=dbsession)...
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?
