Fix rescheduling (1.13) (#2141)

Co-authored-by: JP Bruins Slot <jpbruinsslot@gmail.com>
Co-authored-by: Jan Klopper <janklopper+underdark@gmail.com>

Author: 3 people

mula/scheduler/models/queue.py

@@ -20,16 +20,16 @@ class PrioritizedItem(BaseModel):
 
     model_config = ConfigDict(from_attributes=True)
 
-    id: uuid.UUID = Field(default_factory=uuid.uuid4)
+    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4)
 
     scheduler_id: Optional[str] = None
 
     # A unique generated identifier for the object contained in data
     hash: Optional[str] = Field(None, max_length=32)
 
-    priority: Optional[int]
+    priority: Optional[int] = 0
 
-    data: Dict
+    data: Dict = Field(default_factory=dict)
 
     created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
 

mula/scheduler/models/tasks.py

@@ -109,7 +109,7 @@ class NormalizerTask(BaseModel):
 
     type: ClassVar[str] = "normalizer"
 
-    id: uuid.UUID = Field(default_factory=lambda: uuid.uuid4())
+    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4)
     normalizer: Normalizer
     raw_data: RawData
 
@@ -128,7 +128,7 @@ class BoefjeTask(BaseModel):
 
     type: ClassVar[str] = "boefje"
 
-    id: uuid.UUID = Field(default_factory=lambda: uuid.uuid4())
+    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4)
     boefje: Boefje
     input_ooi: Optional[str]
     organization: str

mula/scheduler/server/server.py

@@ -442,15 +442,23 @@ def push_queue(self, queue_id: str, item: models.PrioritizedItem) -> Any:
             )
 
         try:
-            p_item = models.PrioritizedItem(**item.model_dump())
+            # Load default values
+            p_item = models.PrioritizedItem()
+
+            # Set default values
             if p_item.scheduler_id is None:
                 p_item.scheduler_id = s.scheduler_id
 
+            p_item.priority = item.priority
+
             if s.queue.item_type == models.BoefjeTask:
-                p_item.data = models.BoefjeTask(**p_item.data).dict()
+                p_item.data = models.BoefjeTask(**item.data).dict()
             elif s.queue.item_type == models.NormalizerTask:
-                p_item.data = models.NormalizerTask(**p_item.data).dict()
+                p_item.data = models.NormalizerTask(**item.data).dict()
+            else:
+                p_item.data = item.data
         except Exception as exc:
+            self.logger.exception(exc)
             raise fastapi.HTTPException(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail=str(exc),
@@ -475,7 +483,7 @@ def push_queue(self, queue_id: str, item: models.PrioritizedItem) -> Any:
                 detail=str(exc_not_allowed),
             ) from exc_not_allowed
 
