KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use template_resources method in localstack
Best Python code snippet using localstack_python
test_sam_template_validator.py
Source:test_sam_template_validator.py 
1from unittest import TestCase2from unittest.mock import Mock, patch3from samcli.lib.utils.packagetype import IMAGE4from samtranslator.public.exceptions import InvalidDocumentException5from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException6from samcli.commands.validate.lib.sam_template_validator import SamTemplateValidator7class TestSamTemplateValidator(TestCase):8 @patch("samcli.commands.validate.lib.sam_template_validator.Session")9 @patch("samcli.commands.validate.lib.sam_template_validator.Translator")10 @patch("samcli.commands.validate.lib.sam_template_validator.parser")11 def test_is_valid_returns_true(self, sam_parser, sam_translator, boto_session_patch):12 managed_policy_mock = Mock()13 managed_policy_mock.load.return_value = {"policy": "SomePolicy"}14 template = {"a": "b"}15 parser = Mock()16 sam_parser.Parser.return_value = parser17 boto_session_mock = Mock()18 boto_session_patch.return_value = boto_session_mock19 translate_mock = Mock()20 translate_mock.translate.return_value = {"c": "d"}21 sam_translator.return_value = translate_mock22 validator = SamTemplateValidator(template, managed_policy_mock, profile="profile", region="region")23 # Should not throw an Exception24 validator.is_valid()25 boto_session_patch.assert_called_once_with(profile_name="profile", region_name="region")26 sam_translator.assert_called_once_with(27 managed_policy_map={"policy": "SomePolicy"}, sam_parser=parser, plugins=[], boto_session=boto_session_mock28 )29 translate_mock.translate.assert_called_once_with(sam_template=template, parameter_values={})30 sam_parser.Parser.assert_called_once()31 @patch("samcli.commands.validate.lib.sam_template_validator.Session")32 @patch("samcli.commands.validate.lib.sam_template_validator.Translator")33 @patch("samcli.commands.validate.lib.sam_template_validator.parser")34 def test_is_valid_raises_exception(self, sam_parser, sam_translator, boto_session_patch):35 managed_policy_mock = Mock()36 managed_policy_mock.load.return_value = {"policy": "SomePolicy"}37 template = {"a": "b"}38 parser = Mock()39 sam_parser.Parser.return_value = parser40 boto_session_mock = Mock()41 boto_session_patch.return_value = boto_session_mock42 translate_mock = Mock()43 translate_mock.translate.side_effect = InvalidDocumentException([Exception("message")])44 sam_translator.return_value = translate_mock45 validator = SamTemplateValidator(template, managed_policy_mock)46 with self.assertRaises(InvalidSamDocumentException):47 validator.is_valid()48 sam_translator.assert_called_once_with(49 managed_policy_map={"policy": "SomePolicy"}, sam_parser=parser, plugins=[], boto_session=boto_session_mock50 )51 boto_session_patch.assert_called_once_with(profile_name=None, region_name=None)52 translate_mock.translate.assert_called_once_with(sam_template=template, parameter_values={})53 sam_parser.Parser.assert_called_once()54 def test_init(self):55 managed_policy_mock = Mock()56 template = {"a": "b"}57 validator = SamTemplateValidator(template, managed_policy_mock)58 self.assertEqual(validator.managed_policy_loader, managed_policy_mock)59 self.assertEqual(validator.sam_template, template)60 # check to see if SamParser was created61 self.assertIsNotNone(validator.sam_parser)62 def test_uri_is_s3_uri(self):63 self.assertTrue(SamTemplateValidator.is_s3_uri("s3://bucket/key"))64 def test_uri_is_not_s3_uri(self):65 self.assertFalse(SamTemplateValidator.is_s3_uri("www.amazon.com"))66 def test_int_is_not_s3_uri(self):67 self.assertFalse(SamTemplateValidator.is_s3_uri(100))68 def test_update_to_s3_uri_with_non_s3_uri(self):69 property_value = {"CodeUri": "somevalue"}70 SamTemplateValidator._update_to_s3_uri("CodeUri", property_value)71 self.assertEqual(property_value.get("CodeUri"), "s3://bucket/value")72 def test_update_to_s3_url_with_dict(self):73 property_value = {"CodeUri": {"Bucket": "mybucket-name", "Key": "swagger", "Version": 121212}}74 SamTemplateValidator._update_to_s3_uri("CodeUri", property_value)75 self.assertEqual(76 property_value.get("CodeUri"), {"Bucket": "mybucket-name", "Key": "swagger", "Version": 121212}77 )78 def test_update_to_s3_url_with_s3_uri(self):79 property_value = {"CodeUri": "s3://bucket/key/version"}80 SamTemplateValidator._update_to_s3_uri("CodeUri", property_value)81 self.assertEqual(property_value.get("CodeUri"), "s3://bucket/key/version")82 def test_replace_local_codeuri(self):83 template = {84 "AWSTemplateFormatVersion": "2010-09-09",85 "Transform": "AWS::Serverless-2016-10-31",86 "Resources": {87 "ServerlessApi": {88 "Type": "AWS::Serverless::Api",89 "Properties": {"StageName": "Prod", "DefinitionUri": "./"},90 },91 "ServerlessFunction": {92 "Type": "AWS::Serverless::Function",93 "Properties": {"Handler": "index.handler", "CodeUri": "./", "Runtime": "nodejs6.10", "Timeout": 60},94 },95 "ServerlessLayerVersion": {"Type": "AWS::Serverless::LayerVersion", "Properties": {"ContentUri": "./"}},96 "ServerlessStateMachine": {97 "Type": "AWS::Serverless::StateMachine",98 "Properties": {"DefinitionUri": "./", "Role": "test-role-arn"},99 },100 },101 }102 managed_policy_mock = Mock()103 validator = SamTemplateValidator(template, managed_policy_mock)104 validator._replace_local_codeuri()105 # check template106 template_resources = validator.sam_template.get("Resources")107 self.assertEqual(108 template_resources.get("ServerlessApi").get("Properties").get("DefinitionUri"), "s3://bucket/value"109 )110 self.assertEqual(111 template_resources.get("ServerlessFunction").get("Properties").get("CodeUri"), "s3://bucket/value"112 )113 self.assertEqual(114 template_resources.get("ServerlessLayerVersion").get("Properties").get("ContentUri"), "s3://bucket/value"115 )116 self.assertEqual(117 template_resources.get("ServerlessStateMachine").get("Properties").get("DefinitionUri"), "s3://bucket/value"118 )119 def test_replace_local_codeuri_when_no_codeuri_given(self):120 template = {121 "AWSTemplateFormatVersion": "2010-09-09",122 "Transform": "AWS::Serverless-2016-10-31",123 "Resources": {124 "ServerlessApi": {"Type": "AWS::Serverless::Api", "Properties": {"StageName": "Prod"}},125 "ServerlessFunction": {126 "Type": "AWS::Serverless::Function",127 "Properties": {"Handler": "index.handler", "Runtime": "nodejs6.10", "Timeout": 60},128 },129 },130 }131 managed_policy_mock = Mock()132 validator = SamTemplateValidator(template, managed_policy_mock)133 validator._replace_local_codeuri()134 # check template135 tempalte_resources = validator.sam_template.get("Resources")136 self.assertEqual(137 tempalte_resources.get("ServerlessFunction").get("Properties").get("CodeUri"), "s3://bucket/value"138 )139 def test_dont_replace_local_codeuri_when_no_codeuri_given_packagetype_image(self):140 template = {141 "AWSTemplateFormatVersion": "2010-09-09",142 "Transform": "AWS::Serverless-2016-10-31",143 "Resources": {144 "ServerlessApi": {"Type": "AWS::Serverless::Api", "Properties": {"StageName": "Prod"}},145 "ServerlessFunction": {146 "Type": "AWS::Serverless::Function",147 "Properties": {"PackageType": IMAGE, "ImageUri": "myimage:latest", "Timeout": 60},148 },149 },150 }151 managed_policy_mock = Mock()152 validator = SamTemplateValidator(template, managed_policy_mock)153 validator._replace_local_codeuri()154 # check template155 template_resources = validator.sam_template.get("Resources")156 self.assertEqual(157 template_resources.get("ServerlessFunction").get("Properties").get("CodeUri", "NotPresent"), "NotPresent"158 )159 def test_dont_replace_codeuri_when_global_code_uri_given_packagetype_image(self):160 template = {161 "AWSTemplateFormatVersion": "2010-09-09",162 "Transform": "AWS::Serverless-2016-10-31",163 "Globals": {"Function": {"CodeUri": "globalcodeuri", "Timeout": "3"}},164 "Resources": {165 "ServerlessApi": {"Type": "AWS::Serverless::Api", "Properties": {"StageName": "Prod"}},166 "ServerlessFunction": {167 "Type": "AWS::Serverless::Function",168 "Properties": {"PackageType": IMAGE, "ImageUri": "myimage:latest", "Timeout": 60},169 },170 },171 }172 managed_policy_mock = Mock()173 validator = SamTemplateValidator(template, managed_policy_mock)174 validator._replace_local_codeuri()175 # check template176 template_resources = validator.sam_template.get("Resources")177 self.assertEqual(178 template_resources.get("ServerlessFunction").get("Properties").get("CodeUri", "NotPresent"), "NotPresent"179 )180 def test_dont_replace_codeuri_when_global_code_uri_given__both_packagetype(self):181 template = {182 "AWSTemplateFormatVersion": "2010-09-09",183 "Transform": "AWS::Serverless-2016-10-31",184 "Globals": {185 "Function": {186 "CodeUri": "s3://globalcodeuri",187 }188 },189 "Resources": {190 "ServerlessApi": {"Type": "AWS::Serverless::Api", "Properties": {"StageName": "Prod"}},191 "ServerlessFunctionImage": {192 "Type": "AWS::Serverless::Function",193 "Properties": {"PackageType": IMAGE, "ImageUri": "myimage:latest", "Timeout": 60},194 },195 "ServerlessFunctionZip": {196 "Type": "AWS::Serverless::Function",197 "Properties": {"Handler": "index.handler", "Runtime": "nodejs6.10", "Timeout": 60},198 },199 },200 }201 managed_policy_mock = Mock()202 validator = SamTemplateValidator(template, managed_policy_mock)203 validator._replace_local_codeuri()204 # check template205 template_resources = validator.sam_template.get("Resources")206 self.assertEqual(207 template_resources.get("ServerlessFunctionImage").get("Properties").get("CodeUri", "NotPresent"),208 "NotPresent",209 )210 # Globals not set since they cant apply to both Zip and Image based packagetypes.211 self.assertEqual(212 template_resources.get("ServerlessFunctionZip").get("Properties").get("CodeUri"), "s3://bucket/value"213 )214 def test_DefinitionUri_does_not_get_added_to_template_when_DefinitionBody_given(self):215 template = {216 "AWSTemplateFormatVersion": "2010-09-09",217 "Transform": "AWS::Serverless-2016-10-31",218 "Resources": {219 "ServerlessApi": {220 "Type": "AWS::Serverless::Api",221 "Properties": {"StageName": "Prod", "DefinitionBody": {"swagger": {}}},222 }223 },224 }225 managed_policy_mock = Mock()226 validator = SamTemplateValidator(template, managed_policy_mock)227 validator._replace_local_codeuri()228 tempalte_resources = validator.sam_template.get("Resources")229 self.assertNotIn("DefinitionUri", tempalte_resources.get("ServerlessApi").get("Properties"))230 self.assertIn("DefinitionBody", tempalte_resources.get("ServerlessApi").get("Properties"))231 def test_replace_local_codeuri_with_no_resources(self):232 template = {233 "AWSTemplateFormatVersion": "2010-09-09",234 "Transform": "AWS::Serverless-2016-10-31",235 "Resources": {},236 }237 managed_policy_mock = Mock()238 validator = SamTemplateValidator(template, managed_policy_mock)239 validator._replace_local_codeuri()240 # check template...
scenario_base.py
Source:scenario_base.py
1#2# Copyright 2014-2017 AT&T Intellectual Property3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Scenario Base."""16import os17from tempest import config18from tempest import exceptions19from tempest import test20from tempest_lib.common.utils import data_utils21import time22import traceback23from valet.tests.tempest.scenario.analyzer import Analyzer24from valet.tests.tempest.scenario.resources import TemplateResources25from valet.tests.tempest.services.client import ValetClient26CONF = config.CONF27class ScenarioTestCase(test.BaseTestCase):28 """Base class for Scenario Test cases."""29 credentials = ['primary']30 @classmethod31 def skip_checks(cls):32 """Skip checks, if valet service not available, raise exception."""33 super(ScenarioTestCase, cls).skip_checks()34 if not CONF.service_available.valet:35 skip_msg = ("%s skipped as valet is not available" % cls.__name__)36 raise cls.skipException(skip_msg)37 @classmethod38 def resource_setup(cls):39 """Setup resource, set catalog_type."""40 super(ScenarioTestCase, cls).resource_setup()41 cls.catalog_type = CONF.placement.catalog_type42 @classmethod43 def resource_cleanup(cls):44 """Class method resource cleanup."""45 super(ScenarioTestCase, cls).resource_cleanup()46 @classmethod47 def setup_clients(cls):48 """Setup clients (valet)."""49 super(ScenarioTestCase, cls).setup_clients()50 cls.heat_client = cls.os.orchestration_client51 cls.nova_client = cls.os.servers_client52 cls.tenants_client = cls.os.identity_client53 cls.valet_client = ValetClient(54 cls.os.auth_provider, CONF.placement.catalog_type,55 CONF.identity.region, **cls.os.default_params_with_timeout_values)56 cls.possible_topdir = os.path.normpath(57 os.path.join(os.path.abspath(__file__), os.pardir))58 cls.stack_identifier = None59 cls.tries = CONF.valet.TRIES_TO_CREATE60 def run_test(self, logger, stack_name, template_path):61 """Scenario.62 create new stack63 checks if host (or rack) is the same for all instances64 """65 self.log = logger66 self.log.log_info(" ******** Running Test ******** ")67 tmplt_url = self.possible_topdir + template_path68 template = TemplateResources(tmplt_url)69 env_data = self.get_env_file(tmplt_url)70 self.log.log_info(" ******** Creating Stack ******** ")71 name = data_utils.rand_name(name=stack_name)72 self.assertEqual(True, self.create_stack(name, env_data, template))73 self.log.log_info(" ******** Analyzing Stack ******** ")74 analyzer = Analyzer(self.log, self.stack_identifier, self.heat_client,75 self.nova_client)76 self.assertEqual(True, analyzer.check(template))77 self.log.log_info(" ********** THE END ****************")78 def create_stack(self, stack_name, env_data, template_resources):79 """Create stack with name/env/resource. Create all groups/instances."""80 try:81 groups = template_resources.groups82 for key in groups:83 if groups[key].group_type == "exclusivity":84 self.log.log_info(" creating valet group ")85 grp_name = data_utils.rand_name(name=groups[key].group_name)86 template_resources.template_data = \87 template_resources.template_data.replace(88 groups[key].group_name, grp_name)89 self.create_valet_group(grp_name)90 for instance in template_resources.instances:91 generated_name = data_utils.rand_name(instance.name)92 template_resources.template_data = \93 template_resources.template_data.replace(94 instance.name, generated_name)95 instance.name = generated_name96 res = self.wait_for_stack(stack_name, env_data, template_resources)97 self.addCleanup(self.delete_stack)98 return res99 except Exception:100 self.log.log_error("Failed to prepare stack for creation", traceback.format_exc())101 return False102 return True103 def create_valet_group(self, group_name):104 """Create valet group with name using valet client. Add members."""105 try:106 v_group = self.valet_client.create_group(name=group_name,107 group_type='exclusivity',108 description="description")109 group_id = v_group['id']110 tenant_id = self.tenants_client.tenant_id111 self.addCleanup(self._delete_group, group_id)112 self.valet_client.add_members(group_id, [tenant_id])113 except Exception:114 self.log.log_error("Failed to create valet group",115 traceback.format_exc())116 raise117 def get_env_file(self, template):118 try:119 env_url = template.replace(".yml", ".env")120 self.log.log_debug("loading environment file (%s)" % env_url)121 if os.path.exists(env_url):122 with open(env_url, "r") as f:123 filedata = f.read()124 filedata = filedata.replace('image_place_holder', CONF.compute.image_ref)125 filedata = filedata.replace('flavor_place_holder', CONF.compute.flavor_ref)126 filedata = filedata.replace('network_place_holder', CONF.compute.fixed_network_name)127 return filedata128 else:129 return None130 except Exception:131 self.log.log_error("Failed to load environment file", traceback.format_exc())132 def _delete_group(self, group_id):133 try:134 self.valet_client.delete_all_members(group_id)135 self.valet_client.delete_group(group_id)136 except Exception:137 self.log.log_error("Failed to delete group", traceback.format_exc())138 raise139 def delete_stack(self):140 """Use heat client to delete stack."""141 try:142 self.heat_client.delete_stack(self.stack_identifier)143 self.heat_client.wait_for_stack_status(144 self.stack_identifier, "DELETE_COMPLETE",145 failure_pattern='^.*DELETE_FAILED$')146 except Exception:147 self.log.log_error("Failed to delete stack", traceback.format_exc())148 raise149 def show_stack(self, stack_id):150 """Return show stack with given id from heat client."""151 return self.heat_client.show_stack(stack_id)152 def wait_for_stack(self, stack_name, env_data, template_resources):153 """Use heat client to create stack, then wait for status."""154 try:155 self.log.log_info("Trying to create stack "156 "Waiting for stack status")157 new_stack = self.heat_client.create_stack(158 stack_name, environment=env_data,159 template=template_resources.template_data)160 stack_id = new_stack["stack"]["id"]161 self.stack_identifier = stack_name + "/" + stack_id162 self.heat_client.wait_for_stack_status(163 self.stack_identifier, "CREATE_COMPLETE",164 failure_pattern='^.*CREATE_FAILED$')165 except exceptions.StackBuildErrorException as ex:166 if "Ostro error" in str(ex) and self.tries > 0:167 self.log.log_error("Ostro error - try number %d" %168 (CONF.valet.TRIES_TO_CREATE - self.tries + 2))169 self.tries -= 1170 self.delete_stack()171 time.sleep(CONF.valet.PAUSE)172 self.wait_for_stack(stack_name, env_data, template_resources)173 else:174 self.log.log_error("Failed to create stack", traceback.format_exc())175 return False...
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?
