Fix TypeError when tokens lacks a job number

Author: MetRonnie

cylc/flow/id.py

@@ -366,6 +366,20 @@ def is_null(self) -> bool:
             self[key] for key in self._REGULAR_KEYS
         )
 
+    @property
+    def submit_num(self) -> int | None:
+        """The job submit number as an integer, or None if not set.
+
+        Examples:
+            >>> Tokens('//c/t/01').submit_num
+            1
+            >>> Tokens('//c/t').submit_num is None
+            True
+        """
+        if self['job'] is None:
+            return None
+        return int(self['job'])
+
     @overload
     def duplicate(
         self,

cylc/flow/network/resolvers.py

@@ -32,7 +32,6 @@
     Optional,
     Tuple,
     TYPE_CHECKING,
-    Union,
     cast,
 )
 from uuid import uuid4
@@ -69,9 +68,9 @@
 
 class TaskMsg(NamedTuple):
     """Tuple for Scheduler.message_queue"""
-    job_id: Union[Tokens, str]
+    job_id: Tokens
     event_time: str
-    severity: Union[str, int]
+    severity: str | int
     message: str
 
 
@@ -839,7 +838,12 @@ def put_messages(
         """
         for severity, message in messages:
             self.schd.message_queue.put(
-                TaskMsg(task_job, event_time, severity, message)
+                TaskMsg(
+                    Tokens(task_job, relative=True),
+                    event_time,
+                    severity,
+                    message,
+                )
             )
         return (True, f'Messages queued: {len(messages)}')
 

cylc/flow/scheduler.py

@@ -964,7 +964,7 @@ def process_queued_task_messages(self) -> None:
         """Process incoming task messages for each task proxy.
 
         """
-        messages: 'Dict[str, List[Tuple[Optional[int], TaskMsg]]]' = {}
+        messages: dict[str, list[TaskMsg]] = {}
 
         # Retrieve queued messages
         while self.message_queue.qsize():
@@ -973,29 +973,23 @@ def process_queued_task_messages(self) -> None:
             except Empty:
                 break
             self.message_queue.task_done()
-            tokens = Tokens(task_msg.job_id, relative=True)
             # task ID (job stripped)
-            task_id = tokens.duplicate(job=None).relative_id
-            messages.setdefault(task_id, [])
-            # job may be None (e.g. simulation mode)
-            job = int(tokens['job']) if tokens['job'] else None
-            messages[task_id].append(
-                (job, task_msg)
-            )
+            task_id = task_msg.job_id.duplicate(job=None).relative_id
+            messages.setdefault(task_id, []).append(task_msg)
 
         unprocessed_messages: List[TaskMsg] = []
         # Poll tasks for which messages caused a backward state change.
         to_poll_tasks: List[TaskProxy] = []
         for task_id, message_items in messages.items():
             itask = self.pool._get_task_by_id(task_id)
             if itask is None:
-                unprocessed_messages.extend(tm for _, tm in message_items)
+                unprocessed_messages.extend(message_items)
                 continue
             should_poll = False
-            for submit_num, tm in message_items:
+            for tm in message_items:
                 if self.task_events_mgr.process_message(
                     itask, tm.severity, tm.message, tm.event_time,
-                    self.task_events_mgr.FLAG_RECEIVED, submit_num
+                    self.task_events_mgr.FLAG_RECEIVED, tm.job_id.submit_num
                 ):
                     should_poll = True
             if should_poll:
@@ -1008,9 +1002,7 @@ def process_queued_task_messages(self) -> None:
         # proxy can be removed, but the orphaned job still sends messages.
         warn = ""
         for tm in unprocessed_messages:
-            job_tokens = self.tokens.duplicate(
-                Tokens(tm.job_id, relative=True)
-            )
+            job_tokens = self.tokens.duplicate(tm.job_id)
             tdef = self.config.get_taskdef(job_tokens['task'])
             if not self.task_events_mgr.process_job_message(
                 job_tokens, tdef, tm.message, tm.event_time

cylc/flow/task_events_mgr.py

@@ -647,7 +647,7 @@ def process_events(self, schd: 'Scheduler') -> None:
                     SubProcContext(
                         (
                             (id_key.handler, id_key.event),
-                            int(id_key.tokens['job']),
+                            id_key.tokens.submit_num,
                         ),
                         timer.ctx.cmd,
                         env=os.environ,
@@ -994,7 +994,7 @@ def process_job_message(
             job_tokens.workflow,
             tdef,
             get_point(job_tokens['cycle']),
-            submit_num=int(job_tokens['job']),
+            submit_num=job_tokens.submit_num,
             data_mode=True,
             transient=True,
         )
@@ -1286,7 +1286,7 @@ def _job_logs_retrieval_callback(self, proc_ctx, schd) -> None:
                 log_ctx = SubProcContext(
                     (
                         (id_key.handler, id_key.event),
-                        int(id_key.tokens['job'])
+                        id_key.tokens.submit_num
                     ),
                     None,
                 )

cylc/flow/task_proxy.py

@@ -222,7 +222,7 @@ def __init__(
         flow_nums: Optional['FlowNums'] = None,
         status: str = TASK_STATUS_WAITING,
         is_held: bool = False,
-        submit_num: int = 0,
+        submit_num: int | None = 0,
         is_late: bool = False,
         is_manual_submit: bool = False,
         flow_wait: bool = False,

tests/integration/test_optional_outputs.py

@@ -32,7 +32,7 @@
 )
 from cylc.flow.cycling.integer import IntegerPoint
 from cylc.flow.cycling.iso8601 import ISO8601Point
-from cylc.flow.id import TaskTokens
+from cylc.flow.id import TaskTokens, Tokens
 from cylc.flow.network.resolvers import TaskMsg
 from cylc.flow.task_events_mgr import (
     TaskEventsManager,
@@ -208,7 +208,7 @@ async def test_expire_orthogonality(flow, scheduler, start):
         # tell the scheduler that the task *submit-failed*
         schd.message_queue.put(
             TaskMsg(
-                '1/a/01',
+                Tokens('//1/a/01'),
                 '2000-01-01T00:00:00+00',
                 'INFO',
                 TaskEventsManager.EVENT_SUBMIT_FAILED
@@ -222,7 +222,7 @@ async def test_expire_orthogonality(flow, scheduler, start):
         # tell the scheduler that the task *failed*
         schd.message_queue.put(
             TaskMsg(
-                '1/a/01',
+                Tokens('//1/a/01'),
                 '2000-01-01T00:00:00+00',
                 'INFO',
                 TaskEventsManager.EVENT_FAILED,
@@ -236,7 +236,7 @@ async def test_expire_orthogonality(flow, scheduler, start):
         # tell the scheduler that the task *expired*
         schd.message_queue.put(
             TaskMsg(
-                '1/a/01',
+                Tokens('//1/a/01'),
                 '2000-01-01T00:00:00+00',
                 'INFO',
                 TaskEventsManager.EVENT_EXPIRED,

tests/integration/test_task_events_mgr.py

@@ -275,20 +275,23 @@ async def test__process_message_failed_with_retry(
         assert 'failed/OOK' in failed_record[1]
 
 
-async def test__unhandled_message(one: Scheduler, start, log_filter):
+@pytest.mark.parametrize('id_', ['1/no_such_task/01', '1/no_job'])
+async def test__unhandled_message(id_, one: Scheduler, start, log_filter):
     """It should log unhandled messages."""
 
     async with start(one):
         one.message_queue.put(
-            TaskMsg("1/no_such_task/01", "time", 'INFO', "the quick brown")
+            TaskMsg(
+                Tokens(id_, relative=True), "time", 'INFO', "the quick brown"
+            )
         )
         one.process_queued_task_messages()
 
         _, warning_msg = log_filter(level=logging.WARNING)[-1]
         assert (
             'Undeliverable task messages received and ignored:' in warning_msg
         )
-        assert '1/no_such_task/01: INFO - "the quick brown"' in warning_msg
+        assert f'{id_}: INFO - "the quick brown"' in warning_msg
 
 
 @pytest.mark.parametrize('template', TEMPLATES)