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"