How to use get_physical_resource_id method in localstack

Best Python code snippet using localstack_python

test_create_update_neutron_trunk.py

Source:test_create_update_neutron_trunk.py

...94            'sub_ports'] = new_sub_port95        updated_template = yaml.safe_dump(parsed_template)96        self.update_stack(stack_identifier, updated_template)97        # Fix the port_id in the template for assertion98        new_sub_port[0]['port'] = self.get_physical_resource_id(99            stack_identifier, 'sub_port_one')100        parent_id = self.get_stack_output(101            stack_identifier, 'trunk_parent_port')102        parent_port = self.network_client.show_port(parent_id)['port']103        trunk_sub_port = parent_port['trunk_details']['sub_ports']104        self.assertEqual(self._sub_ports_dict_to_set(new_sub_port),105                         self._sub_ports_dict_to_set(trunk_sub_port))106    def test_add_a_second_sub_port(self):107        parsed_template = yaml.safe_load(test_template)108        sub_ports = [{'port': {'get_resource': 'sub_port_one'},109                      'segmentation_type': 'vlan',110                      'segmentation_id': 10}, ]111        parsed_template['resources']['trunk']['properties'][112            'sub_ports'] = sub_ports113        template_with_sub_ports = yaml.safe_dump(parsed_template)114        stack_identifier = self.stack_create(template=template_with_sub_ports)115        new_sub_port = {'port': {'get_resource': 'sub_port_two'},116                        'segmentation_id': 20,117                        'segmentation_type': 'vlan'}118        parsed_template['resources']['trunk']['properties'][119            'sub_ports'].append(new_sub_port)120        updated_template = yaml.safe_dump(parsed_template)121        self.update_stack(stack_identifier, updated_template)122        # Fix the port_ids in the templates for assertion123        sub_ports[0]['port'] = self.get_physical_resource_id(124            stack_identifier, 'sub_port_one')125        new_sub_port['port'] = self.get_physical_resource_id(126            stack_identifier, 'sub_port_two')127        expected_sub_ports = [sub_ports[0], new_sub_port]128        parent_id = self.get_stack_output(129            stack_identifier, 'trunk_parent_port')130        parent_port = self.network_client.show_port(parent_id)['port']131        trunk_sub_ports = parent_port['trunk_details']['sub_ports']132        self.assertEqual(self._sub_ports_dict_to_set(expected_sub_ports),133                         self._sub_ports_dict_to_set(trunk_sub_ports))134    def test_remove_sub_port_from_trunk(self):135        sub_ports = [{'port': {'get_resource': 'sub_port_one'},136                      'segmentation_type': 'vlan',137                      'segmentation_id': 10},138                     {'port': {'get_resource': 'sub_port_two'},139                      'segmentation_type': 'vlan',140                      'segmentation_id': 20}]141        parsed_template = yaml.safe_load(test_template)142        parsed_template['resources']['trunk']['properties'][143            'sub_ports'] = sub_ports144        template_with_sub_ports = yaml.safe_dump(parsed_template)145        stack_identifier = self.stack_create(template=template_with_sub_ports)146        sub_port_to_be_removed = {'port': {'get_resource': 'sub_port_two'},147                                  'segmentation_type': 'vlan',148                                  'segmentation_id': 20}149        parsed_template['resources']['trunk'][150            'properties']['sub_ports'].remove(sub_port_to_be_removed)151        updated_template = yaml.safe_dump(parsed_template)152        self.update_stack(stack_identifier, updated_template)153        # Fix the port_ids in the templates for assertion154        sub_ports[0]['port'] = self.get_physical_resource_id(155            stack_identifier, 'sub_port_one')156        expected_sub_ports = [sub_ports[0]]157        parent_id = self.get_stack_output(158            stack_identifier, 'trunk_parent_port')159        parent_port = self.network_client.show_port(parent_id)['port']160        trunk_sub_ports = parent_port['trunk_details']['sub_ports']161        self.assertEqual(self._sub_ports_dict_to_set(expected_sub_ports),162                         self._sub_ports_dict_to_set(trunk_sub_ports))163    def test_remove_last_sub_port_from_trunk(self):164        sub_ports = [{'port': {'get_resource': 'sub_port_one'},165                      'segmentation_type': 'vlan',166                      'segmentation_id': 10}]167        parsed_template = yaml.safe_load(test_template)168        parsed_template['resources']['trunk']['properties'][169            'sub_ports'] = sub_ports170        template_with_sub_ports = yaml.safe_dump(parsed_template)171        stack_identifier = self.stack_create(template=template_with_sub_ports)172        sub_port_to_be_removed = {'port': {'get_resource': 'sub_port_one'},173                                  'segmentation_type': 'vlan',174                                  'segmentation_id': 10}175        parsed_template['resources']['trunk'][176            'properties']['sub_ports'] = []177        updated_template = yaml.safe_dump(parsed_template)178        self.update_stack(stack_identifier, updated_template)179        sub_port_to_be_removed['port'] = self.get_physical_resource_id(180            stack_identifier, 'sub_port_one')181        parent_id = self.get_stack_output(182            stack_identifier, 'trunk_parent_port')183        parent_port = self.network_client.show_port(parent_id)['port']184        trunk_sub_ports = parent_port['trunk_details']['sub_ports']185        self.assertNotEqual(186            self._sub_ports_dict_to_set([sub_port_to_be_removed]),187            self._sub_ports_dict_to_set(trunk_sub_ports))188        self.assertFalse(trunk_sub_ports,189                         'The returned sub ports (%s) in trunk_details is '190                         'not empty!' % trunk_sub_ports)191    def test_update_existing_sub_port_on_trunk(self):192        sub_ports = [{'port': {'get_resource': 'sub_port_one'},193                      'segmentation_type': 'vlan',194                      'segmentation_id': 10}]195        parsed_template = yaml.safe_load(test_template)196        parsed_template['resources']['trunk']['properties'][197            'sub_ports'] = sub_ports198        template_with_sub_ports = yaml.safe_dump(parsed_template)199        stack_identifier = self.stack_create(template=template_with_sub_ports)200        sub_port_id = self.get_physical_resource_id(201            stack_identifier, 'sub_port_one')202        parsed_template['resources']['trunk']['properties']['sub_ports'][0][203            'segmentation_id'] = 99204        updated_template = yaml.safe_dump(parsed_template)205        self.update_stack(stack_identifier, updated_template)206        updated_sub_port = {'port': sub_port_id,207                            'segmentation_type': 'vlan',208                            'segmentation_id': 99}209        parent_id = self.get_stack_output(210            stack_identifier, 'trunk_parent_port')211        parent_port = self.network_client.show_port(parent_id)['port']212        trunk_sub_ports = parent_port['trunk_details']['sub_ports']213        self.assertEqual(self._sub_ports_dict_to_set([updated_sub_port]),214                         self._sub_ports_dict_to_set(trunk_sub_ports))...

