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

How to use _create_deployment_parameters method in lisa

Best Python code snippet using lisa_python

test_deploy_operation.py

Source:test_deploy_operation.py

...365        ami_model.custom_tags= ""366        network_actions = None367        vpc = Mock()368        self.deploy_operation._get_block_device_mappings = Mock()369        aws_model = self.deploy_operation._create_deployment_parameters(ec2_session=ec2_session,370                                                                        aws_ec2_resource_model=self.ec2_datamodel,371                                                                        ami_deployment_model=ami_model,372                                                                        network_actions=network_actions,373                                                                        vpc=vpc,374                                                                        security_group=None,375                                                                        key_pair='keypair',376                                                                        reservation=Mock(),377                                                                        network_config_results=MagicMock(),378                                                                        logger=self.logger)379        self.assertEquals(aws_model.min_count, 1)380        self.assertEquals(aws_model.max_count, 1)381        self.assertEquals(aws_model.aws_key, 'keypair')382        self.assertTrue(len(aws_model.security_group_ids) == 1)383        self.assertTrue(len(aws_model.network_interfaces) == 1)384    def test_create_deployment_parameters_no_iam_role(self):385        image = Mock()386        image.state = 'available'387        ec2_session = Mock()388        ec2_session.Image = Mock(return_value=image)389        ami_model = Mock()390        ami_model.iam_role = ""391        ami_model.custom_tags = ""392        network_actions = None393        vpc = Mock()394        self.deploy_operation._get_block_device_mappings = Mock()395        aws_model = self.deploy_operation._create_deployment_parameters(ec2_session=ec2_session,396                                                                        aws_ec2_resource_model=self.ec2_datamodel,397                                                                        ami_deployment_model=ami_model,398                                                                        network_actions=network_actions,399                                                                        vpc=vpc,400                                                                        security_group=None,401                                                                        key_pair='keypair',402                                                                        reservation=Mock(),403                                                                        network_config_results=MagicMock(),404                                                                        logger=self.logger,)405        # if instance doesnt have iam role, the deployment params will have an empty iam instance profile dict406        self.assertTrue(not any(aws_model.iam_role))  # not any(some_dict) => is empty dictionary407    def test_create_deployment_parameters_iam_role_not_arn(self):408        image = Mock()409        image.state = 'available'410        ec2_session = Mock()411        ec2_session.Image = Mock(return_value=image)412        ami_model = Mock()413        ami_model.iam_role = "admin_role"414        ami_model.custom_tags = ""415        vpc = Mock()416        self.deploy_operation._get_block_device_mappings = Mock()417        network_actions = None418        aws_model = self.deploy_operation._create_deployment_parameters(ec2_session=ec2_session,419                                                                        aws_ec2_resource_model=self.ec2_datamodel,420                                                                        ami_deployment_model=ami_model,421                                                                        network_actions=network_actions,422                                                                        vpc=vpc,423                                                                        security_group=None,424                                                                        key_pair='keypair',425                                                                        reservation=Mock(),426                                                                        network_config_results=MagicMock(),427                                                                        logger=self.logger,)428        # if instance has iam role, but not in the form of arn, will return dict with iam role name429        self.assertTrue(aws_model.iam_role['Name'] == ami_model.iam_role)430    def test_create_deployment_parameters_iam_role_arn(self):431        image = Mock()432        image.state = 'available'433        ec2_session = Mock()434        ec2_session.Image = Mock(return_value=image)435        ami_model = Mock()436        network_actions = None437        ami_model.iam_role = "arn:aws:iam::admin_role"438        ami_model.custom_tags = ""439        vpc = Mock()440        self.deploy_operation._get_block_device_mappings = Mock()441        aws_model = self.deploy_operation._create_deployment_parameters(ec2_session=ec2_session,442                                                                        aws_ec2_resource_model=self.ec2_datamodel,443                                                                        ami_deployment_model=ami_model,444                                                                        network_actions=network_actions,445                                                                        vpc=vpc,446                                                                        security_group=None,447                                                                        key_pair='keypair',448                                                                        reservation=Mock(),449                                                                        network_config_results=MagicMock(),450                                                                        logger=self.logger)451        # if instance has iam role, but not in the form of arn, will return dict with iam role name452        self.assertTrue(aws_model.iam_role['Arn'] == ami_model.iam_role)453    def test_prepare_network_interfaces_multi_subnets_with_public_ip(self):454        ami_model = Mock()455        ami_model.add_public_ip = True...

