monitor.http-check-result

from typing import Optional, Sequence

from .monitor_http_check_result_types import CheckOutcome, HttpExpectation, HttpProbe

_ERROR_MESSAGES = {
    "timeout": "timed out",
    "dns": "DNS lookup failed",
    "connection": "connection failed",
    "tls": "TLS handshake failed",
    "protocol": "protocol error",
}


def _whole(v: object) -> bool:
    return isinstance(v, int) and not isinstance(v, bool)


def _expected_text(statuses: Optional[Sequence[int]]) -> str:
    if statuses is None:
        return "2xx"
    if len(statuses) == 1:
        return str(statuses[0])
    return "%s or %d" % (", ".join(str(s) for s in statuses[:-1]), statuses[-1])


def _threshold(name: str, v: Optional[int]) -> None:
    if v is not None and (not _whole(v) or v < 1):
        raise ValueError("%s must be a whole number 1 or more, received %r" % (name, v))


def classify_http_check(probe: HttpProbe, expect: HttpExpectation) -> CheckOutcome:
    """One probe, one verdict, checked in a fixed order so the reason is the
    most fundamental thing wrong: a transport error, then the status code,
    then the body, then latency. A 503 that also took ten seconds is
    reported as the 503, because that is what someone has to fix."""
    status_code, latency_ms, body, error = probe.status_code, probe.latency_ms, probe.body, probe.error
    if status_code is not None and (not _whole(status_code) or status_code < 100 or status_code > 599):
        raise ValueError("statusCode must be 100 to 599, received %r" % (status_code,))
    if error is not None and error not in _ERROR_MESSAGES:
        raise ValueError("unknown probe error: %s" % (error,))
    if status_code is None and error is None:
        raise ValueError("probe must have a statusCode or an error")
    if latency_ms is not None and (not _whole(latency_ms) or latency_ms < 0):
        raise ValueError("latencyMs must be a whole number 0 or more, received %r" % (latency_ms,))
    statuses, body_contains = expect.statuses, expect.body_contains
    degraded, down = expect.degraded_latency_ms, expect.down_latency_ms
    if statuses is not None:
        if len(statuses) == 0:
            raise ValueError("statuses must not be empty; use null to accept any 2xx")
        for s in statuses:
            if not _whole(s) or s < 100 or s > 599:
                raise ValueError("expected statuses must be 100 to 599, received %r" % (s,))
    _threshold("degradedLatencyMs", degraded)
    _threshold("downLatencyMs", down)
    if degraded is not None and down is not None and down < degraded:
        raise ValueError("downLatencyMs must not be below degradedLatencyMs: %d < %d" % (down, degraded))

    if error is not None:
        return CheckOutcome(status="down", reason=error, message=_ERROR_MESSAGES[error])
    assert status_code is not None
    accepted = 200 <= status_code <= 299 if statuses is None else status_code in statuses
    if not accepted:
        return CheckOutcome(status="down", reason="unexpected-status", message="HTTP %d, expected %s" % (status_code, _expected_text(statuses)))
    if body_contains is not None and (body is None or body_contains not in body):
        return CheckOutcome(status="down", reason="body-mismatch", message='body does not contain "%s"' % (body_contains,))
    if latency_ms is not None:
        if down is not None and latency_ms >= down:
            return CheckOutcome(status="down", reason="too-slow", message="latency %d ms at or above %d ms" % (latency_ms, down))
        if degraded is not None and latency_ms >= degraded:
            return CheckOutcome(status="degraded", reason="slow", message="latency %d ms at or above %d ms" % (latency_ms, degraded))
    message = "HTTP %d" % status_code if latency_ms is None else "HTTP %d in %d ms" % (status_code, latency_ms)
    return CheckOutcome(status="up", reason="ok", message=message)