Coverage for backend/core/auxiliary/managers/TaskManager.py: 86%

1from common.models.notifications.payloads import NotificationServiceMessageType 

2from common.services import messaging 

3from core.auxiliary.enums.generalEnums import TaskStatus 

4from core.auxiliary.models.Task import Task 

5from core.auxiliary.serializers import TaskSerializer 

6 

7 

8def handle_task_running_event(task_id: int): 

9 """Mark a pending task as running and broadcast the status change.""" 

10 

11 task = Task.objects.get(id=task_id) 

12 

13 # Ignore task status update if task is not pending 

14 # to avoid overwriting status of tasks that have already completed or failed 

15 if task.status != TaskStatus.Pending: 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true

16 return 

17 

18 task.status = TaskStatus.Running 

19 task.save() 

20 

21 # Notify users that the task has started running 

22 messaging.send_flowsheet_notification_message( 

23 task.flowsheet_id, 

24 TaskSerializer(task).data, 

25 NotificationServiceMessageType.TASK_UPDATED 

26 )