-        return models.PrioritizedItem(**p_item.model_dump())
+        return p_item
 
     def run(self) -> None:
         uvicorn.run(

mula/tests/integration/test_api.py

@@ -132,15 +132,19 @@ def test_push_queue(self):
 
         item = create_p_item(self.organisation.id, 1)
 
+        # Remove id, scheduler should create id for us
+        delattr(item, "id")
+
         response = self.client.post(f"/queues/{self.scheduler.scheduler_id}/push", data=item.model_dump_json())
         self.assertEqual(response.status_code, 201)
         self.assertEqual(1, self.scheduler.queue.qsize())
+        self.assertIsNotNone(response.json().get("id"))
 
     def test_push_incorrect_item_type(self):
         response = self.client.post(
             f"/queues/{self.scheduler.scheduler_id}/push", json={"priority": 0, "item": "not a task"}
         )
-        self.assertEqual(response.status_code, 422)
+        self.assertEqual(response.status_code, 400)
 
     def test_push_queue_full(self):
         # Set maxsize of the queue to 1
@@ -227,7 +231,7 @@ def test_push_replace_allowed(self):
         self.assertEqual(1, self.scheduler.queue.qsize())
 
         # Check if the item on the queue is the replaced item
-        self.assertEqual(response.json().get("id"), str(self.scheduler.queue.peek(0).id))
+        self.assertEqual(response.json().get("data").get("id"), str(self.scheduler.queue.peek(0).data.get("id")))
 
     def test_push_updates_not_allowed(self):
         # Set queue to no allow updates
@@ -282,7 +286,7 @@ def test_push_updates_allowed(self):
         self.assertEqual(1, self.scheduler.queue.qsize())
 
         # Check if the item on the queue is the updated item
-        self.assertEqual(response.json().get("id"), str(self.scheduler.queue.peek(0).id))
+        self.assertEqual(response.json().get("data").get("id"), str(self.scheduler.queue.peek(0).data.get("id")))
         self.assertEqual(response.json().get("data").get("name"), "updated-name")
 
     def test_push_priority_updates_not_allowed(self):
@@ -339,7 +343,7 @@ def test_update_priority_higher(self):
         self.assertEqual(1, self.scheduler.queue.qsize())
 
         # Check if the item on the queue is the updated item
-        self.assertEqual(response.json().get("id"), str(self.scheduler.queue.peek(0).id))
+        self.assertEqual(response.json().get("data").get("id"), str(self.scheduler.queue.peek(0).data.get("id")))
 
     def test_update_priority_lower(self):
         """When updating the priority of the initial item on the priority queue
@@ -370,7 +374,7 @@ def test_update_priority_lower(self):
         self.assertEqual(1, self.scheduler.queue.qsize())
 
         # Check if the item on the queue is the updated item
-        self.assertEqual(response.json().get("id"), str(self.scheduler.queue.peek(0).id))
+        self.assertEqual(response.json().get("data").get("id"), str(self.scheduler.queue.peek(0).data.get("id")))
 
     def test_pop_queue(self):
         # Add one task to the queue
@@ -395,12 +399,14 @@ def test_pop_queue_filters(self):
             f"/queues/{self.scheduler.scheduler_id}/push",
             data=first_item.model_dump_json(),
         )
+        first_item_id = response.json().get("id")
         self.assertEqual(response.status_code, 201)
         self.assertEqual(1, self.scheduler.queue.qsize())
 
         # Add second item to the queue
         second_item = create_p_item(self.organisation.id, 2, data=functions.TestModel(id="456", name="test"))
         response = self.client.post(f"/queues/{self.scheduler.scheduler_id}/push", data=second_item.model_dump_json())
+        second_item_id = response.json().get("id")
         self.assertEqual(response.status_code, 201)
         self.assertEqual(2, self.scheduler.queue.qsize())
 
@@ -409,7 +415,7 @@ def test_pop_queue_filters(self):
             f"/queues/{self.scheduler.scheduler_id}/pop", json=[{"field": "name", "operator": "eq", "value": "test"}]
         )
         self.assertEqual(200, response.status_code)
-        self.assertEqual(str(first_item.id), response.json().get("id"))
+        self.assertEqual(first_item_id, response.json().get("id"))
         self.assertEqual(1, self.scheduler.queue.qsize())
 
         # Should not return any items
@@ -425,7 +431,7 @@ def test_pop_queue_filters(self):
             f"/queues/{self.scheduler.scheduler_id}/pop", json=[{"field": "name", "operator": "eq", "value": "test"}]
         )
         self.assertEqual(200, response.status_code)
-        self.assertEqual(str(second_item.id), response.json().get("id"))
+        self.assertEqual(second_item_id, response.json().get("id"))
         self.assertEqual(0, self.scheduler.queue.qsize())
 
     def test_pop_empty(self):
@@ -449,10 +455,11 @@ def setUp(self):
             ),
         )
         response = self.client.post(f"/queues/{self.scheduler.scheduler_id}/push", data=first_item.model_dump_json())
+        initial_item_id = response.json().get("id")
         self.assertEqual(response.status_code, 201)
         self.assertEqual(1, self.scheduler.queue.qsize())
 
-        self.first_item_api = self.client.get(f"/tasks/{first_item.id}").json()
+        self.first_item_api = self.client.get(f"/tasks/{initial_item_id}").json()
 
         # Add second item to the queue
         second_item = create_p_item(
@@ -461,10 +468,11 @@ def setUp(self):
             data=functions.TestModel(id="456", name="test"),
         )
         response = self.client.post(f"/queues/{self.scheduler.scheduler_id}/push", data=second_item.model_dump_json())
+        second_item_id = response.json().get("id")
         self.assertEqual(response.status_code, 201)
         self.assertEqual(2, self.scheduler.queue.qsize())
 
-        self.second_item_api = self.client.get(f"/tasks/{second_item.id}").json()
+        self.second_item_api = self.client.get(f"/tasks/{second_item_id}").json()
 
     def test_create_task(self):
         item = create_p_item(self.organisation.id, 1)

mula/tests/mocks/queue.py

@@ -1,8 +1,7 @@
-from scheduler import queues
-
-from tests.utils import functions
+from scheduler import models, queues
+from scheduler.utils import dict_utils
 
 
 class MockPriorityQueue(queues.PriorityQueue):
-    def create_hash(self, item: functions.TestModel):
-        return item.id.hex
+    def create_hash(self, p_item: models.PrioritizedItem) -> str:
+        return dict_utils.deep_get(p_item.model_dump(), ["data", "id"])

rocky/katalogus/views/mixins.py

@@ -1,6 +1,5 @@
 from logging import getLogger
 from typing import List, Optional, Union
-from uuid import uuid4
 
 from account.mixins import OrganizationView
 from django.contrib import messages
@@ -82,12 +81,11 @@ class NormalizerMixin(OctopoesView):
 
     def run_normalizer(self, normalizer: KATalogusNormalizer, raw_data: RawData) -> None:
         normalizer_task = NormalizerTask(
-            id=uuid4(),
             normalizer=Normalizer.parse_obj(normalizer.dict()),
             raw_data=raw_data,
         )
 
-        task = QueuePrioritizedItem(id=normalizer_task.id, priority=1, data=normalizer_task)
+        task = QueuePrioritizedItem(priority=1, data=normalizer_task)
 
         schedule_task(self.request, self.organization.code, task)
 
@@ -100,13 +98,12 @@ class BoefjeMixin(OctopoesView):
 
     def run_boefje(self, katalogus_boefje: KATalogusBoefje, ooi: Optional[OOI]) -> None:
         boefje_task = BoefjeTask(
-            id=uuid4().hex,
             boefje=Boefje.parse_obj(katalogus_boefje.dict()),
             input_ooi=ooi.reference if ooi else None,
             organization=self.organization.code,
         )
 
-        task = QueuePrioritizedItem(id=boefje_task.id, priority=1, data=boefje_task)
+        task = QueuePrioritizedItem(priority=1, data=boefje_task)
         schedule_task(self.request, self.organization.code, task)
 
     def run_boefje_for_oois(

rocky/rocky/locale/django.pot

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PACKAGE VERSION\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2023-12-11 09:21+0000\n"
+"POT-Creation-Date: 2023-12-14 15:21+0000\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <LL@li.org>\n"
@@ -3337,6 +3337,10 @@ msgid ""
 "refresh of the page may be needed to show the results."
 msgstr ""
 
+#: tools/view_helpers.py rocky/scheduler.py
+msgid "Task not found."
+msgstr ""
+
 #: rocky/messaging.py
 msgid ""
 "You have trusted this member with a clearance level of L{}. This member "
@@ -3360,10 +3364,6 @@ msgstr ""
 msgid "Task already queued."
 msgstr ""
 
-#: rocky/scheduler.py
-msgid "Task not found."
-msgstr ""
-
 #: rocky/settings.py
 msgid "Blue light"
 msgstr ""

rocky/rocky/scheduler.py

@@ -65,7 +65,7 @@ class NormalizerMeta(BaseModel):
 class NormalizerTask(BaseModel):
     """NormalizerTask represent data needed for a Normalizer to run."""
 
-    id: uuid.UUID
+    id: Optional[uuid.UUID]
     normalizer: Normalizer
     raw_data: RawData
     type: str = "normalizer"
@@ -74,7 +74,7 @@ class NormalizerTask(BaseModel):
 class BoefjeTask(BaseModel):
     """BoefjeTask represent data needed for a Boefje to run."""
 
-    id: uuid.UUID
+    id: Optional[uuid.UUID] = None
     boefje: Boefje
     input_ooi: Optional[str]
     organization: str
@@ -87,7 +87,7 @@ class QueuePrioritizedItem(BaseModel):
     representation.
     """
 
-    id: uuid.UUID
+    id: Optional[uuid.UUID]
     priority: int
     hash: Optional[str]
     data: Union[BoefjeTask, NormalizerTask]
@@ -105,7 +105,7 @@ class TaskStatus(Enum):
 
 
 class Task(BaseModel):
-    id: uuid.UUID
+    id: Optional[uuid.UUID] = None
     scheduler_id: str
     type: str
     p_item: QueuePrioritizedItem
@@ -227,7 +227,7 @@ def get_lazy_task_list(
     def get_task_details(self, organization_code: str, task_id: str) -> Optional[Task]:
         res = self.session.get(f"{self._base_uri}/tasks/{task_id}")
         res.raise_for_status()
-        task_details = Task.model_validate_json(res.content)
+        task_details = Task.parse_raw(res.content)
 
         if task_details.type == "normalizer":
             organization = task_details.p_item.data.raw_data.boefje_meta.organization

rocky/tools/view_helpers.py

@@ -178,10 +178,31 @@ def schedule_task(request: HttpRequest, organization_code: str, p_item: QueuePri
         )
 
 
+# FIXME: Tasks should be (re)created with supplied data, not by fetching prior
+# task info from the scheduler. Task data should be available from the context
+# from which the task is created.
 def reschedule_task(request: HttpRequest, organization_code: str, task_id: str) -> None:
     try:
         task = client.get_task_details(organization_code, task_id)
     except SchedulerError as error:
         messages.error(request, error.message)
-    else:
-        schedule_task(request, organization_code, task.p_item)
+        return
+
+    if not task:
+        messages.error(request, _("Task not found."))
+        return
+
+    # Remove id from task data, this should be created by the scheduler
+    new_task = task.p_item.data
+    delattr(new_task, "id")
+
+    try:
+        new_p_item = QueuePrioritizedItem(
+            data=new_task,
+            priority=1,
+        )
+
+        schedule_task(request, organization_code, new_p_item)
+    except SchedulerError as error:
+        messages.error(request, error.message)
+        return