Http tests (#3350)

* Set up for http tests

* Setup basics for testing and first test

* Add testing consts

* Cleanup db creation

* Add one more check to test

* Get event that does not exist

* Get events working with cleaner db

* Test retain / un-retain

* Test setting and deleting sub label

* Test getting list of sub labels

* Fix bug caught in tests

* Test deleting event

* Test geting list of events

* Expand test

* Test more event filters

* Write version module so tests don't fail on version import

* Test config

* Test recordings endpoint

* Formatting

* Remove unused imports

* Test stats

* Add cleanup files in const

* Add name to match other checks

.github/workflows/pull_request.yml

@@ -54,6 +54,7 @@ jobs:
 
   python_tests:
     runs-on: ubuntu-latest
+    name: Python Tests
     steps:
       - name: Check out code
         uses: actions/checkout@v2
@@ -69,6 +70,8 @@ jobs:
         uses: docker/setup-qemu-action@v1
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v1
+      - name: Create Version Module
+        run: make version
       - name: Build
         run: make
       - name: Run mypy

frigate/http.py

@@ -256,7 +256,10 @@ def get_sub_labels():
         )
 
     sub_labels = [e.sub_label for e in events]
-    sub_labels.remove(None)
+
+    if None in sub_labels:
+        sub_labels.remove(None)
+
     return jsonify(sub_labels)
 
 

frigate/test/const.py

@@ -0,0 +1,4 @@
+"""Consts for testing."""
+
+TEST_DB = "test.db"
+TEST_DB_CLEANUPS = ["test.db", "test.db-shm", "test.db-wal"]

frigate/test/test_http.py

