impl/python.py

from typing import Sequence

from .monitor_check_status_types import CheckStatus


def aggregate_status(statuses: Sequence[CheckStatus], down_quorum: int) -> CheckStatus:
    """One status for a target probed from several places. Down needs a
    quorum, so one location with a broken route cannot page anyone; a down
    vote below the quorum still counts as degraded, because something is
    wrong somewhere."""
    if len(statuses) == 0:
        raise ValueError("statuses must not be empty")
    if not isinstance(down_quorum, int) or isinstance(down_quorum, bool) or down_quorum < 1 or down_quorum > len(statuses):
        raise ValueError("downQuorum must be 1 to %d, received %r" % (len(statuses), down_quorum))
    down = 0
    degraded = 0
    for s in statuses:
        if s == "down":
            down += 1
        elif s == "degraded":
            degraded += 1
        elif s != "up":
            raise ValueError("unknown check status: %s" % (s,))
    if down >= down_quorum:
        return "down"
    return "degraded" if down > 0 or degraded > 0 else "up"