def _generate_min_capability()

in lisa/sut_orchestrator/azure/features.py [0:0]

• 95 lines of code
• 22 McCabe index (conditional complexity)

    def _generate_min_capability(self, capability: Any) -> Any:
        assert isinstance(
            capability, AzureDiskOptionSettings
        ), f"actual: {type(capability)}"

        assert (
            capability.disk_type
        ), "capability should have at least one disk type, but it's None"
        min_value = AzureDiskOptionSettings()
        cap_disk_type = capability.disk_type
        if isinstance(cap_disk_type, search_space.SetSpace):
            assert (
                len(cap_disk_type) > 0
            ), "capability should have at least one disk type, but it's empty"
        elif isinstance(cap_disk_type, schema.DiskType):
            cap_disk_type = search_space.SetSpace[schema.DiskType](
                is_allow_set=True, items=[cap_disk_type]
            )
        else:
            raise LisaException(
                f"unknown disk type on capability, type: {cap_disk_type}"
            )

        min_value.disk_type = (
            search_space.generate_min_capability_setspace_from_priority(
                self.disk_type, capability.disk_type, _ordered_disk_types
            )
        )

        # below values affect data disk only.
        if self.data_disk_count is not None or capability.data_disk_count is not None:
            min_value.data_disk_count = search_space.generate_min_capability_countspace(
                self.data_disk_count, capability.data_disk_count
            )

        if (
            self.max_data_disk_count is not None
            or capability.max_data_disk_count is not None
        ):
            min_value.max_data_disk_count = (
                search_space.generate_min_capability_countspace(
                    self.max_data_disk_count, capability.max_data_disk_count
                )
            )

        disk_type_iops = _disk_size_iops_map.get(min_value.disk_type, None)
        # ignore unsupported disk type like Ephemeral. It supports only os
        # disk. Calculate for iops, if it has value. If not, try disk size
        if disk_type_iops:
            if self.data_disk_iops:
                req_disk_iops = search_space.count_space_to_int_range(
                    self.data_disk_iops
                )
                cap_disk_iops = search_space.count_space_to_int_range(
                    capability.data_disk_iops
                )
                min_iops = max(req_disk_iops.min, cap_disk_iops.min)
                max_iops = min(req_disk_iops.max, cap_disk_iops.max)

                min_value.data_disk_iops = min(
                    iops
                    for iops, _ in disk_type_iops
                    if iops >= min_iops and iops <= max_iops
                )
                min_value.data_disk_size = self._get_disk_size_from_iops(
                    min_value.data_disk_iops, disk_type_iops
                )
            elif self.data_disk_size:
                req_disk_size = search_space.count_space_to_int_range(
                    self.data_disk_size
                )
                cap_disk_size = search_space.count_space_to_int_range(
                    capability.data_disk_size
                )
                min_size = max(req_disk_size.min, cap_disk_size.min)
                max_size = min(req_disk_size.max, cap_disk_size.max)

                min_value.data_disk_iops = min(
                    iops
                    for iops, disk_size in disk_type_iops
                    if disk_size >= min_size and disk_size <= max_size
                )
                min_value.data_disk_size = self._get_disk_size_from_iops(
                    min_value.data_disk_iops, disk_type_iops
                )
            else:
                # if req is not specified, query minimum value.
                cap_disk_size = search_space.count_space_to_int_range(
                    capability.data_disk_size
                )
                min_value.data_disk_iops = min(
                    iops
                    for iops, _ in disk_type_iops
                    if iops >= cap_disk_size.min and iops <= cap_disk_size.max
                )
                min_value.data_disk_size = self._get_disk_size_from_iops(
                    min_value.data_disk_iops, disk_type_iops
                )
        else:
            # The Ephemeral doesn't support data disk, but it needs a value.
            min_value.data_disk_iops = 0
            min_value.data_disk_size = 0

        # all caching types are supported, so just take the value from requirement.
        min_value.data_disk_caching_type = self.data_disk_caching_type

        min_value.has_resource_disk = self._generate_min_capability_has_resource_disk(
            self.has_resource_disk, capability.has_resource_disk
        )

        return min_value