utils.py

Source:utils.py

...5log = logging.getLogger()6log.setLevel(logging.INFO)7# global variables8ec2_client = boto3.client('ec2')9def get_physical_resource_id(request_id):10    return hashlib.md5(request_id.encode()).hexdigest()11def get_network_interface_by_id(eni_id):12    """13    Use describe network interfaces function instead of ec2_resource.NetworkInterface14    :param eni_id:15    :return:16    """17    response = ec2_client.describe_network_interfaces(18        Filters=[{19            "Name": "network-interface-id",20            "Values": [eni_id]21        }]22    )23    if len(response["NetworkInterfaces"]) == 1:24        return response["NetworkInterfaces"][0]25    else:26        log.error("Get network interface by id %s failed: %s" % (eni_id, str(response)))27def cfn_success_response(event, reuse_physical_id=False, data=None):28    return {29        "Status": "SUCCESS",30        "RequestId": event["RequestId"],31        "LogicalResourceId": event["LogicalResourceId"],32        "StackId": event["StackId"],33        "PhysicalResourceId":34            event["PhysicalResourceId"]35            if reuse_physical_id36            else get_physical_resource_id(event["RequestId"]),37        "Data": {} if not data else data38    }39def cfn_failure_response(event, reason, data=None):40    return {41        "Status": "FAILED",42        "Reason": reason,43        "LogicalResourceId": event["LogicalResourceId"],44        "StackId": event["StackId"],45        "PhysicalResourceId": get_physical_resource_id(event["RequestId"]),46        "Data": {} if not data else data...

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.