Add unit tests to Quasar

.github/workflows/unit-tests.yml

@@ -0,0 +1,29 @@
+name: Quasar Unit Tests
+
+on:
+  push:
+    branches: [dev]
+  pull_request:
+    branches: [dev]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v3
+
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.11'
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r printer/requirements.txt
+
+    - name: Run tests
+      run: |
+        python printer/test_server.py

.gitignore

@@ -13,4 +13,6 @@ config.json
 
 **/.DS_Store
 
+venv/
+
 tmp/

printer/requirements.txt

@@ -1,4 +1,6 @@
-fastapi==0.84.0
+fastapi==0.115.12
 uvicorn==0.18.3
 py-grpc-prometheus==0.7.0
 python-multipart==0.0.9
+httpx==0.28.1
+requests==2.32.3

printer/server.py

@@ -1,6 +1,4 @@
 import argparse
-import base64
-import json
 import logging
 import os
 import pathlib
@@ -9,7 +7,7 @@
 import time
 import uuid
 
-from fastapi import FastAPI, File, Form, Request, HTTPException, UploadFile
+from fastapi import FastAPI, File, Form, HTTPException, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import PlainTextResponse
 import prometheus_client
@@ -63,9 +61,6 @@ def get_args() -> argparse.Namespace:
 
 args = get_args()
 
-# only the right printer works right now, so we default to it
-PRINTER_NAME = os.environ.get("RIGHT_PRINTER_NAME")
-
 
 def maybe_reopen_ssh_tunnel():
     """
@@ -97,6 +92,9 @@ def send_file_to_printer(
         # to speciy page ranges, we can do:
         # `-o page-ranges=<whatever user sent>` OR `-P <whatever user sent>`
         maybe_page_range = f"-o page-ranges={page_range}"
+
+    # only the right printer works right now, so we default to it
+    PRINTER_NAME = os.environ.get("RIGHT_PRINTER_NAME")
     command = f"lp -n {num_copies} {maybe_page_range} -o sides={sides} -o media=na_letter_8.5x11in -d {PRINTER_NAME} {file_path}"
     metrics_handler.print_jobs_recieved.inc()
     if args.development:

printer/test_server.py

@@ -0,0 +1,80 @@
+import io
+import os
+import unittest
+from unittest import mock
+
+from fastapi.testclient import TestClient
+from server import app
+
+
+class TestFastAPI(unittest.TestCase):
+    def setUp(self):
+        os.environ["RIGHT_PRINTER_NAME"] = "HP_P2015_DN"
+        self.client = TestClient(app)
+
+    def test_health_check(self):
+        response = self.client.get("/healthcheck/printer")
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.text, '"printer is up!"')
+
+    @mock.patch("server.uuid.uuid4", return_value="test-id")
+    @mock.patch("server.subprocess.Popen")
+    @mock.patch("builtins.open", new_callable=mock.mock_open)
+    @mock.patch("pathlib.Path.unlink")
+    def test_print_endpoint(self, mock_pathlib_unlink, mock_open_func, mock_popen, _):
+        test_file = io.BytesIO(b"dummy file content")
+        response = self.client.post(
+            "/print",
+            files={"file": ("test.txt", test_file, "text/plain")},
+            data={"copies": "1", "sides": "one-sided"},
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.text, '"worked!"')
+
+        mock_open_func.assert_called_once_with("/tmp/test-id", "wb")
+
+        mock_open_func().write.assert_called_once()
+        self.assertEqual(
+            mock_open_func().write.call_args_list[0], mock.call(b"dummy file content")
+        )
+
+        mock_popen.assert_called_once()
+
+        self.assertEqual(
+            mock_popen.call_args_list[0],
+            mock.call(
+                "lp -n 1  -o sides=one-sided -o media=na_letter_8.5x11in -d HP_P2015_DN /tmp/test-id",
+                shell=True,
+            ),
+        )
+
+        mock_pathlib_unlink.assert_called_once()
+
+    @mock.patch("server.subprocess.Popen")
+    @mock.patch("builtins.open", side_effect=FileNotFoundError("sorry!"))
+    @mock.patch("pathlib.Path.unlink")
+    def test_print_endpoint(self, mock_pathlib_unlink, _, mock_popen):
+        test_file = io.BytesIO(b"dummy file content")
+        response = self.client.post(
+            "/print",
+            files={"file": ("test.txt", test_file, "text/plain")},
+            data={"copies": "1", "sides": "one-sided"},
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response.json(),
+            {
+                "status_code": 500,
+                "detail": "printing failed, check logs",
+                "headers": None,
+            },
+        )
+
+        mock_popen.assert_not_called()
+        mock_pathlib_unlink.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()