KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _check_or_create_security_group method in lisa

Best Python code snippet using lisa_python
platform_.py
Source:platform_.py
...191            self._log.error("Couldn't create key %s.", key_name)192            raise193        else:194            return key_pair195    def _check_or_create_security_group(  # noqa: C901196        self, security_group_name: str, group_description: str197    ) -> Any:198        try:199            ec2_resource = boto3.resource("ec2")200            # By default, AWS users can create up to 5 VPCs201            for i in range(50, 55):202                cidr_block = "173." + str(i) + ".0.0/16"203                vpcs = list(204                    ec2_resource.vpcs.filter(205                        Filters=[{"Name": "cidr", "Values": [cidr_block]}]206                    )207                )208                if len(vpcs) == 0:209                    self._vpc = ec2_resource.create_vpc(CidrBlock=cidr_block)210                    self._log.info(211                        f"Create a new VPC: {self._vpc.id}"212                        f"with CIDR block {self._vpc.cidr_block}"213                    )214                    self._internet_gateway = ec2_resource.create_internet_gateway()215                    self._vpc.attach_internet_gateway(216                        InternetGatewayId=self._internet_gateway.id217                    )218                    self._route_table = ec2_resource.create_route_table(219                        VpcId=self._vpc.id220                    )221                    self._route_table.create_route(222                        DestinationCidrBlock="0.0.0.0/0",223                        GatewayId=self._internet_gateway.id,224                    )225                    self._log.info(226                        "Create an internet gateway: %s and a route table %s",227                        self._internet_gateway.id,228                        self._route_table.id,229                    )230                    break231            if self._vpc is None:232                raise LisaException(233                    "Couldn't get/create VPCs as there are 5 exiting VPCs."234                    "Please wait for others finishing test."235                )236        except ClientError:237            self._log.exception("Couldn't get/create VPCs.")238            raise239        try:240            security_group = self._vpc.create_security_group(241                GroupName=security_group_name, Description=group_description242            )243            self._log.info(244                "Created security group %s in VPC %s.",245                security_group_name,246                self._vpc.id,247            )248        except ClientError:249            self._log.exception(250                "Couldn't create security group %s.", security_group_name251            )252            raise253        try:254            ip_permissions: List[IpPermissionTypeDef] = [255                {256                    # SSH ingress open to anyone257                    "IpProtocol": "tcp",258                    "FromPort": 22,259                    "ToPort": 22,260                    "IpRanges": [{"CidrIp": "0.0.0.0/0"}],261                },262                {263                    # Open to ips in the vpc264                    "IpProtocol": "-1",265                    "FromPort": -1,266                    "ToPort": -1,267                    "IpRanges": [{"CidrIp": self._vpc.cidr_block}],268                },269            ]270            security_group.authorize_ingress(IpPermissions=ip_permissions)271            self._log.info("Set inbound rules for %s to allow SSH.", security_group.id)272        except ClientError:273            self._log.exception(274                "couldn't authorize inbound rules for %s.", security_group_name275            )276            raise277        else:278            return security_group279    def _prepare_environment(  # noqa: C901280        self, environment: Environment, log: Logger281    ) -> bool:282        # TODO: Reduce this function's complexity and remove the disabled warning.283        """284        Main flow285        1. load location, vm size patterns firstly.286        2. load available vm sizes for each location.287        3. match vm sizes by pattern.288        for each environment289        1. If predefined location exists on node level, check conflict and use it.290        2. If predefined vm size exists on node level, check exists and use it.291        3. check capability for each node by order of pattern.292        4. get min capability for each match293        """294        is_success: bool = True295        ec2_resource = boto3.resource("ec2")296        if environment.runbook.nodes_requirement:297            is_success = False298            nodes_requirement = environment.runbook.nodes_requirement299            node_count = len(nodes_requirement)300            # fills predefined locations here.301            predefined_caps: List[Any] = [None] * node_count302            # make sure all vms are in same location.303            existing_location: str = ""304            predefined_cost: float = 0305            for req in nodes_requirement:306                # covert to aws node space, so the aws extensions can be loaded.307                _convert_to_aws_node_space(req)308                # check locations309                # apply aws specified values310                node_runbook: AwsNodeSchema = req.get_extended_runbook(311                    AwsNodeSchema, AWS312                )313                if node_runbook.location:314                    if existing_location:315                        # if any one has different location, calculate again316                        if existing_location != node_runbook.location:317                            raise LisaException(318                                f"predefined node must be in same location, "319                                f"previous: {existing_location}, "320                                f"found: {node_runbook.location}"321                            )322                    else:323                        existing_location = node_runbook.location324            if existing_location:325                locations = [existing_location]326            else:327                locations = LOCATIONS328            # check eligible locations329            found_or_skipped = False330            for location_name in locations:331                predefined_cost = 0332                predefined_caps = [None] * node_count333                for req_index, req in enumerate(nodes_requirement):334                    found_or_skipped = False335                    node_runbook = req.get_extended_runbook(AwsNodeSchema, AWS)336                    if not node_runbook.vm_size:337                        # not to check, if no vm_size set338                        found_or_skipped = True339                        continue340                    # find predefined vm size on all available's.341                    location_info: AwsLocation = self._get_location_info(342                        location_name, log343                    )344                    matched_score: float = 0345                    matched_cap: Optional[AwsCapability] = None346                    matcher = SequenceMatcher(None, node_runbook.vm_size.lower(), "")347                    for aws_cap in location_info.capabilities:348                        matcher.set_seq2(aws_cap.vm_size.lower())349                        if (350                            node_runbook.vm_size.lower() in aws_cap.vm_size.lower()351                            and matched_score < matcher.ratio()352                        ):353                            matched_cap = aws_cap354                            matched_score = matcher.ratio()355                    if matched_cap:356                        predefined_cost += matched_cap.estimated_cost357                        min_cap = self._generate_min_capability(358                            req, matched_cap, location_name359                        )360                        if not existing_location:361                            existing_location = location_name362                        predefined_caps[req_index] = min_cap363                        found_or_skipped = True364                    else:365                        # if not found any, skip and try next location366                        break367                if found_or_skipped:368                    # if found all, skip other locations369                    break370            if found_or_skipped:371                for location_name in locations:372                    # in each location, all node must be found373                    # fill them as None and check after met capability374                    found_capabilities: List[Any] = list(predefined_caps)375                    # skip unmatched location376                    if existing_location and existing_location != location_name:377                        continue378                    estimated_cost: float = 0379                    location_caps = self.get_eligible_vm_sizes(location_name, log)380                    for req_index, req in enumerate(nodes_requirement):381                        node_runbook = req.get_extended_runbook(AwsNodeSchema, AWS)382                        image = ec2_resource.Image(node_runbook.get_image_id())383                        for aws_cap in location_caps:384                            if found_capabilities[req_index]:385                                # found, so skipped386                                break387                            # Check if the instance type is on the same architecture388                            # as the image.389                            processor_info = aws_cap.resource_sku["ProcessorInfo"]390                            supported_archs = processor_info["SupportedArchitectures"]391                            if image.architecture != supported_archs[0]:392                                continue393                            check_result = req.check(aws_cap.capability)394                            if check_result.result:395                                min_cap = self._generate_min_capability(396                                    req, aws_cap, aws_cap.location397                                )398                                estimated_cost += aws_cap.estimated_cost399                                found_capabilities[req_index] = min_cap400                        if all(x for x in found_capabilities):401                            break402                    if all(x for x in found_capabilities):403                        # all found and replace current requirement404                        environment.runbook.nodes_requirement = found_capabilities405                        environment.cost = estimated_cost + predefined_cost406                        is_success = True407                        log.debug(408                            f"requirement meet, "409                            f"cost: {environment.cost}, "410                            f"cap: {environment.runbook.nodes_requirement}"411                        )412                        break413        return is_success414    def _deploy_environment(self, environment: Environment, log: Logger) -> None:415        assert self._ec2_client416        assert self._aws_runbook417        environment_context = get_environment_context(environment=environment)418        normalized_run_name = constants.NORMALIZE_PATTERN.sub("_", constants.RUN_NAME)419        if self._aws_runbook.security_group_name:420            security_group_name = self._aws_runbook.security_group_name421        else:422            security_group_name = f"{normalized_run_name}__sec_group"423        if self._aws_runbook.key_pair_name:424            key_pair_name = self._aws_runbook.key_pair_name425        else:426            key_pair_name = f"{normalized_run_name}_keypair"427        environment_context.security_group_name = security_group_name428        environment_context.key_pair_name = key_pair_name429        if self._aws_runbook.dry_run:430            log.info(f"dry_run: {self._aws_runbook.dry_run}")431        else:432            try:433                if self._aws_runbook.deploy:434                    log.info(435                        f"creating or updating security group: [{security_group_name}]"436                    )437                    self._security_group = self._check_or_create_security_group(438                        security_group_name=security_group_name,439                        group_description="Lisa security group for testing.",440                    )441                    environment_context.security_group_is_created = True442                    environment_context.security_group_id = self._security_group.id443                    if self.runbook.admin_private_key_file:444                        self._key_pair = self._create_key_pair(445                            key_pair_name, self.runbook.admin_private_key_file446                        )447                else:448                    log.info(449                        f"reusing security group: [{security_group_name}]"450                        f" and key pair: [{key_pair_name}]"451                    )...
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.