How to use _create_network_interfaces method in lisa

Best Python code snippet using lisa_python

platform_.py

Source:platform_.py

...566        instances = {}567        subnets = self._create_subnets(self._vpc.id, deployment_parameters, log)568        block_device_mappings = self._create_block_devices(deployment_parameters, log)569        for node in deployment_parameters.nodes:570            network_interfaces = self._create_network_interfaces(571                deployment_parameters, node, subnets, log572            )573            try:574                instance = ec2_resource.create_instances(575                    ImageId=node.get_image_id(),576                    InstanceType=cast(InstanceTypeType, node.vm_size),577                    NetworkInterfaces=network_interfaces,578                    BlockDeviceMappings=block_device_mappings,579                    KeyName=deployment_parameters.key_pair_name,580                    MinCount=1,581                    MaxCount=1,582                )[0]583                instance.wait_until_running()584                instance.load()585                log.info("Created instance %s.", instance.id)586                # Enable ENA support if the test case requires.587                # Don't support the Intel 82599 Virtual Function (VF) interface now.588                # Refer to the document about AWS Enhanced networking on Linux.589                if node.enable_sriov and (not instance.ena_support):590                    self._ec2_client.modify_instance_attribute(591                        InstanceId=instance.id,592                        EnaSupport={593                            "Value": True,594                        },595                    )596                instances[node.name] = instance.instance_id597            except ClientError:598                log.exception(599                    "Couldn't create instance with image %s, "600                    "instance type %s, and key %s.",601                    node.get_image_id(),602                    node.vm_size,603                    deployment_parameters.key_pair_name,604                )605                raise606        return instances607    def _delete_environment(self, environment: Environment, log: Logger) -> None:608        environment_context = get_environment_context(environment=environment)609        security_group_name = environment_context.security_group_name610        # the resource group name is empty when it is not deployed for some reasons,611        # like capability doesn't meet case requirement.612        if not security_group_name:613            return614        assert self._aws_runbook615        if not environment_context.security_group_is_created:616            log.info(617                f"skipped to delete security resource group: {security_group_name}, "618                f"as it's not created by this run."619            )620        elif self._aws_runbook.dry_run:621            log.info(622                f"skipped to delete security resource group: {security_group_name}, "623                f"as it's a dry run."624            )625        else:626            ec2_resource = boto3.resource("ec2")627            for node in environment.nodes.list():628                node_context = get_node_context(node)629                instance_id = node_context.instance_id630                self.terminate_instance(ec2_resource, instance_id, log)631            self.delete_security_group(632                ec2_resource,633                environment_context.security_group_id,634                environment_context.security_group_name,635                log,636            )637            self.delete_key_pair(ec2_resource, environment_context.key_pair_name, log)638            try:639                log.info(f"deleting vpc: {self._vpc.id}")640                for association in self._route_table.associations:641                    association.delete()642                self._route_table.delete()643                self._internet_gateway.detach_from_vpc(VpcId=self._vpc.id)644                self._internet_gateway.delete()645                for subnet in self._vpc.subnets.all():646                    subnet.delete()647                self._vpc.delete()648            except ClientError:649                log.exception(650                    "Couldn't delete vpc %s.",651                    self._vpc.id,652                )653                raise654    def terminate_instance(655        self, ec2_resource: Any, instance_id: str, log: Logger656    ) -> None:657        if not instance_id:658            return659        try:660            instance = ec2_resource.Instance(instance_id)661            instance.terminate()662            instance.wait_until_terminated()663            log.info("Terminating instance %s.", instance_id)664        except ClientError:665            log.exception("Couldn't terminate instance %s.", instance_id)666    def delete_security_group(667        self, ec2_resource: Any, group_id: str, security_group_name: str, log: Logger668    ) -> None:669        try:670            ec2_resource.SecurityGroup(group_id).delete()671            log.info("Deleting security group: %s.", security_group_name)672        except ClientError:673            log.exception(674                "Couldn't delete security group %s.",675                security_group_name,676            )677    def delete_key_pair(self, ec2_resource: Any, key_name: str, log: Logger) -> None:678        try:679            ec2_resource.KeyPair(key_name).delete()680            log.info("Deleted key pair %s.", key_name)681        except ClientError:682            log.exception("Couldn't delete key pair %s.", key_name)683    def _create_subnets(684        self, vpc_id: str, deployment_parameters: AwsDeployParameter, log: Logger685    ) -> Dict[int, Any]:686        subnets: Dict[int, Any] = {}687        try:688            addrs = self._vpc.cidr_block.split(".")689            for i in range(deployment_parameters.subnet_count):690                cidr_block = f"{addrs[0]}.{addrs[1]}.{str(i)}.0/24"691                subnets[i] = self._ec2_client.create_subnet(692                    CidrBlock=cidr_block,693                    VpcId=vpc_id,694                )695                self._route_table.associate_with_subnet(696                    SubnetId=subnets[i]["Subnet"]["SubnetId"]697                )698        except ClientError:699            log.exception("Could not create a custom subnet.")700            raise701        else:702            return subnets703    def _create_network_interfaces(704        self,705        deployment_parameters: AwsDeployParameter,706        node: AwsNodeSchema,707        subnets: Dict[int, Any],708        log: Logger,709    ) -> List[Any]:710        network_interfaces = [711            {712                "Description": f"{node.name}-extra-0",713                "AssociatePublicIpAddress": True,714                "SubnetId": subnets[0]["Subnet"]["SubnetId"],715                "DeviceIndex": 0,716                "Groups": [deployment_parameters.security_group_id],717            }...

