detect if None is passed to parse_tasks (#1080)

* detect if None is passed to parse_tasks

* add test to ensure proper exception is raised

* fix

* update nix infrastructure to allow tests to run

Author: WillNilges

.envrc

@@ -1,11 +1,10 @@
-{ pkgs ? import <nixpkgs> {}}:
-
-pkgs.mkShell {
-
-  #buildInputs = [
-  #];
-
-  packages = [ 
-    pkgs.python311
-  ];
-}
+{ pkgs ? import <nixpkgs> {} }:
+(pkgs.buildFHSEnv {
+  name = "pipzone";
+  targetPkgs = pkgs: (with pkgs; [
+    python311
+    python311Packages.pip
+    python311Packages.virtualenv
+  ]);
+  runScript = "zsh";
+}).env

shell.nix

@@ -2128,3 +2128,27 @@ def active(self):
         response = c.get("/api/v1/uisp-import/status/")
 
         self.assertEqual(500, response.status_code)
+
+    @patch("meshapi.views.uisp_import.app.control.inspect")
+    def test_view_uisp_on_demand_import_status_none(self, mock_celery_app):
+        class BrokenMockInspect(MagicMock):
+            def scheduled(self):
+                return None
+
+            def reserved(self):
+                return None
+
+            def active(self):
+                return None
+
+        mock_celery_app.side_effect = BrokenMockInspect()
+
+        # Create a client
+        self.admin_user = User.objects.create_superuser(
+            username="admin", password="admin_password", email="admin@example.com"
+        )
+        c = Client()
+        c.login(username="admin", password="admin_password")
+        response = c.get("/api/v1/uisp-import/status/")
+
+        self.assertEqual(500, response.status_code)

src/meshapi/tests/test_uisp_import.py

@@ -46,6 +46,8 @@ def view_uisp_on_demand_import_status(request: Request) -> Response:
 
         def parse_tasks(tasks_by_worker: dict[str, list], status: str) -> list[dict[str, str]]:
             parsed_tasks = []
+            if tasks_by_worker is None:
+                raise Exception("Got None for tasks_by_worker. Is something wrong with celery-worker?")
             for _, tasks in tasks_by_worker.items():
                 for t in tasks:
                     if "run_uisp_on_demand_import" in t.get("name"):