Deconflict post-merge

- All task proxies that are `waiting_on_job_prep` should be in the preparing state, so remove redundant check
- Add back in simple trigger-while-paused integration test

Author: MetRonnie

cylc/flow/scheduler.py

@@ -1355,8 +1355,8 @@ def release_tasks_to_run(self) -> bool:
                 pre_prep_tasks.update(self.pool.release_queued_tasks())
 
         if (
-            # Manually triggered tasks will be in preparing state and should
-            # be submitted even if paused (unless stopping).
+            # Manually triggered tasks will be preparing and should
+            # be submitted even if paused (unless workflow is stopping).
             self.is_paused and not self.stop_mode
         ) or (
             # Need to get preparing tasks to submit before auto restart
@@ -1369,8 +1369,7 @@ def release_tasks_to_run(self) -> bool:
             pre_prep_tasks.update({
                 itask
                 for itask in self.pool.get_tasks()
-                if itask.state(TASK_STATUS_PREPARING)
-                and itask.waiting_on_job_prep
+                if itask.waiting_on_job_prep
             })
 
         # Return, if no tasks to submit.

tests/integration/test_force_trigger.py

@@ -42,7 +42,17 @@
 )
 
 
-async def test_trigger_workflow_paused(
+async def test_workflow_paused_simple(
+    one_conf, flow, scheduler, run, complete
+):
+    """It should run triggered tasks even if the workflow is paused."""
+    schd: Scheduler = scheduler(flow(one_conf), paused_start=True)
+    async with run(schd):
+        await run_cmd(force_trigger_tasks(schd, ['1/one'], ['1']))
+        await complete(schd, '1/one', allow_paused=True, timeout=1)
+
+
+async def test_workflow_paused_queues(
     flow: 'Fixture',
     scheduler: 'Fixture',
     start: 'Fixture',
@@ -111,22 +121,17 @@ async def test_trigger_workflow_paused(
 
 
 async def test_trigger_group_whilst_paused(flow, scheduler, run, complete):
-    """Only group start tasks should run whilst the scheduler is paused.
+    """Only group start tasks should run if the group is triggered whilst
+    the scheduler is paused.
 
     Group start tasks have only off-group dependencies.
 
     Others (with in-group dependencies) should run as normal when their
     prerequisites are satisfied once the workflow is resumed.
 
     """
-    id_ = flow(
-        {
-            'scheduling': {
-                'graph': {'R1': 'a => b => c => d'},
-            },
-        }
-    )
-    schd = scheduler(id_)
+    id_ = flow('a => b => c => d')
+    schd = scheduler(id_, paused_start=True)
     async with run(schd):
         # trigger the chain
         await run_cmd(force_trigger_tasks(schd, ['1/a'], []))

tests/integration/test_reload.py

@@ -16,19 +16,20 @@
 
 """Tests for reload behaviour in the scheduler."""
 
-from contextlib import suppress
 
-from cylc.flow import commands
+from cylc.flow import (
+    commands,
+    flags,
+)
+from cylc.flow.cfgspec.glbl_cfg import glbl_cfg
 from cylc.flow.data_store_mgr import TASK_PROXIES
+from cylc.flow.platforms import get_platform
 from cylc.flow.scheduler import Scheduler
 from cylc.flow.task_state import (
-    TASK_STATUS_WAITING,
     TASK_STATUS_PREPARING,
     TASK_STATUS_SUBMITTED,
+    TASK_STATUS_WAITING,
 )
-from cylc.flow.cfgspec.glbl_cfg import glbl_cfg
-from cylc.flow.platforms import get_platform
-from cylc.flow import flags
 
 
 async def test_reload_waits_for_pending_tasks(
@@ -68,8 +69,10 @@ async def test_reload_waits_for_pending_tasks(
 
         # set the task to go through some state changes
         def submit_task_jobs(*a, **k):
-            with suppress(IndexError):
+            try:
                 foo.state_reset(state_seq.pop(0))
+            except IndexError:
+                foo.waiting_on_job_prep = False
             return [foo]
 
         monkeypatch.setattr(