Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use get_local_region method in localstack
Best Python code snippet using localstack_python
aws_stack.py
Source:aws_stack.py 
...29class Environment(object):30 def __init__(self, region=None, prefix=None):31 # target is the runtime environment to use, e.g.,32 # 'local' for local mode33 self.region = region or get_local_region()34 # prefix can be 'prod', 'stg', 'uat-1', etc.35 self.prefix = prefix36 def apply_json(self, j):37 if isinstance(j, str):38 j = json.loads(j)39 self.__dict__.update(j)40 @staticmethod41 def from_string(s):42 parts = s.split(':')43 if len(parts) == 1:44 if s in PREDEFINED_ENVIRONMENTS:45 return PREDEFINED_ENVIRONMENTS[s]46 parts = [get_local_region(), s]47 if len(parts) > 2:48 raise Exception('Invalid environment string "%s"' % s)49 region = parts[0]50 prefix = parts[1]51 return Environment(region=region, prefix=prefix)52 @staticmethod53 def from_json(j):54 if not isinstance(j, dict):55 j = j.to_dict()56 result = Environment()57 result.apply_json(j)58 return result59 def __str__(self):60 return '%s:%s' % (self.region, self.prefix)61PREDEFINED_ENVIRONMENTS = {62 ENV_DEV: Environment(region=REGION_LOCAL, prefix=ENV_DEV)63}64def get_environment(env=None, region_name=None):65 """66 Return an Environment object based on the input arguments.67 Parameter `env` can be either of:68 * None (or empty), in which case the rules below are applied to (env = os.environ['ENV'] or ENV_DEV)69 * an Environment object (then this object is returned)70 * a string '<region>:<name>', which corresponds to Environment(region='<region>', prefix='<prefix>')71 * the predefined string 'dev' (ENV_DEV), which implies Environment(region='local', prefix='dev')72 * a string '<name>', which implies Environment(region=DEFAULT_REGION, prefix='<name>')73 Additionally, parameter `region_name` can be used to override DEFAULT_REGION.74 """75 if not env:76 if 'ENV' in os.environ:77 env = os.environ['ENV']78 else:79 env = ENV_DEV80 elif not is_string(env) and not isinstance(env, Environment):81 raise Exception('Invalid environment: %s' % env)82 if is_string(env):83 env = Environment.from_string(env)84 if region_name:85 env.region = region_name86 if not env.region:87 raise Exception('Invalid region in environment: "%s"' % env)88 return env89def connect_to_resource(service_name, env=None, region_name=None, endpoint_url=None):90 """91 Generic method to obtain an AWS service resource using boto3, based on environment, region, or custom endpoint_url.92 """93 return connect_to_service(service_name, client=False, env=env, region_name=region_name, endpoint_url=endpoint_url)94def get_boto3_credentials():95 if CUSTOM_BOTO3_SESSION:96 return CUSTOM_BOTO3_SESSION.get_credentials()97 return boto3.session.Session().get_credentials()98def get_boto3_session():99 if CUSTOM_BOTO3_SESSION:100 return CUSTOM_BOTO3_SESSION101 if CREATE_NEW_SESSION_PER_BOTO3_CONNECTION:102 return boto3.session.Session()103 # return default session104 return boto3105def get_local_region():106 session = boto3.session.Session()107 return session.region_name or DEFAULT_REGION108def get_local_service_url(service_name):109 if service_name == 's3api':110 service_name = 's3'111 return os.environ['TEST_%s_URL' % (service_name.upper().replace('-', '_'))]112def is_service_enabled(service_name):113 """ Return whether the service with the given name (e.g., "lambda") is available. """114 try:115 url = get_local_service_url(service_name)116 assert url117 return is_port_open(url, http_path='/', expect_success=False)118 except Exception:119 return False120def connect_to_service(service_name, client=True, env=None, region_name=None, endpoint_url=None, config=None):121 """122 Generic method to obtain an AWS service client using boto3, based on environment, region, or custom endpoint_url.123 """124 env = get_environment(env, region_name=region_name)125 my_session = get_boto3_session()126 method = my_session.client if client else my_session.resource127 verify = True128 if not endpoint_url:129 if env.region == REGION_LOCAL:130 endpoint_url = get_local_service_url(service_name)131 verify = False132 region = env.region if env.region != REGION_LOCAL else get_local_region()133 return method(service_name, region_name=region, endpoint_url=endpoint_url, verify=verify, config=config)134class VelocityInput:135 """Simple class to mimick the behavior of variable '$input' in AWS API Gateway integration velocity templates.136 See: http://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html"""137 def __init__(self, value):138 self.value = value139 def path(self, path):140 from jsonpath_rw import parse141 value = self.value if isinstance(self.value, dict) else json.loads(self.value)142 jsonpath_expr = parse(path)143 result = [match.value for match in jsonpath_expr.find(value)]144 result = result[0] if len(result) == 1 else result145 return result146 def json(self, path):147 return json.dumps(self.path(path))148class VelocityUtil:149 """Simple class to mimick the behavior of variable '$util' in AWS API Gateway integration velocity templates.150 See: http://docs.aws.amazon.com/apigateway/latest/developerguide/api-gateway-mapping-template-reference.html"""151 def base64Encode(self, s):152 if not isinstance(s, str):153 s = json.dumps(s)154 encoded_str = s.encode(config.DEFAULT_ENCODING)155 encoded_b64_str = base64.b64encode(encoded_str)156 return encoded_b64_str.decode(config.DEFAULT_ENCODING)157 def base64Decode(self, s):158 if not isinstance(s, str):159 s = json.dumps(s)160 return base64.b64decode(s)161def render_velocity_template(template, context, as_json=False):162 import airspeed163 t = airspeed.Template(template)164 variables = {165 'input': VelocityInput(context),166 'util': VelocityUtil()167 }168 replaced = t.merge(variables)169 if as_json:170 replaced = json.loads(replaced)171 return replaced172def check_valid_region(headers):173 """ Check whether a valid region is provided, and if not then raise an Exception. """174 auth_header = headers.get('Authorization')175 if not auth_header:176 raise Exception('Unable to find "Authorization" header in request')177 replaced = re.sub(r'.*Credential=([^,]+),.*', r'\1', auth_header)178 if auth_header == replaced:179 raise Exception('Unable to find "Credential" section in "Authorization" header')180 # Format is: <your-access-key-id>/<date>/<aws-region>/<aws-service>/aws4_request181 # See https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-auth-using-authorization-header.html182 parts = replaced.split('/')183 region = parts[2]184 if region not in config.VALID_REGIONS:185 raise Exception('Invalid region specified in "Authorization" header: "%s"' % region)186def get_s3_client():187 return boto3.resource('s3',188 endpoint_url=config.TEST_S3_URL,189 config=boto3.session.Config(190 s3={'addressing_style': 'path'}),191 verify=False)192def get_account_id(account_id=None, env=None):193 if account_id:194 return account_id195 env = get_environment(env)196 if env.region == REGION_LOCAL:197 return os.environ['TEST_AWS_ACCOUNT_ID']198 raise Exception('Unable to determine AWS account ID')199def role_arn(role_name, account_id=None, env=None):200 if not role_name:201 return role_name202 if role_name.startswith('arn:aws:iam::'):203 return role_name204 env = get_environment(env)205 account_id = get_account_id(account_id, env=env)206 return 'arn:aws:iam::%s:role/%s' % (account_id, role_name)207def iam_resource_arn(resource, role=None, env=None):208 env = get_environment(env)209 if not role:210 role = get_iam_role(resource, env=env)211 return role_arn(role_name=role, account_id=get_account_id())212def get_iam_role(resource, env=None):213 env = get_environment(env)214 return 'role-%s' % resource215def dynamodb_table_arn(table_name, account_id=None):216 account_id = get_account_id(account_id)217 return 'arn:aws:dynamodb:%s:%s:table/%s' % (get_local_region(), account_id, table_name)218def dynamodb_stream_arn(table_name, account_id=None):219 account_id = get_account_id(account_id)220 return ('arn:aws:dynamodb:%s:%s:table/%s/stream/%s' %221 (get_local_region(), account_id, table_name, timestamp()))222def lambda_function_arn(function_name, account_id=None):223 pattern = 'arn:aws:lambda:.*:.*:function:.*'224 if re.match(pattern, function_name):225 return function_name226 if ':' in function_name:227 raise Exception('Lambda function name should not contain a colon ":"')228 account_id = get_account_id(account_id)229 return pattern.replace('.*', '%s') % (get_local_region(), account_id, function_name)230def lambda_function_name(name_or_arn):231 if ':' not in name_or_arn:232 return name_or_arn233 parts = name_or_arn.split(':')234 # name is index #6 in pattern: arn:aws:lambda:.*:.*:function:.*235 return parts[6]236def state_machine_arn(name, account_id=None):237 if ':' in name:238 return name239 account_id = get_account_id(account_id)240 pattern = 'arn:aws:states:%s:%s:stateMachine:%s'241 return pattern % (get_local_region(), account_id, name)242def fix_arn(arn):243 """ Function that attempts to "canonicalize" the given ARN. This includes converting244 resource names to ARNs, replacing incorrect regions, account IDs, etc. """245 if arn.startswith('arn:aws:lambda'):246 return lambda_function_arn(lambda_function_name(arn))247 LOG.warning('Unable to fix/canonicalize ARN: %s' % arn)248 return arn249def cognito_user_pool_arn(user_pool_id, account_id=None):250 account_id = get_account_id(account_id)251 return 'arn:aws:cognito-idp:%s:%s:userpool/%s' % (get_local_region(), account_id, user_pool_id)252def kinesis_stream_arn(stream_name, account_id=None):253 account_id = get_account_id(account_id)254 return 'arn:aws:kinesis:%s:%s:stream/%s' % (get_local_region(), account_id, stream_name)255def firehose_stream_arn(stream_name, account_id=None):256 account_id = get_account_id(account_id)257 return ('arn:aws:firehose:%s:%s:deliverystream/%s' % (get_local_region(), account_id, stream_name))258def s3_bucket_arn(bucket_name, account_id=None):259 return 'arn:aws:s3:::%s' % (bucket_name)260def sqs_queue_arn(queue_name, account_id=None):261 account_id = get_account_id(account_id)262 # ElasticMQ sets a static region of "elasticmq"263 return ('arn:aws:sqs:elasticmq:%s:%s' % (account_id, queue_name))264def sns_topic_arn(topic_name, account_id=None):265 account_id = get_account_id(account_id)266 return ('arn:aws:sns:%s:%s:%s' % (get_local_region(), account_id, topic_name))267def get_sqs_queue_url(queue_name):268 client = connect_to_service('sqs')269 response = client.get_queue_url(QueueName=queue_name)270 return response['QueueUrl']271def dynamodb_get_item_raw(request):272 headers = mock_aws_request_headers()273 headers['X-Amz-Target'] = 'DynamoDB_20120810.GetItem'274 new_item = make_http_request(url=config.TEST_DYNAMODB_URL,275 method='POST', data=json.dumps(request), headers=headers)276 new_item = json.loads(new_item.text)277 return new_item278def mock_aws_request_headers(service='dynamodb'):279 ctype = APPLICATION_AMZ_JSON_1_0280 if service == 'kinesis':...
test_rclv.py
Source:test_rclv.py
...54 lat0 = -6055 verts_proj_60S = rclv.project_vertices(verts, lon0, lat0, 0.1, 0.1)56 np.testing.assert_allclose(rclv.polygon_area(verts_proj_60S),57 area_equator_expected/2, rtol=0.01)58def test_get_local_region():59 # create some data60 n = 1061 x, y = np.meshgrid(np.arange(n), np.arange(n))62 (j,i), x_reg = rclv.get_local_region(x, (2,2), border_j=(2,2), border_i=(2,2))63 assert x_reg.shape == (5,5)64 assert x_reg[j,i] == 065 assert x_reg[j,0] == 266 assert x_reg[j,-1] == -267 with pytest.raises(ValueError) as ve:68 (j,i), x_reg = rclv.get_local_region(x, (2,2), border_j=(3,2), border_i=(2,2))69 with pytest.raises(ValueError) as ve:70 (j,i), x_reg = rclv.get_local_region(x, (2,2), border_j=(2,8), border_i=(2,2))71 with pytest.raises(ValueError) as ve:72 (j,i), x_reg = rclv.get_local_region(x, (2,2), border_j=(2,2), border_i=(3,2))73 with pytest.raises(ValueError) as ve:74 (j,i), x_reg = rclv.get_local_region(x, (2,2), border_j=(2,2), border_i=(2,8))75def test_get_local_region_periodic():76 # create some data77 n = 1078 x, y = np.meshgrid(np.arange(n), np.arange(n))79 # check behavior for periodic in the i direction80 periodic=(False, True)81 _, x_reg = rclv.get_local_region(x, (2, 1), periodic=periodic,82 border_j=(2, 2), border_i=(2, 2))83 assert x_reg.shape == (5, 5)84 assert x_reg[0, 0] == -885 _, x_reg = rclv.get_local_region(x, (2, 8), periodic=periodic,86 border_j=(2, 2), border_i=(2 ,2))87 assert x_reg.shape == (5, 5)88 assert x_reg[0, -1] == 889 # check behavior for periodic in the j direction90 periodic=(True, False)91 _, y_reg = rclv.get_local_region(y, (1, 2), periodic=periodic,92 border_j=(2, 2), border_i=(2,2 ))93 assert y_reg.shape == (5, 5)94 assert y_reg[0, 0] == -895 _, y_reg = rclv.get_local_region(y, (8, 2), periodic=periodic,96 border_j=(2, 2), border_i=(2, 2))97 assert y_reg.shape == (5, 5)98 assert y_reg[-1, 0] == 899 # check behavior for doubly periodic, all four corners100 periodic = (True, True)101 # lower left102 ji = (1, 1)103 _, y_reg = rclv.get_local_region(y, ji, periodic=periodic,104 border_j=(2, 2), border_i=(2, 2))105 _, x_reg = rclv.get_local_region(x, ji, periodic=periodic,106 border_j=(2, 2), border_i=(2, 2))107 assert x_reg.shape == (5, 5)108 assert x_reg[0, 0] == -8109 assert y_reg.shape == (5, 5)110 assert y_reg[0, 0] == -8111 # lower right112 ji = (1, 8)113 _, y_reg = rclv.get_local_region(y, ji, periodic=periodic,114 border_j=(2, 2), border_i=(2, 2))115 _, x_reg = rclv.get_local_region(x, ji, periodic=periodic,116 border_j=(2, 2), border_i=(2, 2))117 assert x_reg.shape == (5, 5)118 assert x_reg[0, -1] == 8119 assert y_reg.shape == (5, 5)120 assert y_reg[0, 0] == -8121 # upper left122 ji = (8, 1)123 _, y_reg = rclv.get_local_region(y, ji, periodic=periodic,124 border_j=(2, 2), border_i=(2, 2))125 _, x_reg = rclv.get_local_region(x, ji, periodic=periodic,126 border_j=(2, 2), border_i=(2, 2))127 assert x_reg.shape == (5, 5)128 assert x_reg[0, 0] == -8129 assert y_reg.shape == (5, 5)130 assert y_reg[-1, 0] == 8131 # upper right132 ji = (8, 8)133 _, y_reg = rclv.get_local_region(y, ji, periodic=periodic,134 border_j=(2, 2), border_i=(2, 2))135 _, x_reg = rclv.get_local_region(x, ji, periodic=periodic,136 border_j=(2, 2), border_i=(2, 2))137 assert x_reg.shape == (5, 5)138 assert x_reg[0, -1] == 8139 assert y_reg.shape == (5, 5)140 assert y_reg[-1, 0] == 8141def test_is_contour_closed(square_verts):142 assert rclv.is_contour_closed(square_verts)143 assert not rclv.is_contour_closed(square_verts[:-1])144def test_point_in_contour(square_verts):145 assert rclv.point_in_contour(square_verts, (0., 0.))146 assert not rclv.point_in_contour(square_verts, (1., 0.))147def test_contour_area(square_verts):148 region_area, hull_area, convex_def = rclv.contour_area(square_verts)149 assert region_area == 1.0...
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?