deploy_operation.py

Source:deploy_operation.py

...91                                                                      reservation=reservation,92                                                                      vpc=vpc,93                                                                      logger=logger)94            self.cancellation_service.check_if_cancelled(cancellation_context)95            ami_deployment_info = self._create_deployment_parameters(ec2_session=ec2_session,96                                                                     aws_ec2_resource_model=aws_ec2_cp_resource_model,97                                                                     ami_deployment_model=ami_deployment_model,98                                                                     network_actions=network_actions,99                                                                     vpc=vpc,100                                                                     security_group=security_group,101                                                                     key_pair=key_name,102                                                                     reservation=reservation,103                                                                     network_config_results=network_config_results,104                                                                     logger=logger)105            instance = self.instance_service.create_instance(106                ec2_session=ec2_session,107                name=name,108                reservation=reservation,109                ami_deployment_info=ami_deployment_info,110                ec2_client=ec2_client,111                wait_for_status_check=ami_deployment_model.wait_for_status_check,112                cancellation_context=cancellation_context,113                logger=logger)114            logger.info("Instance created, populating results with interface data")115            self.instance_service.wait_for_instance_to_run_in_aws(ec2_client=ec2_client,116                                                 instance=instance,117                                                 wait_for_status_check=ami_deployment_model.wait_for_status_check,118                                                 cancellation_context=cancellation_context,119                                                 logger=logger)120            self._populate_network_config_results_with_interface_data(instance=instance,121                                                                      network_config_results=network_config_results)122            self.cancellation_service.check_if_cancelled(cancellation_context)123            self.elastic_ip_service.set_elastic_ips(ec2_session=ec2_session,124                                                    ec2_client=ec2_client,125                                                    instance=instance,126                                                    ami_deployment_model=ami_deployment_model,127                                                    network_actions=network_actions,128                                                    network_config_results=network_config_results,129                                                    logger=logger)130            self.cancellation_service.check_if_cancelled(cancellation_context)131        except Exception as e:132            self._rollback_deploy(ec2_session=ec2_session,133                                  instance_id=self._extract_instance_id_on_cancellation(e, instance),134                                  custom_security_group=security_group,135                                  network_config_results=network_config_results,136                                  logger=logger)137            raise  # re-raise original exception after rollback138        logger.info("Instance {} created, getting ami credentials".format(instance.id))139        ami_credentials = self._get_ami_credentials(key_pair_location=aws_ec2_cp_resource_model.key_pairs_location,140                                                    wait_for_credentials=ami_deployment_model.wait_for_credentials,141                                                    instance=instance,142                                                    reservation=reservation,143                                                    s3_session=s3_session,144                                                    ami_deploy_action=ami_deploy_action,145                                                    cancellation_context=cancellation_context,146                                                    logger=logger)147        logger.info("Preparing result")148        deployed_app_attributes = self._prepare_deployed_app_attributes(ami_credentials=ami_credentials,149                                                                        ami_deployment_model=ami_deployment_model,150                                                                        network_config_results=network_config_results)151        vm_details_data = self.vm_details_provider.create(instance)152        network_actions_results_dtos = \153            self._prepare_network_config_results_dto(network_config_results=network_config_results,154                                                     network_actions=network_actions)155        deploy_app_result = DeployAppResult(vmName=self._get_name_from_tags(instance),156                                            vmUuid=instance.instance_id,157                                            deployedAppAttributes=convert_dict_to_attributes_list(deployed_app_attributes),158                                            deployedAppAddress=instance.private_ip_address,159                                            vmDetailsData=vm_details_data,160                                            deployedAppAdditionalData={'inbound_ports': ami_deployment_model.inbound_ports,161                                                                       'public_ip': instance.public_ip_address})162        deploy_app_result.actionId = ami_deploy_action.actionId163        network_actions_results_dtos.append(deploy_app_result)164        return network_actions_results_dtos165    def _validate_public_subnet_exist_if_requested_public_or_elastic_ips(self, ami_deployment_model, network_actions,166                                                                         logger):167        """168        :param DeployAWSEc2AMIInstanceResourceModel ami_deployment_model:169        :param cloudshell.cp.core.models.ConnectSubnet network_actions:170        :param logging.Logger logger:171        """172        if ami_deployment_model.add_public_ip or ami_deployment_model.allocate_elastic_ip:173            connect_subnet_actions = filter(lambda x: isinstance(x, ConnectSubnet), network_actions)174            if not any(x.actionParams.isPublic for x in connect_subnet_actions):175                msg = "Cannot deploy app with elastic or public ip when connected only to private subnets"176                logger.error(msg)177                raise ValueError(msg)178    def _prepare_network_config_results_dto(self, network_config_results, network_actions):179        """180        :param list[DeployNetworkingResultModel] network_config_results:181        :param cloudshell.cp.core.models.ConnectSubnet network_actions:182        :return:183         :rtype" list[DeployNetworkingResultDto]184        """185        if not network_actions:186            return []  # for the moment if we didnt received a connectivity action we shouldnot return anything187        return list(map(self._convertDeployNetworkResultModelToDto, network_config_results))188    def _convertDeployNetworkResultModelToDto(self, network_config_result):189        """190        :param DeployNetworkingResultModel network_config_result:191        :rtype: ConnectToSubnetActionResult192        """193        import json194        interface_data_json_str = json.dumps({195            'interface_id': network_config_result.interface_id,196            'IP': network_config_result.private_ip,197            'Public IP': network_config_result.public_ip,198            'Elastic IP': network_config_result.is_elastic_ip,199            'MAC Address': network_config_result.mac_address,200            'Device Index': network_config_result.device_index,201        })202        return ConnectToSubnetActionResult(actionId=network_config_result.action_id,203                                           interface=interface_data_json_str)204    def _prepare_network_result_models(self, network_actions):205        """206        :type cloudshell.cp.core.models.ConnectSubnet network_actions207        :rtype: list[DeployNetworkingResultModel]208        """209        network_config_results = []210        if not network_actions:211            network_config_results.append(DeployNetworkingResultModel(''))  # init a result object with empty action id212        else:213            for net_config in network_actions:214                if isinstance(net_config.actionParams, ConnectToSubnetParams):215                    network_config_results.append(DeployNetworkingResultModel(net_config.actionId))216        return network_config_results217    def _extract_instance_id_on_cancellation(self, exception, instance):218        """219        :param exception:220        :param instance:221        :return:222        """223        instance_id = None224        if exception and hasattr(exception, "data") and exception.data and 'instance_ids' in exception.data:225            instance_id = exception.data['instance_ids'][226                0]  # we assume at this point that we are working on a single app227        elif instance:228            instance_id = instance.id229        return instance_id230    def _get_ami_credentials(self, s3_session, key_pair_location, reservation, wait_for_credentials, instance,231                             ami_deploy_action, cancellation_context, logger):232        """233        Will load win234        :param s3_session:235        :param key_pair_location:236        :param reservation: reservation model237        :type reservation: cloudshell.cp.aws.models.reservation_model.ReservationModel238        :param wait_for_credentials:239        :param instance:240        :param logging.Logger logger:241        :param CancellationContext cancellation_context:242        :return:243        :rtype: cloudshell.cp.aws.models.ami_credentials.AMICredentials244        """245        # has value for windows instances only246        if instance.platform:247            key_value = self.key_pair_service.load_key_pair_by_name(s3_session=s3_session,248                                                                    bucket_name=key_pair_location,249                                                                    reservation_id=reservation.reservation_id)250            result = None251            try:252                result = self.credentials_service.get_windows_credentials(instance=instance,253                                                                          key_value=key_value,254                                                                          wait_for_password=wait_for_credentials,255                                                                          cancellation_context=cancellation_context)256            except TimeoutError:257                logger.info(258                        "Timeout when waiting for windows credentials. Traceback: {0}".format(traceback.format_exc()))259                return None260        else:261            return None if self._get_deployed_app_resource_user_attribute(ami_deploy_action) else \262                self.credentials_service.get_default_linux_credentials()263        return result264    @staticmethod265    def _get_deployed_app_resource_user_attribute(ami_deploy_action):266        """267        check if deployed app resource has a user attribute, while respecting 2nd gen shell namespaces.268        2nd gen shells are a kind of resource whose attributes are namespaced,269        i.e. User is namespace.User in 2nd gen shells270        :param ami_deploy_action: cloudshell.cp.core.models.DeployApp271        :return:272        """273        attribute_names_in_deployed_resource = ami_deploy_action.actionParams.appResource.attributes.keys()274        return next((attr for attr in attribute_names_in_deployed_resource if attr.split('.')[-1]=='User'))275    @staticmethod276    def _get_name_from_tags(result):277        return [tag['Value'] for tag in result.tags if tag['Key'] == 'Name'][0]278    def _create_security_group_for_instance(self, ami_deployment_model, ec2_session, reservation, vpc, logger):279        if not ami_deployment_model.inbound_ports and not ami_deployment_model.outbound_ports:280            return None281        logger.info("Parsing inbound_ports attribute")282        inbound_ports = PortGroupAttributeParser.parse_port_group_attribute(ami_deployment_model.inbound_ports)283        outbound_ports = PortGroupAttributeParser.parse_port_group_attribute(ami_deployment_model.outbound_ports)284        if not inbound_ports and not outbound_ports:285            logger.info("No data found in inbound_ports attribute")286            return None287        security_group_name = SecurityGroupService.CLOUDSHELL_CUSTOM_SECURITY_GROUP.format(str(uuid.uuid4()))288        security_group = self.security_group_service.create_security_group(ec2_session=ec2_session,289                                                                           vpc_id=vpc.id,290                                                                           security_group_name=security_group_name)291        tags = self.tag_service.get_security_group_tags(name=security_group_name,292                                                        isolation=IsolationTagValues.Exclusive,293                                                        reservation=reservation,294                                                        type=TypeTagValues.InboundPorts)295        self.tag_service.set_ec2_resource_tags(security_group, tags)296        self.security_group_service.set_security_group_rules(security_group=security_group,297                                                             inbound_ports=inbound_ports,298                                                             outbound_ports=outbound_ports)299        if outbound_ports:300            self.security_group_service.remove_allow_all_outbound_rule(security_group=security_group)301        logger.info("Created security group {0} from inbound_ports attribute: {1}"302                    .format(security_group.group_id, ami_deployment_model.inbound_ports))303        return security_group304    def _create_deployment_parameters(self,305                                      ec2_session,306                                      aws_ec2_resource_model,307                                      ami_deployment_model,308                                      network_actions,309                                      vpc,310                                      security_group,311                                      key_pair,312                                      reservation,313                                      network_config_results,314                                      logger):315        """316        :param ec2_session:317        :param aws_ec2_resource_model: The resource model of the AMI deployment option318        :type aws_ec2_resource_model: cloudshell.cp.aws.models.aws_ec2_cloud_provider_resource_model.AWSEc2CloudProviderResourceModel...

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.