KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use expand_by_node_count method in lisa

Best Python code snippet using lisa_python

schema.py

Source:schema.py

...775                    result.add_reason(776                        f"excluded feature '{feature.type}' found in capability"777                    )778        return result779    def expand_by_node_count(self) -> List[Any]:780        # expand node count in requirement to one,781        # so that's easy to compare equalization later.782        expanded_requirements: List[NodeSpace] = []783        node_count = search_space.generate_min_capability_countspace(784            self.node_count, self.node_count785        )786        for _ in range(node_count):787            expanded_copy = copy.copy(self)788            expanded_copy.node_count = 1789            expanded_requirements.append(expanded_copy)790        return expanded_requirements791    def has_feature(self, find_type: str) -> bool:792        result = False793        if not self.features:794            return result795        return any(feature for feature in self.features if feature.type == find_type)796    def _call_requirement_method(self, method_name: str, capability: Any) -> Any:797        assert isinstance(capability, NodeSpace), f"actual: {type(capability)}"798        # copy to duplicate extended schema799        value: NodeSpace = copy.deepcopy(self)800        if self.node_count or capability.node_count:801            if isinstance(self.node_count, int) and isinstance(802                capability.node_count, int803            ):804                # capability can have more node805                value.node_count = capability.node_count806            else:807                value.node_count = getattr(search_space, f"{method_name}_countspace")(808                    self.node_count, capability.node_count809                )810        else:811            raise LisaException("node_count cannot be zero")812        if self.core_count or capability.core_count:813            value.core_count = getattr(search_space, f"{method_name}_countspace")(814                self.core_count, capability.core_count815            )816        else:817            raise LisaException("core_count cannot be zero")818        if self.memory_mb or capability.memory_mb:819            value.memory_mb = getattr(search_space, f"{method_name}_countspace")(820                self.memory_mb, capability.memory_mb821            )822        else:823            raise LisaException("memory_mb cannot be zero")824        if self.disk or capability.disk:825            value.disk = getattr(search_space, method_name)(self.disk, capability.disk)826        if self.network_interface or capability.network_interface:827            value.network_interface = getattr(search_space, method_name)(828                self.network_interface, capability.network_interface829            )830        if self.gpu_count or capability.gpu_count:831            value.gpu_count = getattr(search_space, f"{method_name}_countspace")(832                self.gpu_count, capability.gpu_count833            )834        else:835            value.gpu_count = 0836        if (837            capability.features838            and method_name == search_space.RequirementMethod.generate_min_capability839        ):840            # The requirement features are ignored, if cap doesn't have it.841            value.features = search_space.SetSpace[FeatureSettings](is_allow_set=True)842            for original_cap_feature in capability.features:843                capability_feature = self._get_or_create_feature_settings(844                    original_cap_feature845                )846                requirement_feature = (847                    self._find_feature_by_type(capability_feature.type, self.features)848                    or capability_feature849                )850                current_feature = getattr(requirement_feature, method_name)(851                    capability_feature852                )853                value.features.add(current_feature)854        elif method_name == search_space.RequirementMethod.intersect and (855            capability.features or self.features856        ):857            # This is a hack to work with lisa_runner. The capability features858            # are joined case req and runbook req. Here just take the results859            # from capability.860            value.features = capability.features861        if (862            capability.excluded_features863            and method_name == search_space.RequirementMethod.generate_min_capability864        ):865            # TODO: the min value for excluded feature is not clear. It may need866            # to be improved with real scenarios.867            value.excluded_features = search_space.SetSpace[FeatureSettings](868                is_allow_set=True869            )870            for original_cap_feature in capability.excluded_features:871                capability_feature = self._get_or_create_feature_settings(872                    original_cap_feature873                )874                requirement_feature = (875                    self._find_feature_by_type(876                        capability_feature.type, self.excluded_features877                    )878                    or capability_feature879                )880                current_feature = getattr(requirement_feature, method_name)(881                    capability_feature882                )883                value.excluded_features.add(current_feature)884        elif method_name == search_space.RequirementMethod.intersect and (885            capability.excluded_features or self.excluded_features886        ):887            # This is a hack to work with lisa_runner. The capability features888            # are joined case req and runbook req. Here just take the results889            # from capability.890            value.excluded_features = capability.excluded_features891        return value892    def _find_feature_by_type(893        self,894        find_type: str,895        features: Optional[search_space.SetSpace[Any]],896    ) -> Optional[FeatureSettings]:897        result: Optional[FeatureSettings] = None898        if not features:899            return result900        is_found = False901        for original_feature in features.items:902            feature = self._get_or_create_feature_settings(original_feature)903            if feature.type == find_type:904                is_found = True905                break906        if is_found:907            result = feature908        return result909    def _create_feature_settings_list(910        self, features: Optional[search_space.SetSpace[Any]]911    ) -> Optional[FeaturesSpace]:912        result: Optional[FeaturesSpace] = None913        if features is None:914            return result915        result = cast(916            FeaturesSpace,917            search_space.SetSpace[FeatureSettings](is_allow_set=features.is_allow_set),918        )919        for raw_feature in features.items:920            feature = self._get_or_create_feature_settings(raw_feature)921            result.add(feature)922        return result923    def _get_or_create_feature_settings(self, feature: Any) -> FeatureSettings:924        if isinstance(feature, str):925            feature_setting = FeatureSettings.create(feature)926        elif isinstance(feature, FeatureSettings):927            feature_setting = feature928        else:929            raise LisaException(930                f"unsupported type {type(feature)} found in features, "931                "only str and FeatureSettings supported."932            )933        return feature_setting934@dataclass_json()935@dataclass936class Capability(NodeSpace):937    type: str = constants.ENVIRONMENTS_NODES_REQUIREMENT938    def __post_init__(self, *args: Any, **kwargs: Any) -> None:939        super().__post_init__(*args, **kwargs)940        self.node_count = 1941@dataclass_json()942@dataclass943class Node(TypedSchema, ExtendableSchemaMixin):944    capability: Capability = field(default_factory=Capability)945    name: str = ""946    is_default: bool = field(default=False)947@dataclass_json()948@dataclass949class LocalNode(Node):950    type: str = constants.ENVIRONMENTS_NODES_LOCAL951@dataclass_json()952@dataclass953class RemoteNode(Node):954    type: str = constants.ENVIRONMENTS_NODES_REMOTE955    address: str = ""956    port: int = field(957        default=22,958        metadata=field_metadata(959            field_function=fields.Int, validate=validate.Range(min=1, max=65535)960        ),961    )962    public_address: str = ""963    public_port: int = field(964        default=22,965        metadata=field_metadata(966            field_function=fields.Int, validate=validate.Range(min=1, max=65535)967        ),968    )969    username: str = constants.DEFAULT_USER_NAME970    password: str = ""971    private_key_file: str = ""972    def __post_init__(self, *args: Any, **kwargs: Any) -> None:973        add_secret(self.username, PATTERN_HEADTAIL)974        add_secret(self.password)975        add_secret(self.private_key_file)976@dataclass_json()977@dataclass978class Environment:979    name: str = field(default="")980    topology: str = field(981        default=constants.ENVIRONMENTS_SUBNET,982        metadata=field_metadata(983            validate=validate.OneOf([constants.ENVIRONMENTS_SUBNET])984        ),985    )986    nodes_raw: Optional[List[Any]] = field(987        default=None,988        metadata=field_metadata(data_key=constants.NODES),989    )990    nodes_requirement: Optional[List[NodeSpace]] = None991    _original_nodes_requirement: Optional[List[NodeSpace]] = None992    def __post_init__(self, *args: Any, **kwargs: Any) -> None:993        self._original_nodes_requirement = self.nodes_requirement994        self.reload_requirements()995    def reload_requirements(self) -> None:996        results: List[Node] = []997        self.nodes = []998        self.nodes_requirement = None999        if self._original_nodes_requirement:1000            self.nodes_requirement = []1001            self.nodes_requirement.extend(copy.copy(self._original_nodes_requirement))1002        if self.nodes_raw:1003            for node_raw in self.nodes_raw:1004                node_type = node_raw[constants.TYPE]1005                if node_type == constants.ENVIRONMENTS_NODES_REQUIREMENT:1006                    original_req: NodeSpace = load_by_type(NodeSpace, node_raw)1007                    expanded_req = original_req.expand_by_node_count()1008                    if self.nodes_requirement is None:1009                        self.nodes_requirement = []1010                    self.nodes_requirement.extend(expanded_req)1011                else:1012                    # load base schema for future parsing1013                    node: Node = load_by_type(Node, node_raw)1014                    results.append(node)1015        self.nodes = results1016@dataclass_json()1017@dataclass1018class EnvironmentRoot:1019    warn_as_error: bool = field(default=False)1020    environments: List[Environment] = field(default_factory=list)1021@dataclass_json()...

environment.py

Source:environment.py

...126    def _expand_node_space(self) -> None:127        if self.nodes:128            expanded_requirements: List[schema.NodeSpace] = []129            for node in self.nodes:130                expanded_requirements.extend(node.expand_by_node_count())131            self.nodes = expanded_requirements132class Environment(ContextMixin, InitializableMixin):133    def __init__(134        self,135        is_predefined: bool,136        warn_as_error: bool,137        id_: int,138        runbook: schema.Environment,139    ) -> None:140        super().__init__()141        self.nodes: Nodes142        self.runbook = runbook143        self.name = runbook.name144        self.is_predefined: bool = is_predefined...

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.