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

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.