Text Parsing Subprocess Command (#125)

epicgdog · evanugarte · web-flow · commit 8db7636d2f62 · 2025-06-28T18:40:43.000-07:00
* retrieve id from print_job

* from loggin.warning to logging.info

* retrieve id from print_job

* from loggin.warning to logging.info

* retrieve id from print_job

* added args to popen

* more fixes

* added unit tests for text-parsing

* added changes for server.py

* added unit tests for returncode 1 and 0

* added better logging

* fixed test typos

* tests are fixed

* lint

* rebase

* return early in if

* better exception, comment for fun

---------

Co-authored-by: evan &lt;evanuxd@gmail.com&gt;
diff --git a/printer/server.py b/printer/server.py
@@ -26,7 +26,8 @@
     allow_headers=["*"],
 )
 logging.basicConfig(
-    format="%(asctime)s.%(msecs)03dZ %(processName)s %(threadName)s %(levelname)s:%(name)s:%(message)s",
+    # in mondo we trust
+    format="%(asctime)s.%(msecs)03dZ %(levelname)s:%(name)s:%(message)s",
     datefmt="%Y-%m-%dT%H:%M:%S",
     level=logging.INFO,
 )
@@ -60,9 +61,9 @@ def get_args() -> argparse.Namespace:
         "--dont-delete-pdfs",
         action="store_true",
         default=False,
-        help="specify if server should delete pdfs after printing"
+        help="specify if server should delete pdfs after printing",
     )
-    
+
     return parser.parse_args()
 
 
