How to use check_countspace method in lisa

Best Python code snippet using lisa_python

schema.py

Source:schema.py Github

copy

Full Screen

...429 return super().__hash__()430 def check(self, capability: Any) -> search_space.ResultReason:431 result = super().check(capability)432 result.merge(433 search_space.check_countspace(434 self.data_disk_count, capability.data_disk_count435 ),436 "data_disk_count",437 )438 result.merge(439 search_space.check_countspace(440 self.max_data_disk_count, capability.max_data_disk_count441 ),442 "max_data_disk_count",443 )444 result.merge(445 search_space.check_countspace(446 self.data_disk_iops, capability.data_disk_iops447 ),448 "data_disk_iops",449 )450 return result451 def _get_key(self) -> str:452 return (453 f"{super()._get_key()}/{self.disk_type}/"454 f"{self.data_disk_count}/{self.data_disk_caching_type}/"455 f"{self.data_disk_iops}/{self.data_disk_size}"456 )457 def _call_requirement_method(self, method_name: str, capability: Any) -> Any:458 assert isinstance(capability, DiskOptionSettings), f"actual: {type(capability)}"459 parent_value = super()._call_requirement_method(method_name, capability)460 # convert parent type to child type461 value = DiskOptionSettings()462 value.extended_schemas = parent_value.extended_schemas463 search_space_countspace_method = getattr(464 search_space, f"{method_name}_countspace"465 )466 if self.disk_type or capability.disk_type:467 value.disk_type = getattr(468 search_space, f"{method_name}_setspace_by_priority"469 )(self.disk_type, capability.disk_type, disk_type_priority)470 if self.data_disk_count or capability.data_disk_count:471 value.data_disk_count = search_space_countspace_method(472 self.data_disk_count, capability.data_disk_count473 )474 if self.data_disk_iops or capability.data_disk_iops:475 value.data_disk_iops = search_space_countspace_method(476 self.data_disk_iops, capability.data_disk_iops477 )478 if self.data_disk_size or capability.data_disk_size:479 value.data_disk_size = search_space_countspace_method(480 self.data_disk_size, capability.data_disk_size481 )482 if self.data_disk_caching_type or capability.data_disk_caching_type:483 value.data_disk_caching_type = (484 self.data_disk_caching_type or capability.data_disk_caching_type485 )486 if self.max_data_disk_count or capability.max_data_disk_count:487 value.max_data_disk_count = search_space_countspace_method(488 self.max_data_disk_count, capability.max_data_disk_count489 )490 return value491class NetworkDataPath(str, Enum):492 Synthetic = "Synthetic"493 Sriov = "Sriov"494_network_data_path_priority: List[NetworkDataPath] = [495 NetworkDataPath.Sriov,496 NetworkDataPath.Synthetic,497]498@dataclass_json()499@dataclass()500class NetworkInterfaceOptionSettings(FeatureSettings):501 type: str = "NetworkInterface"502 data_path: Optional[503 Union[search_space.SetSpace[NetworkDataPath], NetworkDataPath]504 ] = field( # type: ignore505 default_factory=partial(506 search_space.SetSpace,507 items=[508 NetworkDataPath.Synthetic,509 NetworkDataPath.Sriov,510 ],511 ),512 metadata=field_metadata(513 decoder=partial(514 search_space.decode_set_space_by_type, base_type=NetworkDataPath515 )516 ),517 )518 # nic_count is used for specifying associated nic count during provisioning vm519 nic_count: search_space.CountSpace = field(520 default_factory=partial(search_space.IntRange, min=1),521 metadata=field_metadata(decoder=search_space.decode_count_space),522 )523 # max_nic_count is used for getting the size max nic capability, it can be used to524 # check how many nics the vm can be associated after provisioning525 max_nic_count: search_space.CountSpace = field(526 default_factory=partial(search_space.IntRange, min=1),527 metadata=field_metadata(528 allow_none=True, decoder=search_space.decode_count_space529 ),530 )531 def __eq__(self, o: object) -> bool:532 assert isinstance(o, NetworkInterfaceOptionSettings), f"actual: {type(o)}"533 return (534 self.type == o.type535 and self.data_path == o.data_path536 and self.nic_count == o.nic_count537 and self.max_nic_count == o.max_nic_count538 )539 def __repr__(self) -> str:540 return (541 f"data_path:{self.data_path}, nic_count:{self.nic_count},"542 f" max_nic_count:{self.max_nic_count}"543 )544 def __str__(self) -> str:545 return self.__repr__()546 def __hash__(self) -> int:547 return super().__hash__()548 def _get_key(self) -> str:549 return (550 f"{super()._get_key()}/{self.data_path}/{self.nic_count}"551 f"/{self.max_nic_count}"552 )553 def check(self, capability: Any) -> search_space.ResultReason:554 assert isinstance(555 capability, NetworkInterfaceOptionSettings556 ), f"actual: {type(capability)}"557 result = super().check(capability)558 result.merge(559 search_space.check_countspace(self.nic_count, capability.nic_count),560 "nic_count",561 )562 result.merge(563 search_space.check_setspace(self.data_path, capability.data_path),564 "data_path",565 )566 result.merge(567 search_space.check_countspace(self.max_nic_count, capability.max_nic_count),568 "max_nic_count",569 )570 return result571 def _call_requirement_method(self, method_name: str, capability: Any) -> Any:572 assert isinstance(573 capability, NetworkInterfaceOptionSettings574 ), f"actual: {type(capability)}"575 parent_value = super()._call_requirement_method(method_name, capability)576 # convert parent type to child type577 value = NetworkInterfaceOptionSettings()578 value.extended_schemas = parent_value.extended_schemas579 value.max_nic_count = getattr(search_space, f"{method_name}_countspace")(580 self.max_nic_count, capability.max_nic_count581 )582 if self.nic_count or capability.nic_count:583 value.nic_count = getattr(search_space, f"{method_name}_countspace")(584 self.nic_count, capability.nic_count585 )586 else:587 raise LisaException("nic_count cannot be zero")588 value.data_path = getattr(search_space, f"{method_name}_setspace_by_priority")(589 self.data_path, capability.data_path, _network_data_path_priority590 )591 return value592@dataclass_json()593@dataclass()594class FeaturesSpace(595 search_space.SetSpace[Union[str, FeatureSettings]],596):597 def __post_init__(self, *args: Any, **kwargs: Any) -> None:598 if self.items:599 for index, item in enumerate(self.items):600 if isinstance(item, dict):601 item = load_by_type(FeatureSettings, item)602 self.items[index] = item603@dataclass_json()604@dataclass()605class NodeSpace(search_space.RequirementMixin, TypedSchema, ExtendableSchemaMixin):606 type: str = field(607 default=constants.ENVIRONMENTS_NODES_REQUIREMENT,608 metadata=field_metadata(609 required=True,610 validate=validate.OneOf([constants.ENVIRONMENTS_NODES_REQUIREMENT]),611 ),612 )613 name: str = ""614 is_default: bool = field(default=False)615 node_count: search_space.CountSpace = field(616 default=search_space.IntRange(min=1),617 metadata=field_metadata(decoder=search_space.decode_count_space),618 )619 core_count: search_space.CountSpace = field(620 default=search_space.IntRange(min=1),621 metadata=field_metadata(decoder=search_space.decode_count_space),622 )623 memory_mb: search_space.CountSpace = field(624 default=search_space.IntRange(min=512),625 metadata=field_metadata(decoder=search_space.decode_count_space),626 )627 disk: Optional[DiskOptionSettings] = None628 network_interface: Optional[NetworkInterfaceOptionSettings] = None629 gpu_count: search_space.CountSpace = field(630 default=search_space.IntRange(min=0),631 metadata=field_metadata(decoder=search_space.decode_count_space),632 )633 # all features on requirement should be included.634 # all features on capability can be included.635 _features: Optional[FeaturesSpace] = field(636 default=None,637 metadata=field_metadata(allow_none=True, data_key="features"),638 )639 # set by requirements640 # capability's is ignored641 _excluded_features: Optional[FeaturesSpace] = field(642 default=None,643 metadata=field_metadata(644 allow_none=True,645 data_key="excluded_features",646 ),647 )648 def __post_init__(self, *args: Any, **kwargs: Any) -> None:649 # clarify types to avoid type errors in properties.650 self._features: Optional[search_space.SetSpace[FeatureSettings]]651 self._excluded_features: Optional[search_space.SetSpace[FeatureSettings]]652 def __eq__(self, o: object) -> bool:653 assert isinstance(o, NodeSpace), f"actual: {type(o)}"654 return (655 self.type == o.type656 and self.node_count == o.node_count657 and self.core_count == o.core_count658 and self.memory_mb == o.memory_mb659 and self.disk == o.disk660 and self.network_interface == o.network_interface661 and self.gpu_count == o.gpu_count662 and self.features == o.features663 and self.excluded_features == o.excluded_features664 )665 def __repr__(self) -> str:666 """667 override it for shorter text668 """669 return (670 f"type:{self.type},name:{self.name},"671 f"default:{self.is_default},"672 f"count:{self.node_count},core:{self.core_count},"673 f"mem:{self.memory_mb},disk:{self.disk},"674 f"network interface: {self.network_interface}, gpu:{self.gpu_count},"675 f"f:{self.features},ef:{self.excluded_features},"676 f"{super().__repr__()}"677 )678 @property679 def cost(self) -> float:680 core_count = search_space.generate_min_capability_countspace(681 self.core_count, self.core_count682 )683 gpu_count = search_space.generate_min_capability_countspace(684 self.gpu_count, self.gpu_count685 )686 return core_count + gpu_count * 100687 @property688 def features(self) -> Optional[search_space.SetSpace[FeatureSettings]]:689 self._features = self._create_feature_settings_list(self._features)690 if self._features is not None:691 self._features.is_allow_set = True692 return cast(Optional[search_space.SetSpace[FeatureSettings]], self._features)693 @features.setter694 def features(self, value: Optional[search_space.SetSpace[FeatureSettings]]) -> None:695 self._features = cast(FeaturesSpace, value)696 @property697 def excluded_features(self) -> Optional[search_space.SetSpace[FeatureSettings]]:698 if not self._excluded_features:699 self._excluded_features = self._create_feature_settings_list(700 self._excluded_features701 )702 if self._excluded_features is not None:703 self._excluded_features.is_allow_set = False704 return cast(705 Optional[search_space.SetSpace[FeatureSettings]], self._excluded_features706 )707 @excluded_features.setter708 def excluded_features(709 self, value: Optional[search_space.SetSpace[FeatureSettings]]710 ) -> None:711 self._excluded_features = cast(FeaturesSpace, value)712 def check(self, capability: Any) -> search_space.ResultReason:713 result = search_space.ResultReason()714 if capability is None:715 result.add_reason("capability shouldn't be None")716 if self.features:717 assert self.features.is_allow_set, "features should be allow set"718 if self.excluded_features:719 assert (720 not self.excluded_features.is_allow_set721 ), "excluded_features shouldn't be allow set"722 assert isinstance(capability, NodeSpace), f"actual: {type(capability)}"723 if (724 not capability.node_count725 or not capability.core_count726 or not capability.memory_mb727 ):728 result.add_reason(729 "node_count, core_count, memory_mb " "shouldn't be None or zero."730 )731 if isinstance(self.node_count, int) and isinstance(capability.node_count, int):732 if self.node_count > capability.node_count:733 result.add_reason(734 f"capability node count {capability.node_count} "735 f"must be more than requirement {self.node_count}"736 )737 else:738 result.merge(739 search_space.check_countspace(self.node_count, capability.node_count),740 "node_count",741 )742 result.merge(743 search_space.check_countspace(self.core_count, capability.core_count),744 "core_count",745 )746 result.merge(747 search_space.check_countspace(self.memory_mb, capability.memory_mb),748 "memory_mb",749 )750 if self.disk:751 result.merge(self.disk.check(capability.disk))752 if self.network_interface:753 result.merge(self.network_interface.check(capability.network_interface))754 result.merge(755 search_space.check_countspace(self.gpu_count, capability.gpu_count),756 "gpu_count",757 )758 if self.features:759 for feature in self.features:760 cap_feature = self._find_feature_by_type(761 feature.type, capability.features762 )763 if cap_feature:764 result.merge(feature.check(cap_feature))765 else:766 result.add_reason(767 f"no feature '{feature.type}' found in capability"768 )769 if self.excluded_features:...

