def _per_node()

in src/hpc/autoscale/node/nodemanager.py [0:0]

• 29 lines of code
• 6 McCabe index (conditional complexity)

        def _per_node(node: Node, constraints: List[NodeConstraint]) -> int:
            min_space = minimum_space(constraints, node)

            if min_space == -1:
                min_space = remaining_slots()
            else:
                for constraint in constraints:
                    res = constraint.satisfied_by_node(node)
                    assert res, "{} {} {}".format(res, constraint, node.vcpu_count)
                    m = constraint.minimum_space(node)
                    assert (
                        m != 0
                    ), "{} satisfies node {} but minimum_space was {}".format(
                        constraint, node.name, m
                    )

            per_node = 1 if total_iterations < 0 else min(remaining_slots(), min_space)
            if per_node == 0:
                for constraint in constraints:
                    res = constraint.satisfied_by_node(node)
                    assert res, "{} {} {}".format(res, constraint, node.vcpu_count)
                    m = constraint.minimum_space(node)
                    assert (
                        m != 0
                    ), "{} satisfies node {} but minimum_space was {}".format(
                        constraint, node.name, m
                    )

            assert per_node > 0, "{} {} {} {}".format(
                per_node, remaining_slots(), node.resources, constraints
            )
            return per_node