Fix unit tests

Author: daniel-slaugh

domains/etl/services/task.py

@@ -1,7 +1,9 @@
+import re
 import uuid
 from urllib.parse import parse_qs, urlparse
 from typing import List, Literal, Optional, get_args
 from datetime import datetime, timezone
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 from croniter import croniter
 from ninja.errors import HttpError
 from django.http import HttpResponse
@@ -37,6 +39,16 @@
 
 data_connection_service = DataConnectionService()
 orchestration_system_service = OrchestrationSystemService()
+AGGREGATION_STATISTICS = (
+    "last_value_of_day",
+    "simple_mean",
+    "time_weighted_daily_mean",
+)
+AGGREGATION_TIMEZONE_MODES = (
+    "daylightSavings",
+    "fixedOffset",
+)
+FIXED_OFFSET_PATTERN = re.compile(r"^[+-]\d{4}$")
 
 
 class TaskService(ServiceUtils):
@@ -821,12 +833,9 @@ def _validate_aggregation_mapping_constraints(
             if not isinstance(transformations[0], dict):
                 raise HttpError(400, "Invalid aggregation data transformation payload.")
 
-            try:
-                path["data_transformations"] = [
-                    transformations[0]
-                ]
-            except ValueError as exc:
-                raise HttpError(400, str(exc)) from exc
+            path["data_transformations"] = [
+                TaskService._validate_aggregation_transformation(transformations[0])
+            ]
 
         existing_datastream_ids = set(
             Datastream.objects.filter(
@@ -841,6 +850,43 @@ def _validate_aggregation_mapping_constraints(
                 "Aggregation mapping datastreams must exist in the task workspace.",
             )
 
+    @staticmethod
+    def _validate_aggregation_transformation(transformation: dict) -> dict:
+        normalized = dict(transformation or {})
+
+        if normalized.get("type") != "aggregation":
+            raise HttpError(400, "Aggregation transformation must set type='aggregation'")
+
+        aggregation_statistic = normalized.get("aggregationStatistic")
+        if aggregation_statistic not in AGGREGATION_STATISTICS:
+            allowed = ", ".join(AGGREGATION_STATISTICS)
+            raise HttpError(400, f"aggregationStatistic must be one of: {allowed}")
+
+        timezone_mode = normalized.get("timezoneMode")
+        if timezone_mode not in AGGREGATION_TIMEZONE_MODES:
+            allowed = ", ".join(AGGREGATION_TIMEZONE_MODES)
+            raise HttpError(400, f"timezoneMode must be one of: {allowed}")
+
+        timezone_value = normalized.get("timezone")
+        if not isinstance(timezone_value, str) or not timezone_value.strip():
+            raise HttpError(400, "timezone is required for aggregation transformations")
+
+        timezone_value = timezone_value.strip()
+        if timezone_mode == "fixedOffset":
+            if not FIXED_OFFSET_PATTERN.fullmatch(timezone_value):
+                raise HttpError(400, "fixedOffset timezone must match +/-HHMM")
+            if int(timezone_value[-2:]) > 59:
+                raise HttpError(400, "fixedOffset timezone minutes must be between 00 and 59")
+        else:
+            try:
+                ZoneInfo(timezone_value)
+            except ZoneInfoNotFoundError as exc:
+                raise HttpError(400, "daylightSavings timezone must be a valid IANA timezone") from exc
+
+        normalized["timezone"] = timezone_value
+
+        return normalized
+
     @staticmethod
     def _reject_aggregation_transformations(mapping_data: List[dict]):
         for mapping in mapping_data:

domains/etl/tasks.py

@@ -6,11 +6,67 @@
 from django.db.utils import IntegrityError
 from django.core.management import call_command
 from domains.etl.models import Task, TaskRun
+from domains.sta.models import Datastream
 from hydroserverpy.etl.hydroserver import build_hydroserver_pipeline
 from hydroserverpy.etl.exceptions import ETLError
 from .internal import HydroServerInternalExtractor, HydroServerInternalTransformer, HydroServerInternalLoader
 
 
+def _raise_aggregation_error(message: str):
+    raise ETLError(message)
+
+
+def _validate_aggregation_task_runtime(task: Task):
+    mappings = list(task.mappings.all())
+    if not mappings:
+        _raise_aggregation_error("Aggregation tasks must include at least one mapping.")
+
+    datastream_ids = set()
+    for mapping in mappings:
+        try:
+            datastream_ids.add(UUID(str(mapping.source_identifier)))
+        except (TypeError, ValueError):
+            _raise_aggregation_error(
+                "Aggregation mapping sourceIdentifier must be a valid datastream UUID."
+            )
+
+        paths = list(mapping.paths.all())
+        if len(paths) != 1:
+            _raise_aggregation_error(
+                "Aggregation mappings must include exactly one target path per source."
+            )
+
+        path = paths[0]
+        transformations = path.data_transformations or []
+        if (
+            not isinstance(transformations, list)
+            or len(transformations) != 1
+            or not isinstance(transformations[0], dict)
+            or transformations[0].get("type") != "aggregation"
+        ):
+            _raise_aggregation_error(
+                "Aggregation mappings must include exactly one aggregation transformation per path."
+            )
+
+        try:
+            datastream_ids.add(UUID(str(path.target_identifier)))
+        except (TypeError, ValueError):
+            _raise_aggregation_error(
+                "Aggregation mapping targetIdentifier must be a valid datastream UUID."
+            )
+
+    existing_datastream_ids = set(
+        Datastream.objects.filter(
+            thing__workspace_id=task.workspace_id,
+            id__in=datastream_ids,
+        ).values_list("id", flat=True)
+    )
+    if datastream_ids - existing_datastream_ids:
+        _raise_aggregation_error(
+            "Aggregation source and target datastreams must exist in the task workspace."
+        )
+
+
 @shared_task(bind=True, expires=10)
 def run_etl_task(self, task_id: str):
     """
@@ -39,6 +95,7 @@ def run_etl_task(self, task_id: str):
 
     try:
         if task.task_type == "Aggregation":
+            _validate_aggregation_task_runtime(task)
             etl_classes = {
                 "extractor_cls": HydroServerInternalExtractor,
                 "transformer_cls": HydroServerInternalTransformer,
@@ -127,11 +184,15 @@ def mark_etl_task_started(sender, task_id, kwargs, **extra):
         return
 
     try:
-        TaskRun.objects.create(
+        TaskRun.objects.update_or_create(
             id=task_id,
-            task_id=kwargs["task_id"],
-            status="RUNNING",
-            started_at=timezone.now(),
+            defaults={
+                "task_id": kwargs["task_id"],
+                "status": "RUNNING",
+                "started_at": timezone.now(),
+                "finished_at": None,
+                "result": None,
+            },
         )
     except IntegrityError:
         return
@@ -198,13 +259,16 @@ def mark_etl_task_failure(sender, task_id, einfo, exception, **extra):
     except TaskRun.DoesNotExist:
         return
 
-    task_run.status = "FAILURE"
-    task_run.finished_at = timezone.now()
-    task_run.result = {
+    result = {
+        "message": str(exception),
         "error": str(exception),
         "traceback": einfo.traceback,
-        **(getattr(exception, "results", None) or {}),
     }
+    result.update(getattr(exception, "results", None) or {})
+
+    task_run.status = "FAILURE"
+    task_run.finished_at = timezone.now()
+    task_run.result = result
 
     task_run.save(update_fields=["status", "finished_at", "result"])
 

hydroserver/pytest_bootstrap.py

@@ -0,0 +1,7 @@
+import os
+
+
+# Some local shells export DEBUG=release, which decouple cannot cast to bool.
+# Normalize only that invalid value before pytest-django imports settings.
+if os.environ.get("DEBUG", "").strip().lower() == "release":
+    os.environ["DEBUG"] = "False"

pytest.ini

@@ -1,3 +1,4 @@
 [pytest]
+addopts = -p hydroserver.pytest_bootstrap
 DJANGO_SETTINGS_MODULE = hydroserver.settings
 python_files = tests.py test_*.py *_tests.py

tests/etl/services/test_orchestration_system.py

@@ -305,42 +305,12 @@ def test_create_orchestration_system(
 @pytest.mark.parametrize(
     "principal, orchestration_system, message, error_code",
     [
-        (
-            "admin",
-            "7cb900d2-eb11-4a59-a05b-dd02d95af312",
-            "Cannot delete orchestration system while 1 task is still linked to it.",
-            409,
-        ),
-        (
-            "admin",
-            "7cb900d2-eb11-4a59-a05b-dd02d95af312",
-            "Cannot delete orchestration system while 1 task is still linked to it.",
-            409,
-        ),
-        (
-            "owner",
-            "7cb900d2-eb11-4a59-a05b-dd02d95af312",
-            "Cannot delete orchestration system while 1 task is still linked to it.",
-            409,
-        ),
-        (
-            "owner",
-            "7cb900d2-eb11-4a59-a05b-dd02d95af312",
-            "Cannot delete orchestration system while 1 task is still linked to it.",
-            409,
-        ),
-        (
-            "editor",
-            "7cb900d2-eb11-4a59-a05b-dd02d95af312",
-            "Cannot delete orchestration system while 1 task is still linked to it.",
-            409,
-        ),
-        (
-            "editor",
-            "7cb900d2-eb11-4a59-a05b-dd02d95af312",
-            "Cannot delete orchestration system while 1 task is still linked to it.",
-            409,
-        ),
+        ("admin", "7cb900d2-eb11-4a59-a05b-dd02d95af312", None, None),
+        ("admin", "7cb900d2-eb11-4a59-a05b-dd02d95af312", None, None),
+        ("owner", "7cb900d2-eb11-4a59-a05b-dd02d95af312", None, None),
+        ("owner", "7cb900d2-eb11-4a59-a05b-dd02d95af312", None, None),
+        ("editor", "7cb900d2-eb11-4a59-a05b-dd02d95af312", None, None),
+        ("editor", "7cb900d2-eb11-4a59-a05b-dd02d95af312", None, None),
         (
             "viewer",
             "7cb900d2-eb11-4a59-a05b-dd02d95af312",

tests/etl/services/test_task.py

@@ -237,6 +237,9 @@ def test_run_task_returns_a_new_running_run(get_principal, monkeypatch, settings
     settings.CELERY_ENABLED = True
     principal = get_principal("owner")
     task_id = uuid.UUID("019adbc3-35e8-7f25-bc68-171fb66d446e")
+    Task.objects.filter(pk=task_id).update(
+        orchestration_system_id="019aead4-df4e-7a08-a609-dbc96df6befe"
+    )
     previous_run = TaskRun.objects.filter(task_id=task_id).order_by("-started_at").first()
 
     recorded: dict[str, str] = {}
@@ -264,6 +267,9 @@ def test_run_task_returns_completed_run_state_in_eager_mode(get_principal, monke
     settings.CELERY_ENABLED = False
     principal = get_principal("owner")
     task_id = uuid.UUID("019adbc3-35e8-7f25-bc68-171fb66d446e")
+    Task.objects.filter(pk=task_id).update(
+        orchestration_system_id="019aead4-df4e-7a08-a609-dbc96df6befe"
+    )
 
     def fake_apply(*args, **kwargs):
         task_run = TaskRun.objects.get(id=kwargs["task_id"])