def get_load_balancers_details()

in src/modules/get_azure_lb.py [0:0]

• 99 lines of code
• 19 McCabe index (conditional complexity)

    def get_load_balancers_details(self) -> dict:
        """
        Get the details of the load balancers in a specific resource group.

        :return: Dictionary containing the result of the test case.
        :rtype: dict
        """
        self._create_network_client()

        if self.result["status"] == TestStatus.ERROR.value:
            return self.result

        load_balancers = self.get_load_balancers()

        if self.result["status"] == TestStatus.ERROR.value:
            return self.result

        inbound_rules = ast.literal_eval(self.module_params["inbound_rules"])
        load_balancer_ips = list(
            inbound_rule["privateIpAddress"]
            for inbound_rule in inbound_rules
            if "privateIpAddress" in inbound_rule
        )
        found_load_balancer = None

        found_load_balancer = next(
            (
                lb
                for lb in load_balancers
                for frontend_ip_config in lb["frontend_ip_configurations"]
                if frontend_ip_config["private_ip_address"] in load_balancer_ips
            ),
            None,
        )
        parameters = []

        def check_parameters(entity, parameters_dict, entity_type):
            for key, expected_value in parameters_dict.items():
                parameters.append(
                    Parameters(
                        category=entity_type,
                        id=entity["name"],
                        name=key,
                        value=str(entity[key]),
                        expected_value=str(expected_value),
                        status=(
                            TestStatus.SUCCESS.value
                            if entity[key] == expected_value
                            else TestStatus.ERROR.value
                        ),
                    ).to_dict()
                )

        try:
            if found_load_balancer:
                self.log(
                    logging.INFO,
                    f"Found load balancer {found_load_balancer['name']}",
                )
                self.result[
                    "message"
                ] += f"Validating load balancer parameters {found_load_balancer['name']}"
                for rule in found_load_balancer["load_balancing_rules"]:
                    try:
                        check_parameters(
                            rule,
                            self.constants["RULES"],
                            "load_balancing_rule",
                        )
                    except Exception as ex:
                        self.handle_error(ex)
                        self.result[
                            "message"
                        ] += f"Failed to validate load balancer rule parameters. {ex} \n"
                        continue

                for probe in found_load_balancer["probes"]:
                    try:
                        check_parameters(
                            probe,
                            self.constants["PROBES"],
                            "probes",
                        )
                    except Exception as ex:
                        self.handle_error(ex)
                        self.result[
                            "message"
                        ] += f"Failed to validate load balancer probe parameters. {ex} \n"
                        continue

                failed_parameters = [
                    param
                    for param in parameters
                    if param.get("status", TestStatus.ERROR.value) == TestStatus.ERROR.value
                ]
                self.result.update(
                    {
                        "details": {"parameters": parameters},
                        "status": (
                            TestStatus.ERROR.value
                            if failed_parameters
                            else TestStatus.SUCCESS.value
                        ),
                    }
                )
                self.result["message"] += "Successfully validated load balancer parameters"
            else:
                self.result["message"] += (
                    "Load Balancer details not fetched."
                    " Ensure that the Managed Identity (MSI) has sufficient permissions "
                    "to access the load balancer details."
                )

        except Exception as ex:
            self.handle_error(ex)