Add tests for the live tracking API endpoints

tests/api/views/tracking/__init__.py

@@ -0,0 +1,36 @@
+def decode_time(encoded_time):
+    """Decodes an encoded time string"""
+    index = 0
+    time = 0
+    times = []
+
+    while index < len(encoded_time):
+        byte = None
+        result = 0
+        shift = 0
+        while byte is None or byte >= 0x20:
+            byte = ord(encoded_time[index]) - 63
+            index += 1
+            result |= (byte & 0x1F) << shift
+            shift += 5
+            comp = result & 1
+        delta = ~(result >> 1) if comp else (result >> 1)
+        time += delta
+        times.append(time)
+    return times
+
+
+def get_fixes_times_seconds(fixes):
+    """Returns the times of the given fixes in second of day"""
+    start_time = fixes[0].time
+    start_second_of_day = (
+        start_time.hour * 3600 + start_time.minute * 60 + start_time.second
+    )
+
+    seconds = [start_second_of_day]
+
+    for index in range(1, len(fixes)):
+        time = fixes[index].time
+        seconds.append(int((time - start_time).total_seconds() + start_second_of_day))
+
+    return seconds

tests/api/views/tracking/latest_test.py

@@ -0,0 +1,40 @@
+from mock import patch
+from datetime import datetime, timedelta
+from tests.data import add_fixtures, users, live_fix
+
+
+def test_get_latest_default_max_age(db_session, client):
+    """The default max_age is 6 hours"""
+    utcnow = datetime(year=2020, month=12, day=20, hour=12)
+
+    # Only the latest_fix should be returned
+    john = users.john()
+    fix = live_fix.create(john, utcnow - timedelta(hours=2), 10, 20)
+    latest_fix = live_fix.create(john, utcnow - timedelta(minutes=5), 11, 21)
+
+    # Fix older than 6h should not be returned
+    jane = users.jane()
+    old_fix = live_fix.create(jane, utcnow - timedelta(hours=7), 12, 22)
+
+    add_fixtures(db_session, john, fix, latest_fix, jane, old_fix)
+
+    with patch("skylines.model.tracking.datetime") as datetime_mock:
+        datetime_mock.utcnow.return_value = utcnow
+
+        res = client.get("/tracking/latest.json")
+
+        assert res.status_code == 200
+        assert res.json == {
+            u"fixes": [
+                {
+                    u"airspeed": 10,
+                    u"altitude": 100,
+                    u"ground_speed": 10,
+                    u"location": u"POINT(11.0 21.0)",
+                    u"pilot": {u"id": john.id, u"name": u"John Doe"},
+                    u"time": u"2020-12-20T11:55:00Z",
+                    u"track": 0,
+                    u"vario": 0,
+                }
+            ]
+        }

tests/api/views/tracking/live_multiple_test.py

@@ -0,0 +1,101 @@
+from mock import patch
+from datetime import datetime, timedelta
+
+from tests.data import add_fixtures, users, live_fix
+from tests.api.views.tracking import get_fixes_times_seconds, decode_time
+
+
+def test_get_live_default_max_age(db_session, client):
+    """The default max_age is 12 hours"""
+    utcnow = datetime(year=2020, month=12, day=20, hour=20)
+
+    john = users.john()
+    john_fixes = []
+    for age_hour in range(14, 0, -1):
+        time = utcnow - timedelta(hours=age_hour)
+        john_fixes.append(live_fix.create(john, time, 10, 20))
+
+    jane = users.jane()
+    jane_fixes = []
+    for age_hour in range(14, 0, -1):
+        time = utcnow - timedelta(hours=age_hour, minutes=30)
+        jane_fixes.append(live_fix.create(jane, time, 11, 21))
+
+    add_fixtures(db_session, john, *(john_fixes + jane_fixes))
+
+    with patch("skylines.model.tracking.datetime") as datetime_mock:
+        datetime_mock.utcnow.return_value = utcnow
+
+        res = client.get("/live/{john},{jane}".format(john=john.id, jane=jane.id))
+
+        assert res.status_code == 200
+        json = res.json
+
+        assert json == {
+            u"flights": [
+                {
+                    u"additional": {u"color": u"#2b56d8", u"competition_id": u"JD"},
+                    u"barogram_h": u"eE???????????",
+                    u"barogram_t": u"_gw@_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F",
+                    u"contests": None,
+                    u"elevations_h": u"????????????",
+                    u"elevations_t": u"_gw@_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F",
+                    u"enl": u"",
+                    u"geoid": 26.504,
+                    u"points": u"_gayB_c`|@??????????????????????",
+                    u"sfid": john.id,
+                },
+                {
+                    u"additional": {u"color": u"#822bd8", u"competition_id": u"JD"},
+                    u"barogram_h": u"eE??????????",
+                    u"barogram_t": u"owz@_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F",
+                    u"contests": None,
+                    u"elevations_h": u"???????????",
+                    u"elevations_t": u"owz@_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F",
+                    u"enl": u"",
+                    u"geoid": 25.013,
+                    u"points": u"_qd_C_mcbA????????????????????",
+                    u"sfid": jane.id,
+                },
+            ],
+            u"pilots": [
+                {
+                    u"club": None,
+                    u"color": u"#2b56d8",
+                    u"firstName": u"John",
+                    u"followers": 0,
+                    u"following": 0,
+                    u"id": john.id,
+                    u"lastName": u"Doe",
+                    u"name": u"John Doe",
+                    u"trackingCallsign": None,
+                    u"trackingDelay": 0,
+                },
+                {
+                    u"club": None,
+                    u"color": u"#822bd8",
+                    u"firstName": u"Jane",
+                    u"followers": 0,
+                    u"following": 0,
+                    u"id": jane.id,
+                    u"lastName": u"Doe",
+                    u"name": u"Jane Doe",
+                    u"trackingCallsign": None,
+                    u"trackingDelay": 0,
+                },
+            ],
+        }
+
+        expected_fixes = list(
+            filter(lambda f: f.time >= utcnow - timedelta(hours=12), john_fixes)
+        )
+        assert decode_time(
+            json[u"flights"][0][u"barogram_t"]
+        ) == get_fixes_times_seconds(expected_fixes)
+
+        expected_fixes = list(
+            filter(lambda f: f.time >= utcnow - timedelta(hours=12), jane_fixes)
+        )
+        assert decode_time(
+            json[u"flights"][1][u"barogram_t"]
+        ) == get_fixes_times_seconds(expected_fixes)

tests/api/views/tracking/live_test.py

@@ -0,0 +1,87 @@
+from mock import patch
+from datetime import datetime, timedelta
+
+from tests.data import add_fixtures, users, live_fix
+from tests.api.views.tracking import get_fixes_times_seconds, decode_time
+
+
+def test_get_live_default_max_age(db_session, client):
+    """The default max_age is 12 hours"""
+    utcnow = datetime(year=2020, month=12, day=20, hour=20)
+
+    john = users.john()
+    fixes = []
+    for age_hour in range(14, 0, -1):
+        time = utcnow - timedelta(hours=age_hour)
+        fixes.append(live_fix.create(john, time, 10, 20))
+
+    add_fixtures(db_session, john, *fixes)
+
+    with patch("skylines.model.tracking.datetime") as datetime_mock:
+        datetime_mock.utcnow.return_value = utcnow
+
+        res = client.get("/live/{id}/json".format(id=john.id))
+
+        assert res.status_code == 200
+        json = res.json
+
+        assert json == {
+            u"barogram_h": u"eE???????????",
+            u"barogram_t": u"_gw@_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F_`F",
+            u"elevations": u"????????????",
+            u"enl": u"",
+            u"geoid": 26.504,
+            u"points": u"_gayB_c`|@??????????????????????",
+            u"sfid": john.id,
+        }
+
+        expected_fixes = list(
+            filter(lambda f: f.time >= utcnow - timedelta(hours=12), fixes)
+        )
+        assert decode_time(json[u"barogram_t"]) == get_fixes_times_seconds(
+            expected_fixes
+        )
+
+
+def test_get_live_filter_by_last_update(db_session, client):
+    utcnow = datetime(year=2020, month=12, day=20, hour=20)
+    last_update = 50000
+    start_time = datetime(year=2020, month=12, day=20) + timedelta(seconds=last_update)
+
+    john = users.john()
+    fixes = []
+    for age_hour in range(14, 0, -1):
+        time = utcnow - timedelta(hours=age_hour)
+        fixes.append(live_fix.create(john, time, 10, 20))
+
+    add_fixtures(db_session, john, *fixes)
+
+    with patch("skylines.model.tracking.datetime") as datetime_mock:
+        datetime_mock.utcnow.return_value = utcnow
+
+        res = client.get(
+            "/live/{id}/json?last_update={last_update}".format(
+                id=john.id, last_update=last_update
+            )
+        )
+
+        assert res.status_code == 200
+        json = res.json
+
+        print(decode_time(json[u"barogram_t"]))
+        print(get_fixes_times_seconds(fixes))
+
+        assert json == {
+            u"barogram_h": u"eE?????",
+            u"barogram_t": u"_maB_`F_`F_`F_`F_`F",
+            u"elevations": u"??????",
+            u"enl": u"",
+            u"geoid": 26.504,
+            u"points": u"_gayB_c`|@??????????",
+            u"sfid": john.id,
+        }
+
+        expected_fixes = list(filter(lambda f: f.time >= start_time, fixes))
+        assert decode_time(json[u"barogram_t"]) == get_fixes_times_seconds(
+            expected_fixes
+        )

tests/data/live_fix.py

@@ -0,0 +1,20 @@
+from skylines.model import TrackingFix
+
+
+def create(pilot, time, latitude, longitude):
+    """Creates a test fix for the passed pilot"""
+    fix = TrackingFix(
+        pilot=pilot,
+        time=time,
+        time_visible=time,
+        track=0,
+        ground_speed=10,
+        airspeed=10,
+        altitude=100,
+        elevation=0,
+        vario=0,
+    )
+
+    fix.set_location(latitude, longitude)
+
+    return fix