Simplify hc.api.views.flips, add validation and more tests.

hc/api/forms.py

@@ -0,0 +1,28 @@
+from datetime import datetime as dt
+
+from django import forms
+from django.core.exceptions import ValidationError
+import pytz
+
+
+class TimestampField(forms.Field):
+    def to_python(self, value):
+        if value is None:
+            return None
+
+        try:
+            value_int = int(value)
+        except ValueError:
+            raise ValidationError(message="Must be an integer")
+
+        # 10000000000 is year 2286 (a sanity check)
+        if value_int < 0 or value_int > 10000000000:
+            raise ValidationError(message="Out of bounds")
+
+        return dt.fromtimestamp(value_int, pytz.UTC)
+
+
+class FlipsFiltersForm(forms.Form):
+    start = TimestampField(required=False)
+    end = TimestampField(required=False)
+    seconds = forms.IntegerField(required=False, min_value=0, max_value=31536000)

hc/api/models.py

@@ -16,7 +16,6 @@ from hc.api import transports
 from hc.lib import emails
 from hc.lib.date import month_boundaries
 import pytz
-import re
 
 STATUSES = (("up", "Up"), ("down", "Down"), ("new", "New"), ("paused", "Paused"))
 DEFAULT_TIMEOUT = td(days=1)

hc/api/tests/test_get_flips.py

@@ -19,12 +19,6 @@ class GetFlipsTestCase(BaseTestCase):
         self.a1.desc = "This is description"
         self.a1.save()
 
-        self.url = "/api/v1/checks/%s/flips/" % self.a1.code
-
-    def get(self, api_key="X" * 32):
-        return self.client.get(self.url, HTTP_X_API_KEY=api_key)
-
-    def test_it_works(self):
         Flip.objects.create(
             owner=self.a1,
             created=dt(2020, 6, 1, 12, 24, 32, 123000, tzinfo=timezone.utc),
@@ -32,6 +26,13 @@ class GetFlipsTestCase(BaseTestCase):
             new_status="up",
         )
 
+        self.url = "/api/v1/checks/%s/flips/" % self.a1.code
+
+    def get(self, api_key="X" * 32, qs=""):
+        return self.client.get(self.url + qs, HTTP_X_API_KEY=api_key)
+
+    def test_it_works(self):
+
         r = self.get()
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r["Access-Control-Allow-Origin"], "*")
@@ -54,3 +55,23 @@ class GetFlipsTestCase(BaseTestCase):
     def test_it_rejects_post(self):
         r = self.client.post(self.url, HTTP_X_API_KEY="X" * 32)
         self.assertEqual(r.status_code, 405)
+
+    def test_it_rejects_non_integer_start(self):
+        r = self.get(qs="?start=abc")
+        self.assertEqual(r.status_code, 400)
+
+    def test_it_rejects_negative_start(self):
+        r = self.get(qs="?start=-123")
+        self.assertEqual(r.status_code, 400)
+
+    def test_it_rejects_huge_start(self):
+        r = self.get(qs="?start=12345678901234567890")
+        self.assertEqual(r.status_code, 400)
+
+    def test_it_rejects_negative_seconds(self):
+        r = self.get(qs="?seconds=-123")
+        self.assertEqual(r.status_code, 400)
+
+    def test_it_rejects_huge_seconds(self):
+        r = self.get(qs="?seconds=12345678901234567890")
+        self.assertEqual(r.status_code, 400)

hc/api/views.py

@@ -1,5 +1,4 @@
 from datetime import timedelta as td
-from datetime import datetime
 import time
 import uuid
 
@@ -21,6 +20,7 @@ from django.views.decorators.http import require_POST
 from hc.accounts.models import Profile
 from hc.api import schemas
 from hc.api.decorators import authorize, authorize_read, cors, validate_json
+from hc.api.forms import FlipsFiltersForm
 from hc.api.models import MAX_DELTA, Flip, Channel, Check, Notification, Ping
 from hc.lib.badges import check_signature, get_badge_svg
 
@@ -292,42 +292,38 @@ def pings(request, code):
 
     return JsonResponse({"pings": dicts})
 
+
 def flips(request, check):
     if check.project_id != request.project.id:
         return HttpResponseForbidden()
 
-    if all(x not in request.GET for x in ('start','end','seconds')):
-        flips = Flip.objects.filter(
-            owner=check, new_status__in=("down","up"),
-        ).order_by("created")
-    else:
-        if "seconds" in request.GET and ("start" in request.GET or "end" in request.GET):
-            return HttpResponseBadRequest()
+    form = FlipsFiltersForm(request.GET)
+    if not form.is_valid():
+        return HttpResponseBadRequest()
 
-        flips = Flip.objects.filter(
-            owner=check, new_status__in=("down","up"))
+    flips = Flip.objects.filter(owner=check).order_by("-id")
 
-        if 'start' in request.GET:
-            flips = flips.filter(created_gt=datetime.fromtimestamp(int(request.GET['start'])))
-        
-        if 'end' in request.GET:
-            flips = flips.filter(created__lt=datetime.fromtimestamp(int(request.GET['end'])))
+    if form.cleaned_data["start"]:
+        flips = flips.filter(created__gte=form.cleaned_data["start"])
 
-        if 'seconds' in request.GET:
-            flips = flips.filter(created_gt=datetime.now()-td(seconds=int(request.GET['seconds'])))
+    if form.cleaned_data["end"]:
+        flips = flips.filter(created__lt=form.cleaned_data["end"])
 
-        flips = flips.order_by("created")
-        
+    if form.cleaned_data["seconds"]:
+        threshold = timezone.now() - td(seconds=form.cleaned_data["seconds"])
+        flips = flips.filter(created__gte=threshold)
 
     return JsonResponse({"flips": [flip.to_dict() for flip in flips]})
 
+
 @cors("GET")
 @csrf_exempt
 @validate_json()
 @authorize_read
-def flips_by_uuid(request,code):
+def flips_by_uuid(request, code):
     check = get_object_or_404(Check, code=code)
-    return flips(request,check)
+    return flips(request, check)
+
 
 @cors("GET")
 @csrf_exempt
@@ -337,9 +333,10 @@ def flips_by_unique_key(request, unique_key):
     checks = Check.objects.filter(project=request.project.id)
     for check in checks:
         if check.unique_key == unique_key:
-            return flips(request,check)
+            return flips(request, check)
     return HttpResponseNotFound()
 
+
 @never_cache
 @cors("GET")
 def badge(request, badge_key, signature, tag, fmt="svg"):