Refactor UploadBuilder and ProcessService for improved error handling…

backend/core/process/builders/upload_builder.py

@@ -28,7 +28,8 @@ def build(self):
         if not check.get("success"):
             raise ValueError(check.get("message"))
 
-        product.status = ProductStatus.PROCESSING
+        created_product = product.data
+        created_product.status = ProductStatus.PROCESSING
         product.save()
 
-        return product.data
+        return created_product

backend/core/process/service.py

@@ -44,8 +44,11 @@ def submit(self):
         )
 
         self.process.orchestration_process_id = orch_process.get("id")
-        self.process.used_config = json.loads(orch_process.get("used_config", "{}"))
         self.process.path = orch_process.get("path_str")
+        orch_used_config = orch_process.get("used_config", {})
+        if isinstance(orch_used_config, str):
+            orch_used_config = json.loads(orch_used_config or "{}")
+        self.process.used_config = orch_used_config
         self.process.save()
 
     def stop(self):

backend/core/test/test_process_logs.py

@@ -0,0 +1,28 @@
+import tempfile
+from pathlib import Path
+
+from core.utils import get_logs
+from django.test import SimpleTestCase, override_settings
+
+
+class GetLogsTestCase(SimpleTestCase):
+    def test_returns_empty_logs_when_paths_are_missing(self):
+        logs = get_logs(None, None)
+
+        self.assertEqual("", logs["pipeline"]["content"])
+        self.assertEqual("", logs["slurm"]["content"])
+
+    def test_reads_logs_from_product_dir_when_process_dir_is_missing(self):
+        with tempfile.TemporaryDirectory() as upload_root:
+            product_relpath = "validation_results/sample-product"
+            pipeline_log = (
+                Path(upload_root) / product_relpath / "process_info" / "pipeline.log"
+            )
+            pipeline_log.parent.mkdir(parents=True, exist_ok=True)
+            pipeline_log.write_text("pipeline output", encoding="utf-8")
+
+            with override_settings(UPLOAD_DIR=upload_root):
+                logs = get_logs(None, product_relpath)
+
+            self.assertEqual("pipeline output", logs["pipeline"]["content"])
+            self.assertEqual("", logs["slurm"]["content"])

backend/core/test/test_process_service.py