Full Screen

Full Screen

features.py

Source:features.py Github

copy

Full Screen

...307 search_space.check_setspace(self.disk_type, capability.disk_type),308 "disk_type",309 )310 result.merge(311 search_space.check_countspace(312 self.data_disk_count, capability.data_disk_count313 ),314 "data_disk_count",315 )316 result.merge(317 search_space.check_countspace(318 self.data_disk_iops, capability.data_disk_iops319 ),320 "data_disk_iops",321 )322 result.merge(323 search_space.check_countspace(324 self.data_disk_size, capability.data_disk_size325 ),326 "data_disk_size",327 )328 result.merge(329 search_space.check_countspace(330 self.max_data_disk_count, capability.max_data_disk_count331 ),332 "max_data_disk_count",333 )334 return result335 def _call_requirement_method(self, method_name: str, capability: Any) -> Any:336 assert isinstance(337 capability, AwsDiskOptionSettings338 ), f"actual: {type(capability)}"339 assert (340 capability.disk_type341 ), "capability should have at least one disk type, but it's None"342 value = AwsDiskOptionSettings()343 super_value = schema.DiskOptionSettings._call_requirement_method(...

Full Screen

Full Screen

test_search_space.py

Source:test_search_space.py Github

copy

Full Screen

...25class MockItem(RequirementMixin):26 number: CountSpace = IntRange(min=1, max=5)27 def check(self, capability: Any) -> ResultReason:28 assert isinstance(capability, MockItem), f"actual: {type(capability)}"29 return check_countspace(self.number, capability.number)30 def _generate_min_capability(self, capability: Any) -> MockSchema:31 result = MockSchema()32 assert isinstance(capability, MockItem), f"actual: {type(capability)}"33 result.number = generate_min_capability_countspace(34 self.number, capability.number35 )36 return result37class SearchSpaceTestCase(unittest.TestCase):38 def __init__(self, *args: Any, **kwargs: Any) -> None:39 super().__init__(*args, **kwargs)40 id = f"{'.'.join(self.id().split('.')[-2:])}"41 self._log = get_logger(id)42 def test_supported_intrange(self) -> None:43 self._verify_matrix(44 expected_meet=[45 [True, True, True, False, True, True, False, True, False, False],46 [True, True, False, False, True, True, False, False, False, False],47 ],48 expected_min=[49 [12, 10, 15, False, 10, 10, False, 15, False, False],50 [12, 10, False, False, 10, 10, False, False, False, False],51 ],52 requirements=[53 IntRange(min=10, max=15),54 IntRange(min=10, max=15, max_inclusive=False),55 ],56 capabilities=[57 IntRange(12),58 IntRange(10),59 IntRange(15),60 IntRange(20),61 IntRange(5, 11),62 IntRange(5, 10),63 IntRange(5, 10, max_inclusive=False),64 IntRange(15, 20),65 IntRange(1, 5),66 IntRange(20, 100),67 ],68 )69 def test_supported_countspace(self) -> None:70 expected_meet = [71 [True, True, True, True, True, True, True, True, True, True, True],72 [False, True, False, False, False, True, True, True, False, False, False],73 [False, False, True, False, False, True, False, True, True, False, False],74 [False, False, False, False, True, False, False, True, True, True, True],75 [False, True, True, False, False, True, True, True, True, False, False],76 [False, True, False, False, False, True, True, True, True, False, False],77 [False, True, True, False, True, True, True, True, True, True, True],78 ]79 expected_min: List[List[Any]] = [80 [None, 10, 15, 18, 25, 10, 10, 10, 12, 21, 21],81 [False, 10, False, False, False, 10, 10, 10, False, False, False],82 [False, False, 15, False, False, 15, False, 15, 15, False, False],83 [False, False, False, False, 25, False, False, 25, 25, 25, 25],84 [False, 10, 15, False, False, 10, 10, 10, 12, False, False],85 [False, 10, False, False, False, 10, 10, 10, 12, False, False],86 [False, 10, 15, False, 25, 10, 10, 10, 12, 21, 21],87 ]88 self._verify_matrix(89 expected_meet=expected_meet,90 expected_min=expected_min,91 requirements=[92 None,93 10,94 15,95 25,96 IntRange(min=10, max=15),97 IntRange(min=10, max=15, max_inclusive=False),98 [IntRange(min=10, max=15), IntRange(min=20, max=80)],99 ],100 capabilities=[101 None,102 10,103 15,104 18,105 25,106 IntRange(min=10, max=15),107 IntRange(min=10, max=15, max_inclusive=False),108 [IntRange(min=10, max=15), IntRange(min=20, max=80)],109 [IntRange(min=12, max=30)],110 [IntRange(min=21, max=25)],111 IntRange(min=21, max=25),112 ],113 )114 def test_supported_set_space(self) -> None:115 set_aa = set(["aa"])116 set_aa_bb = set(["aa", "bb"])117 set_aa_bb_cc = set(["aa", "bb", "cc"])118 set_aa_cc = set(["aa", "cc"])119 set_cc = set(["cc"])120 self._verify_matrix(121 expected_meet=[122 [True, True, True, True, True],123 [True, True, True, True, True],124 [False, False, False, True, False],125 [True, False, True, False, False],126 [True, False, True, False, False],127 ],128 expected_min=[129 [None, None, None, None, None],130 [None, None, None, None, None],131 [False, False, False, set_aa_bb, False],132 [None, False, None, False, False],133 [None, False, None, False, False],134 ],135 requirements=[136 SetSpace[str](is_allow_set=True),137 SetSpace[str](is_allow_set=False),138 SetSpace[str](items=set_aa_bb, is_allow_set=True),139 SetSpace[str](items=set_aa_bb),140 SetSpace[str](items=set_aa_bb, is_allow_set=False),141 ],142 capabilities=[143 SetSpace[str](is_allow_set=True),144 SetSpace[str](items=set_aa, is_allow_set=True),145 SetSpace[str](items=set_cc, is_allow_set=True),146 SetSpace[str](items=set_aa_bb_cc, is_allow_set=True),147 SetSpace[str](items=set_aa_cc, is_allow_set=True),148 ],149 )150 def test_generate_min_capability_not_supported(self) -> None:151 requirement = IntRange(min=5)152 capability = IntRange(max=4)153 with self.assertRaises(expected_exception=LisaException) as cm:154 requirement.generate_min_capability(capability)155 self.assertIn("doesn't support", str(cm.exception))156 def test_int_range_validation(self) -> None:157 with self.assertRaises(expected_exception=LisaException) as cm:158 IntRange(min=6, max=4)159 self.assertIn("shouldn't be greater than", str(cm.exception))160 # no exception161 IntRange(min=5, max=5)162 with self.assertRaises(expected_exception=LisaException) as cm:163 IntRange(min=5, max=5, max_inclusive=False)164 self.assertIn("shouldn't be equal to", str(cm.exception))165 def _verify_matrix(166 self,167 expected_meet: List[List[bool]],168 expected_min: List[List[Any]],169 requirements: List[T],170 capabilities: List[T],171 ) -> None:172 for r_index, requirement in enumerate(requirements):173 for c_index, capability in enumerate(capabilities):174 extra_msg = (175 f"index: [{r_index},{c_index}], "176 f"requirement: {requirement}, capability: {capability}"177 )178 if isinstance(requirement, RequirementMixin):179 self._assert_check(180 expected_meet[r_index][c_index],181 requirement.check(capability),182 extra_msg=extra_msg,183 )184 if expected_meet[r_index][c_index]:185 actual_min = requirement.generate_min_capability(capability)186 if expected_min[r_index][c_index] != actual_min:187 self._log.info(extra_msg)188 self._log.info(189 f"expected_min: {expected_min[r_index][c_index]}"190 )191 self._log.info(f"actual_min: {actual_min}")192 self.assertEqual(193 expected_min[r_index][c_index], actual_min, extra_msg194 )195 elif (196 isinstance(requirement, IntRange)197 or isinstance(requirement, int)198 or isinstance(capability, IntRange)199 or isinstance(capability, int)200 ):201 self._assert_check(202 expected_meet[r_index][c_index],203 check_countspace(requirement, capability), # type:ignore204 extra_msg=extra_msg,205 )206 if expected_meet[r_index][c_index]:207 actual_min = generate_min_capability_countspace(208 requirement, capability # type:ignore209 )210 if expected_min[r_index][c_index] != actual_min:211 self._log.info(extra_msg)212 self.assertEqual(213 expected_min[r_index][c_index], actual_min, extra_msg214 )215 else:216 self._assert_check(217 expected_meet[r_index][c_index],...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful