Made execution time limit unsettable by reload or broadcast.

Formerly a suppress statement was supressing cases when we
wanted to set itask.summary['execution time limit'] = None.

Added tests.

cylc/flow/task_job_mgr.py

@@ -36,7 +36,7 @@
 )
 from shutil import rmtree
 from time import time
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Union, Optional
 
 from cylc.flow import LOG
 from cylc.flow.job_runner_mgr import JobPollContext
@@ -1262,10 +1262,10 @@ def _prep_submit_task_job_impl(self, workflow, itask, rtconfig):
             itask.submit_num] = itask.platform['name']
 
         itask.summary['job_runner_name'] = itask.platform['job runner']
-        with suppress(TypeError):
-            itask.summary[self.KEY_EXECUTE_TIME_LIMIT] = float(
-                rtconfig['execution time limit']
-            )
+
+        # None is an allowed non-float number for Execution time limit.
+        itask.summary[self.KEY_EXECUTE_TIME_LIMIT] = self.get_execution_time_limit(
+            rtconfig['execution time limit'])
 
         # Location of job file, etc
         self._create_job_log_path(workflow, itask)
@@ -1281,6 +1281,35 @@ def _prep_submit_task_job_impl(self, workflow, itask, rtconfig):
             job_d=job_d
         )
 
+    @staticmethod
+    def get_execution_time_limit(
+        config_execution_time_limit: Any
+    ) -> Union[None, float]:
+        """Get execution time limit from config and process it.
+
+        We are making no assumptions about what can be passed in,
+        but passing silently and returning None if the value is
+        not a type convertable to a float:
+
+        Examples:
+            >>> this = TaskJobManager.get_execution_time_limit
+            >>> this(None)
+            >>> this("54")
+            54.0
+            >>> this({})
+
+        N.b.
+            This function does not save you from a value error:
+            >>> from pytest import raises
+            >>> with raises(ValueError):
+            ...     this("🇳🇿")
+        """
+        execution_time_limit = None
+        if config_execution_time_limit is not None:
+            with suppress(TypeError):
+                execution_time_limit = float(config_execution_time_limit)
+        return execution_time_limit
+
     def get_job_conf(
         self,
         workflow,

tests/integration/test_task_job_mgr.py

@@ -14,6 +14,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from contextlib import suppress
 import logging
 from typing import Any as Fixture
 
@@ -128,3 +129,61 @@ async def test__run_job_cmd_logs_platform_lookup_fail(
         warning = caplog.records[-1]
         assert warning.levelname == 'ERROR'
         assert 'Unable to run command jobs-poll' in warning.msg
+
+
+async def test__prep_submit_task_job_impl_handles_execution_time_limit(
+    flow: Fixture,
+    scheduler: Fixture,
+    start: Fixture,
+    validate: Fixture,
+    pytestconfig: Fixture
+):
+    """Ensure that emptying the execution time limit unsets it.
+
+    Previously unsetting the etl by either broadcast or reload
+    would not unset a previous etl.
+
+    See https://github.com/cylc/cylc-flow/issues/5891
+    """
+    id_ = flow({
+        "scheduling": {
+            "cycling mode": "integer",
+            "graph": {"R1": "a"}
+        },
+        "runtime": {
+            "root": {},
+            "a": {
+                "script": "sleep 10",
+                "execution time limit": 'PT5S'
+            }
+        }
+    })
+    # Debugging only:
+    if pytestconfig.option.verbose > 2:
+        validate(id_)
+
+    # Run in live mode - function not called in sim mode.
+    schd = scheduler(id_, run_mode='live')
+    async with start(schd):
+        task_a = schd.pool.get_tasks()[0]
+        # We're not interested in the job file stuff, just
+        # in the summary state.
+        with suppress(FileExistsError):
+            schd.task_job_mgr._prep_submit_task_job_impl(
+                schd.workflow, task_a, task_a.tdef.rtconfig)
+            assert task_a.summary['execution_time_limit'] == 5.0
+
+            # If we delete the etl it gets deleted in the summary:
+            task_a.tdef.rtconfig['execution time limit'] = None
+            schd.task_job_mgr._prep_submit_task_job_impl(
+                schd.workflow, task_a, task_a.tdef.rtconfig)
+            assert not task_a.summary.get('execution_time_limit', '')
+
+            # put everything back and test broadcast too.
+            task_a.tdef.rtconfig['execution time limit'] = 5.0
+            task_a.summary['execution_time_limit'] = 5.0
+            schd.broadcast_mgr.broadcasts = {
+                '1': {'a': {'execution time limit': None}}}
+            schd.task_job_mgr._prep_submit_task_job_impl(
+                schd.workflow, task_a, task_a.tdef.rtconfig)
+            assert not task_a.summary.get('execution_time_limit', '')