profitbricks.py

Source:profitbricks.py

...111            raise self.InvalidCloudSettingsException(112                'The target_cloud settings in the migration plan are invalid!'113            )114        self._wait_for_request(response['requestId'])115        self._create_network_interfaces(response['id'], network_interfaces)116        return {117            'id': response['id'],118            'volumes': [119                {120                    'id': volume['id'],121                    'device_number': volume['properties']['deviceNumber'],122                    'size': volume['properties']['size'],123                    'name': volume['properties']['name'],124                }125                for volume in self._client.get_attached_volumes(126                    datacenter_id=self._datacenter, server_id=response['id']127                )['items']128            ],129            'network_interfaces': [130                {131                    'id': network_interface['id'],132                    'name': network_interface['properties']['name'],133                    'lan': network_interface['properties']['lan'],134                    'ip': network_interface['properties']['ips'][0],135                }136                for network_interface in self._client.list_nics(137                    datacenter_id=self._datacenter, server_id=response['id']138                )['items']139            ],140        }141    def delete_volume(self, volume_id):142        return self._client.delete_volume(datacenter_id=self._datacenter, volume_id=volume_id)143    def make_volume_boot(self, server_id, volume_id):144        response = self._client.update_server(145            # According to the profitbricks api docs, the boot_volume field should be called boot_volume_id, but this146            # only works with boot_volume. So keep an eye out, whether this will change in future releases!147            datacenter_id=self._datacenter, server_id=server_id, boot_volume=volume_id148        )149        self._wait_for_request(response['requestId'])150        return response151    def delete_nic(self, server_id, nic_id):152        return self._client.delete_nic(datacenter_id=self._datacenter, server_id=server_id, nic_id=nic_id)153    def _create_network_interfaces(self, server_id, network_interfaces):154        """155        creates the network interfaces for a server156        :param server_id: the id of the server to create the network interfaces for157        :type server_id: str158        :param network_interfaces: the network interfaces to create159        :type network_interfaces: [{'ip: str, 'network_id': str}]160        :return:161        """162        try:163            # noinspection PyTypeChecker164            for create_network_interface_request_id in [165                self._client.create_nic(166                    datacenter_id=self._datacenter,167                    server_id=server_id,...

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.