@@ -0,0 +1,116 @@
+from types import SimpleNamespace
+from unittest import mock
+
+from core.models import Pipeline, Process, Product, ProductStatus, ProductType
+from core.process.builders.upload_builder import UploadBuilder
+from core.process.service import ProcessService
+from django.contrib.auth.models import User
+from django.test import TestCase
+
+
+class UploadBuilderTestCase(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(
+            "john", "john@snow.com", "you_know_nothing"
+        )
+        self.product_type = ProductType.objects.create(
+            name="validation_results",
+            display_name="Validation Results",
+        )
+        self.pipeline = Pipeline.objects.create(
+            name="combine_redshift_dedup",
+            display_name="Combine Redshift Dedup",
+            version="1.0.0",
+            output_product_type=self.product_type,
+        )
+
+    def test_build_sets_upload_status_to_processing(self):
+        serializer = SimpleNamespace(
+            initial_data={
+                "pipeline": self.pipeline.pk,
+                "display_name": "Combined Catalog",
+                "description": "Generated by process",
+            }
+        )
+
+        product = UploadBuilder(serializer, self.user).build()
+
+        self.assertEqual(ProductStatus.PROCESSING, product.status)
+        self.assertIsNotNone(product.path)
+
+
+class ProcessServiceTestCase(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(
+            "john", "john@snow.com", "you_know_nothing"
+        )
+        self.product_type = ProductType.objects.create(
+            name="redshift_catalog",
+            display_name="Redshift Catalog",
+        )
+        self.pipeline = Pipeline.objects.create(
+            name="combine_redshift_dedup",
+            display_name="Combine Redshift Dedup",
+            version="1.0.0",
+            output_product_type=self.product_type,
+        )
+        self.upload = Product.objects.create(
+            product_type=self.product_type,
+            user=self.user,
+            display_name="Combined Catalog",
+            internal_name="1_combined_catalog",
+            path="redshift_catalog/1_combined_catalog",
+            status=ProductStatus.PROCESSING,
+        )
+        self.process = Process.objects.create(
+            display_name="Combined Catalog",
+            pipeline=self.pipeline,
+            upload=self.upload,
+            user=self.user,
+        )
+
+    @mock.patch("core.process.service.InputsBuilder.build", return_value=[])
+    @mock.patch("core.process.service.BasePipelineHandler.get_handler")
+    @mock.patch("core.process.service.Maestro")
+    def test_submit_accepts_dict_used_config(
+        self,
+        maestro_cls,
+        get_handler,
+        _build_inputs,
+    ):
+        class FakeHandler:
+            def __init__(self, request, process):
+                self.request = request
+                self.process = process
+
+            def build_config(self):
+                return {"param": {"debug": True}}
+
+        get_handler.return_value = FakeHandler
+
+        maestro = maestro_cls.return_value
+        maestro.start.return_value = {
+            "id": 123,
+            "path_str": "combine_redshift_dedup/00000123",
+            "used_config": {
+                "param": {"debug": True},
+                "inputs": {"specz": []},
+                "output_dir": self.upload.path,
+            },
+        }
+
+        request = SimpleNamespace(data={})
+
+        ProcessService(request, self.process).submit()
+        self.process.refresh_from_db()
+
+        self.assertEqual(123, self.process.orchestration_process_id)
+        self.assertEqual("combine_redshift_dedup/00000123", self.process.path)
+        self.assertEqual(
+            {
+                "param": {"debug": True},
+                "inputs": {"specz": []},
+                "output_dir": self.upload.path,
+            },
+            self.process.used_config,
+        )

backend/core/utils.py

@@ -36,20 +36,29 @@ def get_logs(process_dir, product_dir):
 
     logs = get_lognames()
 
-    process_dir = pathlib.Path(settings.PROCESSING_DIR, process_dir)
-    product_dir = pathlib.Path(settings.UPLOAD_DIR, product_dir)
+    search_roots = []
+
+    if process_dir:
+        search_roots.append(
+            ("process_dir", pathlib.Path(settings.PROCESSING_DIR, process_dir))
+        )
+
+    if product_dir:
+        search_roots.append(
+            ("product_dir", pathlib.Path(settings.UPLOAD_DIR, product_dir))
+        )
 
     for key, data in logs.items():
         log = []
         lognames = data.get("filename", [])
         logger.debug(f"Searching for log files: {key}")
 
         for fname in lognames:
-            log = glob.glob(f"{process_dir}/**/{fname}", recursive=True)
-            logger.debug(f"process_dir: {process_dir}/**/{fname} -> {log}")
-            if not log:
-                log = glob.glob(f"{product_dir}/**/{fname}", recursive=True)
-                logger.debug(f"product_dir: {product_dir}/**/{fname} -> {log}")
+            for root_name, root_path in search_roots:
+                log = glob.glob(f"{root_path}/**/{fname}", recursive=True)
+                logger.debug(f"{root_name}: {root_path}/**/{fname} -> {log}")
+                if log:
+                    break
 
             if log:
                 break

backend/core/views/process.py

@@ -2,7 +2,7 @@
 
 import requests
 from core.maestro import Maestro
-from core.models import Process
+from core.models import Process, ProductStatus
 from core.process.builders.upload_builder import UploadBuilder
 from core.process.service import ProcessService
 from core.serializers import ProcessSerializer
@@ -84,7 +84,19 @@ def create(self, request):
         try:
             ProcessService(request, process).submit()
         except Exception as err:
-            LOGGER.error(err)
+            process.status = "Failed"
+            process.comment = str(err)
+            process.save(update_fields=["status", "comment"])
+            process.upload.status = ProductStatus.FAILED
+            process.upload.save()
+            LOGGER.exception(
+                "Process submission failed: process_id=%s upload_id=%s orchestration_process_id=%s process_path=%s upload_path=%s",
+                process.pk,
+                process.upload_id,
+                process.orchestration_process_id,
+                process.path,
+                process.upload.path,
+            )
             return Response(
                 {"error": str(err)},
                 status=status.HTTP_500_INTERNAL_SERVER_ERROR,

backend/pzserver/test_settings.py

@@ -127,8 +127,9 @@
 STATIC_URL = "/django_static/"
 STATIC_ROOT = os.path.join(BASE_DIR, "django_static")
 
+UPLOAD_DIR = os.getenv("UPLOAD_DIR", os.path.join(BASE_DIR, "archive/data/"))
 MEDIA_URL = os.path.join(BASE_DIR, "archive/data/")
-MEDIA_ROOT = os.path.join(BASE_DIR, "archive/data/")
+MEDIA_ROOT = os.getenv("MEDIA_ROOT", UPLOAD_DIR)
 PRODUCT_DOWNLOAD_COMPRESSION_LEVEL = int(
     os.getenv("PRODUCT_DOWNLOAD_COMPRESSION_LEVEL", "6")
 )
@@ -151,6 +152,8 @@
     os.getenv("PRODUCT_DOWNLOAD_PREPARE_POLL_INTERVAL_SECONDS", "0.5")
 )
 
+PROCESSING_DIR = os.getenv("PROCESSING_DIR", os.path.join(BASE_DIR, "processes"))
+
 
 # Default primary key field type
 # https://docs.djangoproject.com/en/4.0/ref/settings/#default-auto-field