def _generate_node_table()

in azext_edge/edge/providers/check/base/node.py [0:0]

• 69 lines of code
• 9 McCabe index (conditional complexity)

def _generate_node_table(check_manager: CheckManager, nodes: V1NodeList) -> Table:
    from kubernetes.utils import parse_quantity
    # prep table
    table = Table(
        show_header=True, header_style="bold", show_lines=True, caption_justify="left"
    )
    for column_name, justify in [
        ("Name", "left"),
        ("Architecture", "right"),
        ("CPU (vCPU)", "right"),
        ("Memory (GB)", "right"),
        ("Storage (GB)", "right"),
    ]:
        table.add_column(column_name, justify=f"{justify}")
    table.add_row(*[COLOR_STR_FORMAT.format(color="cyan", value=value) for value in [
        "Minimum requirements",
        ", ".join(AIO_SUPPORTED_ARCHITECTURES),
        MIN_NODE_VCPU,
        MIN_NODE_MEMORY[:-1],
        MIN_NODE_STORAGE[:-1]
    ]])
    node: V1Node
    for node in nodes.items:
        node_name = node.metadata.name

        # check_manager target for node
        node_target = f"cluster/nodes/{node_name}"
        check_manager.add_target(target_name=node_target)

        # verify architecture
        # build node table row
        row_status = CheckTaskStatus.success
        row_cells = []
        for condition, expected, actual in [
            (
                "info.architecture",
                AIO_SUPPORTED_ARCHITECTURES,
                node.status.node_info.architecture,
            ),
            (
                "condition.cpu",
                MIN_NODE_VCPU,
                parse_quantity(node.status.capacity.get("cpu", 0)),
            ),
            (
                "condition.memory",
                MIN_NODE_MEMORY,
                parse_quantity(node.status.capacity.get("memory", 0)),
            ),
            (
                "condition.ephemeral-storage",
                MIN_NODE_STORAGE,
                parse_quantity(node.status.capacity.get("ephemeral-storage", 0)),
            ),
        ]:
            # determine strings, expected, status
            condition_str = f"{condition}>={expected}"
            displayed = actual
            cell_status = CheckTaskStatus.success
            if isinstance(expected, list):
                condition_str = f"{condition} in ({','.join(expected)})"
                if actual not in expected:
                    row_status = CheckTaskStatus.error
                    cell_status = CheckTaskStatus.error
            else:
                displayed = _get_display_number(displayed, expected)
                expected = parse_quantity(expected)
                if actual < expected:
                    row_status = CheckTaskStatus.error
                    cell_status = CheckTaskStatus.error
                actual = int(actual)

            check_manager.add_target_conditions(target_name=node_target, conditions=[condition_str])
            check_manager.add_target_eval(target_name=node_target, status=cell_status.value, value={condition: actual})
            row_cells.append(COLOR_STR_FORMAT.format(color=cell_status.color, value=displayed))

        # overall node name color
        table.add_row(COLOR_STR_FORMAT.format(color=row_status.color, value=node_name), *row_cells)
    return table