@@ -105,27 +106,62 @@ def send_file_to_printer(
     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:
-        logging.warning(f"server is in development mode, command would've been `{command}`")
-    else:
-        print_job = subprocess.Popen(
-            command,
-            shell=True,
+        logging.warning(
+            f"server is in development mode, command would've been `{command}`"
+        )
+        return None
+
+    print_job = subprocess.Popen(
+        command,
+        shell=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    print_job.wait()
+
+    if print_job.returncode != 0:
+        logging.error(
+            f"command returned code {print_job.returncode} stderr: {print_job.stderr.read()} stdout: {print_job.stdout.read()}"
+        )
+        return None
+    try:
+        print_id = print_job.stdout.read().strip().split(" ")[3]
+        logging.info(f"extracted print id is {print_id}")
+        return print_id
+    except Exception:
+        logging.exception(
+            f"failed to extract print id from stdout: {print_job.stdout.read()}"
         )
-        print_job.wait()
+        # need to find a better value to return when the command exited
+        # with code 0 but the output could not be parsed for a job id.
+        return ''
+
+
+def maybe_delete_pdf(file_path):
+    if args.dont_delete_pdfs:
+        logging.info(
+            f"--dont-delete-pdfs is set, skipping deletion of file {file_path}"
+        )
+        return
+    pathlib.Path(file_path).unlink()
 
 
 @app.get("/healthcheck/printer")
 def api():
     metrics_handler.last_health_check_request.set(int(time.time()))
     return "printer is up!"
 
+
 @app.get("/metrics", response_class=PlainTextResponse)
 def metrics():
     return prometheus_client.generate_latest()
 
 
 @app.post("/print")
-async def read_item(file: UploadFile = File(...), copies: str = Form(...), sides: str = Form(...)):
+async def read_item(
+    file: UploadFile = File(...), copies: str = Form(...), sides: str = Form(...)
+):
     """
     incoming request to print looks like
     {
@@ -140,16 +176,17 @@ async def read_item(file: UploadFile = File(...), copies: str = Form(...), sides
         file_path = str(base / file_id)
         with open(file_path, "wb") as f:
             f.write(await file.read())
-        send_file_to_printer(
+        print_id = send_file_to_printer(
             str(file_path),
             copies,
             sides=sides,
         )
-        if args.dont_delete_pdfs:
-          logging.info(f'--dont-delete-pdfs is set, skipping deletion of file {file_path}')
-          return "worked!"
-        pathlib.Path(file_path).unlink()
-        return "worked!"
+
+        maybe_delete_pdf(file_path)
+
+        if not args.development and print_id is None:
+            raise Exception("unable to extract print id from print request")
+        return {"print_id": print_id}
     except Exception:
         logging.exception("printing failed!")
         return HTTPException(
diff --git a/printer/test_server.py b/printer/test_server.py
@@ -1,9 +1,11 @@
 import importlib
 import io
 import os
+import subprocess
 import unittest
 from unittest import mock
 
+
 from fastapi.testclient import TestClient
 
 import server
@@ -34,14 +36,29 @@ def test_health_check(self):
     def test_print_endpoint(self, mock_pathlib_unlink, mock_open_func, mock_popen, _):
         client = self.load_server_with_args()
         test_file = io.BytesIO(b"dummy file content")
+
+        mock_popen_result = mock.MagicMock()
+        mock_popen_result.returncode = 0
+        mock_popen_result.stdout.read.return_value = (
+            "request id is HP_LaserJet_p2015dn_Right-53 (1 file(s))"
+        )
+
+        mock_popen.return_value = mock_popen_result
+
         response = 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!"')
+        json_response = response.json()
+        self.assertEqual(
+            json_response,
+            {
+                "print_id": "HP_LaserJet_p2015dn_Right-53",
+            },
+        )
 
         mock_open_func.assert_called_once_with("/tmp/test-id", "wb")
 
@@ -57,6 +74,9 @@ def test_print_endpoint(self, mock_pathlib_unlink, mock_open_func, mock_popen, _
             mock.call(
                 "lp -n 1  -o sides=one-sided -o media=na_letter_8.5x11in -d HP_P2015_DN /tmp/test-id",
                 shell=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
             ),
         )
 
@@ -69,6 +89,14 @@ def test_print_endpoint(self, mock_pathlib_unlink, mock_open_func, mock_popen, _
     def test_print_endpoint_dont_delete_pdf(
         self, mock_pathlib_unlink, mock_open_func, mock_popen, _
     ):
+
+        mock_popen_result = mock.MagicMock()
+        mock_popen_result.returncode = 0
+        mock_popen_result.stdout.read.return_value = (
+            "request id is HP_LaserJet_p2015dn_Right-53 (1 file(s))"
+        )
+
+        mock_popen.return_value = mock_popen_result
         client = self.load_server_with_args(["--dont-delete-pdfs"])
         test_file = io.BytesIO(b"dummy file content")
         response = client.post(
@@ -78,7 +106,13 @@ def test_print_endpoint_dont_delete_pdf(
         )
 
         self.assertEqual(response.status_code, 200)
-        self.assertEqual(response.text, '"worked!"')
+        json_response = response.json()
+        self.assertEqual(
+            json_response,
+            {
+                "print_id": "HP_LaserJet_p2015dn_Right-53",
+            },
+        )
 
         mock_open_func.assert_called_once_with("/tmp/test-id", "wb")
 
@@ -94,6 +128,9 @@ def test_print_endpoint_dont_delete_pdf(
             mock.call(
                 "lp -n 1  -o sides=one-sided -o media=na_letter_8.5x11in -d HP_P2015_DN /tmp/test-id",
                 shell=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
             ),
         )
 
@@ -102,7 +139,7 @@ def test_print_endpoint_dont_delete_pdf(
     @mock.patch("server.subprocess.Popen")
     @mock.patch("builtins.open", side_effect=FileNotFoundError("sorry!"))
     @mock.patch("pathlib.Path.unlink")
-    def test_print_endpoint_error(self, mock_pathlib_unlink, _, mock_popen):
+    def test_print_endpoint_file_not_found(self, mock_pathlib_unlink, _, mock_popen):
         client = self.load_server_with_args()
         test_file = io.BytesIO(b"dummy file content")
         response = client.post(
@@ -124,6 +161,107 @@ def test_print_endpoint_error(self, mock_pathlib_unlink, _, mock_popen):
         mock_popen.assert_not_called()
         mock_pathlib_unlink.assert_not_called()
 
+    @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_nonzero_returncode(
+        self, mock_pathlib_unlink, mock_open_func, mock_popen, _
+    ):
+        client = self.load_server_with_args()
+        test_file = io.BytesIO(b"dummy file content")
+
+        mock_popen_result = mock.MagicMock()
+        mock_popen_result.returncode = 1
+        mock_popen.return_value = mock_popen_result
+
+        response = client.post(
+            "/print",
+            files={"file": ("test.txt", test_file, "text/plain")},
+            data={"copies": "1", "sides": "dark-side"},
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response.json(),
+            {
+                "status_code": 500,
+                "detail": "printing failed, check logs",
+                "headers": None,
+            },
+        )
+
+        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=dark-side -o media=na_letter_8.5x11in -d HP_P2015_DN /tmp/test-id",
+                shell=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            ),
+        )
+
+        mock_pathlib_unlink.assert_called_once()
+
+    @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_junk_print_id(self, mock_pathlib_unlink, mock_open_func, mock_popen, _):
+        client = self.load_server_with_args()
+        test_file = io.BytesIO(b"dummy file content")
+
+        mock_popen_result = mock.MagicMock()
+        mock_popen_result.returncode = 0
+        mock_popen_result.stdout.read.return_value = "junk output"
+        mock_popen.return_value = mock_popen_result
+
+        response = 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(),
+            {
+                "print_id": "",
+            },
+        )
+
+        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,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            ),
+        )
+
+        mock_pathlib_unlink.assert_called_once()
+
 
 if __name__ == "__main__":
     unittest.main()
