test(flask): backend flask server tests

.github/workflows/vidserver-test-ci.yaml

@@ -0,0 +1,34 @@
+name: Video Processing Server Unit Tests
+
+concurrency:
+  group: ${{ github.ref }}
+  cancel-in-progress: true
+
+on:
+  push:
+    branches:
+      - "develop"
+      - "master"
+      - "main"
+  pull_request:
+    types:
+      - opened
+      - reopened
+      - synchronize
+
+jobs:
+  unit-test:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.8' 
+    - name: Check out the repository
+      uses: actions/checkout@v4
+      with:
+        token: ${{ github.token }}
+    - name: Install dependencies
+      run: pip install -r app/back-end/video-processing/requirements.txt
+    - name: Run tests
+      run: python3 -m unittest discover -s ./tests/back-end

.gitignore

@@ -1,3 +1,6 @@
+# we need this guy
+!**/resources/*.mp4
+
 # don't commit test and temp video/image files
 *.mp4
 *.mkv

app/back-end/video-processing/server.py

@@ -12,9 +12,10 @@ def handle_request():
         file = request.data.decode("utf-8")     # expects the filename, in the form <uid>-<file name>-<epoch time> such as "23-yeehaw-1698360721.mp4"
         if os.path.isfile(f"{input_path}/{file}"):    # check if the file exists
             final = f"{out_path}/{file[:-4]}-processed{file[-4:]}"
-            process = mp.Process(target=vp.process_INTERPOLATE, args=(f"{input_path}/{file}", final, ))  # define a new process pointing to process_INTERPOLATE
-            process.start() # start the process on another thread
-            print(f"Process started on {file}")
+            if not app.testing: # if we're running Flask unit tests, don't run the video processing method
+                process = mp.Process(target=vp.process_INTERPOLATE, args=(f"{input_path}/{file}", final, ))  # define a new process pointing to process_INTERPOLATE
+                process.start() # start the process on another thread
+                print(f"Process started on {file}")
             return "Success: file exists"
         else:
             return "Error: file not found"

tests/.gitignore

@@ -0,0 +1,2 @@
+# we need these files for tests
+!/resources/*.mp4

tests/back-end/README.md

@@ -0,0 +1,22 @@
+# Testing the back-end video processing server
+
+## Requirements
+ - Python 3.8
+    - boto3 v1.28.64
+    - botocore v1.31.64
+    - numpy v1.24.4
+    - opencv-python-headless v4.8.1.78
+    - flask v3.0.0
+    - gunicorn v21.2.0
+
+You can installl these dependencies by navigating to `app/back-end/video-processing` and running:
+```bash
+pip install -r requirements.txt
+```
+
+
+## Running tests
+Navigate to the repository's root directory and run:
+```bash
+python3 -m unittest discover -s ./tests/back-end
+```

tests/back-end/test_server.py

@@ -0,0 +1,62 @@
+import unittest, sys, os
+from flask import Flask
+
+# add server.py's directory to sys path and import it
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../app/back-end/video-processing")))
+
+# set temporary environment variables (will be automatically deleted when the session ends)
+os.environ["PRIVACYPAL_INPUT_VIDEO_DIR"] = f"{os.getcwd()}/tests/resources"
+os.environ["PRIVACYPAL_OUT_VIDEO_DIR"] = f"{os.getcwd()}/tests/resources"
+os.environ["AWS_ACCESS_KEY_ID"] = "some-key-id"
+os.environ["AWS_SECRET_ACCESS_KEY"] = "some-access-key"
+os.environ["AWS_SESSION_TOKEN"] = "some-session-token"
+os.environ["AWS_DEFAULT_REGION"] = "ca-central-1"
+
+from server import app  # finally, import our flask server
+
+class ServerTest(unittest.TestCase):
+    app: Flask
+
+    def setUp(self):
+        self.app = app
+        self.app.testing = True
+        self.client = self.app.test_client()
+
+    def test_process_video_file_not_found(self):
+        route = "/process_video"
+        with self.client as c:
+            response = c.post(route, data="blahblahblah invalid file")
+            self.assertEqual(b"Error: file not found", response.data)
+            self.assertEqual(200, response.status_code)
+
+    def test_process_video_file_found(self):
+        route = "/process_video"
+        with self.client as c:
+            response = c.post(route, data="test.mp4")
+            self.assertEqual(b"Success: file exists", response.data)
+            self.assertEqual(200, response.status_code)
+
+    def test_process_video_method_not_allowed(self):
+        route = "/process_video"
+        with self.client as c:
+            for method in [c.get, c.put, c.delete, c.trace, c.patch]:
+                response = method(route)
+                self.assertEqual(405, response.status_code)
+
+    def test_health(self):
+        route = "/health"
+        with self.client as c:
+            response = c.get(route)
+            json = dict(eval(response.data))  # convert binary string to dictionary
+            self.assertEqual({}, json)
+            self.assertEqual(200, response.status_code)
+
+    def test_health_method_not_allowed(self):
+        route = "/health"
+        with self.client as c:
+            for method in [c.put, c.post, c.delete, c.trace, c.patch]:
+                response = method(route)
+                self.assertEqual(405, response.status_code)
+
+if __name__ == "__main__":
+    unittest.main()

tests/back-end/test_video_processor.py

@@ -1,9 +1,14 @@
-import unittest, sys, os, requests, multiprocessing as mp, cv2 as cv, numpy as np, random
+import unittest, sys, os, cv2 as cv, numpy as np
 
 # add video_processing.py's directory to sys path and import it
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../app/back-end/video-processing")))
 from video_processor import VideoProcessor
 
+os.environ["AWS_ACCESS_KEY_ID"] = "some-key-id"
+os.environ["AWS_SECRET_ACCESS_KEY"] = "some-access-key"
+os.environ["AWS_SESSION_TOKEN"] = "some-session-token"
+os.environ["AWS_DEFAULT_REGION"] = "ca-central-1"
+
 class VideoProcessorTest(unittest.TestCase):
     vp: VideoProcessor
     video_path: str
@@ -14,16 +19,13 @@ class VideoProcessorTest(unittest.TestCase):
 
     def setUp(self):
         self.vp = VideoProcessor()
-        self.video_path = "app/back-end/video-processing/videos/1.mp4"
+        self.video_path = "tests/resources/test.mp4"
         # hard-coded test values i retrieved from ffprobe for this ^ video
         self.fps = 25
         self.width = 1280
         self.height = 720
         self.num_frames = 132
 
-    def test_blur_frame(self):
-        pass
-
     def test_get_frames(self):
         frames = self.vp.get_frames(self.video_path)
         self.assertEqual(len(frames), self.num_frames, "incorrect number of frames in get_frames")
@@ -42,21 +44,13 @@ def test_calc_vector_size(self):
             self.assertEqual(vector, answer, "incorrect calculations in calc_vector")
 
     def test_img_to_bytes(self):
-        file_path = "tests/back-end/test.jpg"
+        file_path = "test.jpg"
         img = np.zeros((64, 64, 3), np.uint8)   # 64x64 black opencv image
         cv.imwrite(file_path, img)      # write out a black 64x64 image to disk
         f = open(file_path, "rb")
         self.assertEqual(self.vp.img_to_bytes(img), f.read())   # compare converted bytes to direct read-from-disk bytes
         f.close()
-
-    def test_get_face(self):
-        pass
-
-    def server_test_unfinished(self):
-        url = "http://127.0.0.1:5000"
-        response = requests.post(url, "test")
-        content = response.content.decode("utf-8")
-        self.assertEqual(content, "Error: file not found")
+        os.remove("test.jpg")
 
 if __name__ == "__main__":
     unittest.main()