@@ -0,0 +1,327 @@
+import datetime
+import json
+import logging
+import os
+import unittest
+from unittest.mock import patch
+
+from peewee_migrate import Router
+from playhouse.sqlite_ext import SqliteExtDatabase
+from playhouse.sqliteq import SqliteQueueDatabase
+from playhouse.shortcuts import model_to_dict
+
+from frigate.config import FrigateConfig
+from frigate.http import create_app
+from frigate.models import Event, Recordings
+
+from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
+
+
+class TestHttp(unittest.TestCase):
+    def setUp(self):
+        # setup clean database for each test run
+        migrate_db = SqliteExtDatabase("test.db")
+        del logging.getLogger("peewee_migrate").handlers[:]
+        router = Router(migrate_db)
+        router.run()
+        migrate_db.close()
+        self.db = SqliteQueueDatabase(TEST_DB)
+        models = [Event, Recordings]
+        self.db.bind(models)
+
+        self.minimal_config = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "front_door": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                        ]
+                    },
+                    "detect": {
+                        "height": 1080,
+                        "width": 1920,
+                        "fps": 5,
+                    },
+                }
+            },
+        }
+        self.test_stats = {
+            "detection_fps": 13.7,
+            "detectors": {
+                "cpu1": {
+                    "detection_start": 0.0,
+                    "inference_speed": 91.43,
+                    "pid": 42,
+                },
+                "cpu2": {
+                    "detection_start": 0.0,
+                    "inference_speed": 84.99,
+                    "pid": 44,
+                },
+            },
+            "front_door": {
+                "camera_fps": 0.0,
+                "capture_pid": 53,
+                "detection_fps": 0.0,
+                "pid": 52,
+                "process_fps": 0.0,
+                "skipped_fps": 0.0,
+            },
+            "service": {
+                "storage": {
+                    "/dev/shm": {
+                        "free": 50.5,
+                        "mount_type": "tmpfs",
+                        "total": 67.1,
+                        "used": 16.6,
+                    },
+                    "/media/frigate/clips": {
+                        "free": 42429.9,
+                        "mount_type": "ext4",
+                        "total": 244529.7,
+                        "used": 189607.0,
+                    },
+                    "/media/frigate/recordings": {
+                        "free": 0.2,
+                        "mount_type": "ext4",
+                        "total": 8.0,
+                        "used": 7.8,
+                    },
+                    "/tmp/cache": {
+                        "free": 976.8,
+                        "mount_type": "tmpfs",
+                        "total": 1000.0,
+                        "used": 23.2,
+                    },
+                },
+                "uptime": 101113,
+                "version": "0.10.1",
+                "latest_version": "0.11",
+            },
+        }
+
+    def tearDown(self):
+        if not self.db.is_closed():
+            self.db.close()
+
+        try:
+            for file in TEST_DB_CLEANUPS:
+                os.remove(file)
+        except OSError:
+            pass
+
+    def test_get_event_list(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+        id2 = "7890.random"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            events = client.get(f"/events").json
+            assert events
+            assert len(events) == 1
+            assert events[0]["id"] == id
+            _insert_mock_event(id2)
+            events = client.get(f"/events").json
+            assert events
+            assert len(events) == 2
+            events = client.get(
+                f"/events",
+                query_string={"limit": 1},
+            ).json
+            assert events
+            assert len(events) == 1
+            events = client.get(
+                f"/events",
+                query_string={"has_clip": 0},
+            ).json
+            assert not events
+
+    def test_get_good_event(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            event = client.get(f"/events/{id}").json
+
+        assert event
+        assert event["id"] == id
+        assert event == model_to_dict(Event.get(Event.id == id))
+
+    def test_get_bad_event(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+        bad_id = "654321.other"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            event = client.get(f"/events/{bad_id}").json
+
+        assert not event
+
+    def test_delete_event(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            event = client.get(f"/events/{id}").json
+            assert event
+            assert event["id"] == id
+            client.delete(f"/events/{id}")
+            event = client.get(f"/events/{id}").json
+            assert not event
+
+    def test_event_retention(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            client.post(f"/events/{id}/retain")
+            event = client.get(f"/events/{id}").json
+            assert event
+            assert event["id"] == id
+            assert event["retain_indefinitely"] == True
+            client.delete(f"/events/{id}/retain")
+            event = client.get(f"/events/{id}").json
+            assert event
+            assert event["id"] == id
+            assert event["retain_indefinitely"] == False
+
+    def test_set_delete_sub_label(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+        sub_label = "sub"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            client.post(
+                f"/events/{id}/sub_label",
+                data=json.dumps({"subLabel": sub_label}),
+                content_type="application/json",
+            )
+            event = client.get(f"/events/{id}").json
+            assert event
+            assert event["id"] == id
+            assert event["sub_label"] == sub_label
+            client.post(
+                f"/events/{id}/sub_label",
+                data=json.dumps({"subLabel": ""}),
+                content_type="application/json",
+            )
+            event = client.get(f"/events/{id}").json
+            assert event
+            assert event["id"] == id
+            assert event["sub_label"] == ""
+
+    def test_sub_label_list(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config), self.db, None, None, None
+        )
+        id = "123456.random"
+        sub_label = "sub"
+
+        with app.test_client() as client:
+            _insert_mock_event(id)
+            client.post(
+                f"/events/{id}/sub_label",
+                data=json.dumps({"subLabel": sub_label}),
+                content_type="application/json",
+            )
+            sub_labels = client.get("/sub_labels").json
+            assert sub_labels
+            assert sub_labels == [sub_label]
+
+    def test_config(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config).runtime_config,
+            self.db,
+            None,
+            None,
+            None,
+        )
+
+        with app.test_client() as client:
+            config = client.get("/config").json
+            assert config
+            assert config["cameras"]["front_door"]
+
+    def test_recordings(self):
+        app = create_app(
+            FrigateConfig(**self.minimal_config).runtime_config,
+            self.db,
+            None,
+            None,
+            None,
+        )
+        id = "123456.random"
+
+        with app.test_client() as client:
+            _insert_mock_recording(id)
+            recording = client.get("/front_door/recordings").json
+            assert recording
+            assert recording[0]["id"] == id
+
+    @patch("frigate.http.stats_snapshot")
+    def test_stats(self, mock_stats):
+        app = create_app(
+            FrigateConfig(**self.minimal_config).runtime_config,
+            self.db,
+            None,
+            None,
+            None,
+        )
+        mock_stats.return_value = self.test_stats
+
+        with app.test_client() as client:
+            stats = client.get("/stats").json
+            assert stats == self.test_stats
+
+
+def _insert_mock_event(id: str) -> Event:
+    """Inserts a basic event model with a given id."""
+    return Event.insert(
+        id=id,
+        label="Mock",
+        camera="front_door",
+        start_time=datetime.datetime.now().timestamp(),
+        end_time=datetime.datetime.now().timestamp() + 20,
+        top_score=100,
+        false_positive=False,
+        zones=list(),
+        thumbnail="",
+        region=[],
+        box=[],
+        area=0,
+        has_clip=True,
+        has_snapshot=True,
+    ).execute()
+
+
+def _insert_mock_recording(id: str) -> Event:
+    """Inserts a basic recording model with a given id."""
+    return Recordings.insert(
+        id=id,
+        camera="front_door",
+        path=f"/recordings/{id}",
+        start_time=datetime.datetime.now().timestamp() - 50,
+        end_time=datetime.datetime.now().timestamp() - 60,
+        duration=10,
+        motion=True,
+        objects=True,
+    ).execute()