GlitchTip/glitchtip/uptime/serializers.py

from urllib.parse import urlparse

from django.conf import settings
from django.core.validators import URLValidator
from django.urls import reverse
from django.utils import timezone
from rest_framework import serializers
from rest_framework.fields import ChoiceField

from .constants import HTTP_MONITOR_TYPES
from .models import Monitor, MonitorCheck, MonitorType, StatusPage


class MonitorCheckRelatedSerializer(serializers.ModelSerializer):
    isUp = serializers.BooleanField(source="is_up")
    startCheck = serializers.DateTimeField(source="start_check")

    class Meta:
        model = MonitorCheck
        fields = ("isUp", "startCheck", "reason")


class MonitorCheckSerializer(MonitorCheckRelatedSerializer):
    responseTime = serializers.DurationField(source="response_time")

    class Meta:
        model = MonitorCheck
        fields = ("isUp", "startCheck", "reason", "responseTime")


class HeartBeatCheckSerializer(MonitorCheckSerializer):
    start_check = serializers.DateTimeField(
        default=timezone.now, help_text="Optional, set server check start time."
    )

    class Meta(MonitorCheckSerializer.Meta):
        fields = ("is_up", "start_check")
        read_only_fields = ("is_up",)


class MonitorSerializer(serializers.ModelSerializer):
    isUp = serializers.SerializerMethodField()
    lastChange = serializers.SerializerMethodField()
    monitorType = ChoiceField(choices=MonitorType.choices, source="monitor_type")
    expectedStatus = serializers.IntegerField(source="expected_status", allow_null=True)
    expectedBody = serializers.CharField(
        source="expected_body", required=False, allow_blank=True
    )
    heartbeatEndpoint = serializers.SerializerMethodField()
    projectName = serializers.SerializerMethodField()
    envName = serializers.SerializerMethodField()
    checks = MonitorCheckRelatedSerializer(many=True, read_only=True)

    def get_isUp(self, obj):
        if hasattr(obj, "latest_is_up"):
            return obj.latest_is_up

    def get_lastChange(self, obj):
        if hasattr(obj, "last_change"):
            if obj.last_change:
                return obj.last_change.isoformat().replace("+00:00", "Z")

    def get_heartbeatEndpoint(self, obj):
        if obj.endpoint_id:
            return settings.GLITCHTIP_URL.geturl() + reverse(
                "heartbeat-check",
                kwargs={
                    "organization_slug": obj.organization.slug,
                    "endpoint_id": obj.endpoint_id,
                },
            )

    def get_projectName(self, obj):
        if obj.project:
            return obj.project.name

    def get_envName(self, obj):
        if obj.environment:
            return obj.environment.name

    class Meta:
        model = Monitor
        fields = (
            "id",
            "monitorType",
            "endpoint_id",
            "created",
            "checks",
            "name",
            "url",
            "expectedStatus",
            "expectedBody",
            "environment",
            "project",
            "organization",
            "interval",
            "isUp",
            "lastChange",
            "heartbeatEndpoint",
            "projectName",
            "envName",
            "timeout",
        )
        read_only_fields = (
            "organization",
            "isUp",
            "lastChange",
            "endpoint_id",
            "created",
            "heartbeatEndpoint",
            "projectName",
            "envName",
        )

    def validate(self, attrs):
        monitor_type = attrs["monitor_type"]
        if attrs["url"] == "" and monitor_type in HTTP_MONITOR_TYPES + (
            MonitorType.SSL,
        ):
            raise serializers.ValidationError("URL is required for " + monitor_type)

        if monitor_type in HTTP_MONITOR_TYPES:
            URLValidator()(attrs["url"])

        if attrs.get("expected_status") is None and monitor_type in [
            MonitorType.GET,
            MonitorType.POST,
        ]:
            raise serializers.ValidationError(
                "Expected status is required for " + attrs["monitor_type"]
            )

        if attrs["monitor_type"] == MonitorType.PORT:
            url = attrs["url"].replace("http://", "//", 1)
            if not url.startswith("//"):
                url = "//" + url
            parsed_url = urlparse(url)
            message = "Invalid Port URL, expected hostname and port"
            try:
                if not all([parsed_url.hostname, parsed_url.port]):
                    raise serializers.ValidationError(message)
            except ValueError as err:
                raise serializers.ValidationError(message) from err
            attrs["url"] = f"{parsed_url.hostname}:{parsed_url.port}"

        return attrs


class MonitorDetailSerializer(MonitorSerializer):
    checks = MonitorCheckSerializer(many=True, read_only=True)


class MonitorUpdateSerializer(MonitorSerializer):
    """Remove checks, which otherwise would be have the limit applied"""

    class Meta(MonitorSerializer.Meta):
        fields = [field for field in MonitorSerializer.Meta.fields if field != "checks"]


class StatusPageSerializer(serializers.ModelSerializer):
    isPublic = serializers.BooleanField(source="is_public")

    class Meta:
        model = StatusPage
        fields = ("name", "slug", "isPublic", "monitors")
        read_only_fields = ("slug",)