Adapt parentless spawning tests

cylc/flow/task_pool.py

@@ -244,20 +244,6 @@ def _swap_out(self, itask):
             self.active_tasks[itask.point][itask.identity] = itask
             self.active_tasks_changed = True
 
-    def spawn_to_runahead_limit(self):
-        """Spawn the task pool out to the runahead limit in one go.
-
-        Not strictly necessary, it will spawn ahead per main loop iteration,
-        but useful back-compat for tests that expect this prior to
-        https://github.com/cylc/cylc-flow/pull/7237
-
-        """
-        self.compute_runahead()
-        # Arbitrary limit to avoid infinite loop if something goes wrong.
-        for _ in range(10):
-            if not self.release_runahead_tasks():
-                break
-
     def queue_if_ready(self, itask: 'TaskProxy') -> None:
         """Queue itask if it is ready to run.
 
@@ -301,12 +287,6 @@ def load_from_point(self):
                 ntask, _ = self.get_or_spawn_task(point, tdef, flow_nums)
                 if ntask is not None:
                     self.add_to_pool(ntask)
-        # Spawning to the runahead limit immediately is not strictly necessary
-        # as it would occur over several scheduler main loop iterations; we do
-        # it mainly for compatibility with integration tests pre PR #7237.
-        self.spawn_to_runahead_limit()
-        for itask in self.get_tasks():
-            self.queue_if_ready(itask)
 
     def db_add_new_flow_rows(self, itask: TaskProxy) -> None:
         """Add new rows to DB task tables that record flow_nums.

tests/integration/README.md

@@ -78,8 +78,10 @@ Workflows can be run in two ways:
 
 ```
 with start(schd):
-    # starts the Scheduler but does not start the main loop
-    # (always the better option if its possible)
+    # Starts the Scheduler but does not start the main loop
+    #   (always the better option if its possible).
+    # Use the spawn_ahead fixture if you need to examine how
+    #   parentless tasks advance (without iterating the main loop).
     ...
 
 with run(schd):
@@ -92,3 +94,10 @@ These methods both shut down the workflow / clean up after themselves.
 
 It is necessary to shut down workflows correctly to clean up resorces and
 running tasks.
+
+Note at start-up tasks states remain as as initially loaded until evolved by
+the main loop (or otherwise by direct manipulation in tests).
+
+Parentless tasks can spawn freely to the runahead limit, but that happens
+incremently via main loop iterations. Use the spawn_ahead fixture to do the
+same thing without running the main loop in tests.

tests/integration/conftest.py

@@ -777,3 +777,35 @@ def get_submissions():
         }
 
     return get_submissions
+
+
+def _spawn_ahead(task_pool):
+    """ Release runahead tasks and queue any that are ready to run.
+
+    Use to make parentless tasks spawn ahead without the main loop.
+
+    """
+    n_limit = 10  # arbitrary safety net
+    count = 0
+    while True:
+        count += 1
+        task_pool.compute_runahead()
+        if not task_pool.release_runahead_tasks():
+            break
+        for itask in task_pool.get_tasks():
+            task_pool.queue_if_ready(itask)
+        if count > n_limit:
+            raise AssertionError(
+                f"Exceeded {n_limit} iterations"
+                " waiting for task pool to settle"
+            )
+
+
+@pytest.fixture
+def spawn_ahead():
+    return _spawn_ahead
+
+
+@pytest.fixture(scope='module')
+def mod_spawn_ahead():
+    return _spawn_ahead

tests/integration/run_modes/test_mode_overrides.py

@@ -135,7 +135,7 @@ async def test_run_mode_skip_abides_by_held(flow, scheduler, run):
 
 
 async def test_run_mode_override_from_broadcast(
-    flow, scheduler, start, complete, log_filter, capture_live_submissions
+    flow, scheduler, start, capture_live_submissions, spawn_ahead
 ):
     """Test that run_mode modifications only apply to one task.
     """
@@ -155,6 +155,8 @@ async def test_run_mode_override_from_broadcast(
         schd.broadcast_mgr.put_broadcast(
             ['1000'], ['foo'], [{'run mode': 'skip'}])
 
+        spawn_ahead(schd.pool)
+
         foo_1000 = schd.pool.get_task(ISO8601Point('1000'), 'foo')
         foo_1001 = schd.pool.get_task(ISO8601Point('1001'), 'foo')
 

tests/integration/run_modes/test_simulation.py

@@ -90,7 +90,7 @@ def _run_simjob(schd, point, task):
 
 @pytest.fixture(scope='module')
 async def sim_time_check_setup(
-    mod_flow, mod_scheduler, mod_start, mod_one_conf,
+    mod_flow, mod_scheduler, mod_start, mod_spawn_ahead
 ):
     schd = mod_scheduler(mod_flow({
         'scheduler': {'cycle point format': '%Y'},
@@ -129,6 +129,7 @@ async def sim_time_check_setup(
         }
     }))
     async with mod_start(schd):
+        mod_spawn_ahead(schd.pool)
         itasks = schd.pool.get_tasks()
         [schd.task_job_mgr._set_retry_timers(i) for i in itasks]
         yield schd, itasks
@@ -170,7 +171,7 @@ def test_false_if_not_running(
             id='fail-no-submits'),
     )
 )
-def test_fail_once(sim_time_check_setup, itask, point, results, monkeypatch):
+def test_fail_once(sim_time_check_setup, itask, point, results):
     """A task with a fail cycle point only fails
     at that cycle point, and then only on the first submission.
     """

tests/integration/run_modes/test_skip.py

@@ -168,7 +168,7 @@ async def test_skip_mode_outputs(
 
 
 async def test_doesnt_release_held_tasks(
-    one_conf, flow, scheduler, start, log_filter, capture_live_submissions
+    one_conf, flow, scheduler, start, log_filter, spawn_ahead
 ):
     """Point 5 of the proposal
     https://github.com/cylc/cylc-admin/blob/master/docs/proposal-skip-mode.md
@@ -190,6 +190,7 @@ async def test_doesnt_release_held_tasks(
         assert not log_filter(contains='=> succeeded'), msg.format('succeed')
 
         # Release held task and assert that it now skips successfully:
+        spawn_ahead(schd.pool)
         schd.pool.release_held_tasks({TaskTokens('1', 'one')})
         schd.release_tasks_to_run()
 

tests/integration/scripts/test_dump.py

@@ -29,7 +29,7 @@
 DumpOptions = Options(get_option_parser())
 
 
-async def test_dump_tasks(flow, scheduler, start):
+async def test_dump_tasks(flow, scheduler, start, spawn_ahead):
     """It should show n=0 tasks.
 
     See: https://github.com/cylc/cylc-flow/pull/5600
@@ -46,7 +46,8 @@ async def test_dump_tasks(flow, scheduler, start):
     })
     schd = scheduler(id_)
     async with start(schd):
-        # schd.release_tasks_to_run()
+        spawn_ahead(schd.pool)
+        schd.release_tasks_to_run()
         await schd.update_data_structure()
         ret = []
         await dump(

tests/integration/scripts/test_set.py

@@ -70,6 +70,7 @@ async def test_set_parentless_spawning(
     scheduler,
     run,
     complete,
+    spawn_ahead
 ):
     """Ensure that setting outputs does not interfere with parentless spawning.
 
@@ -96,7 +97,8 @@ async def test_set_parentless_spawning(
             ['1'],
         )
 
-        schd.pool.spawn_to_runahead_limit()
+        spawn_ahead(schd.pool)
+
         # the parentless task "a" should be spawned out to the runahead limit
         assert schd.pool.get_task_ids() == {'2/a', '3/a'}
 

tests/integration/test_data_store_mgr.py

@@ -631,7 +631,9 @@ def test_delta_task_xtrigger(xharness):
     assert xtrig_y.satisfied
 
 
-async def test_absolute_graph_edges(flow, scheduler, start):
+async def test_absolute_graph_edges(
+    flow, scheduler, start, spawn_ahead
+):
     """It should add absolute graph edges to the store.
 
     See: https://github.com/cylc/cylc-flow/issues/5845
@@ -651,6 +653,7 @@ async def test_absolute_graph_edges(flow, scheduler, start):
     schd = scheduler(id_)
 
     async with start(schd):
+        spawn_ahead(schd.pool)
         await schd.update_data_structure()
 
         assert {

tests/integration/test_force_trigger.py

@@ -835,6 +835,7 @@ async def test_pre_warm_start_interraction_with_auto_restart(
     scheduler,
     start,
     log_filter,
+    spawn_ahead
 ):
     """Test interaction between warm-start pre-startcp tasks and auto-restart.
 
@@ -858,6 +859,7 @@ async def test_pre_warm_start_interraction_with_auto_restart(
     async with start(schd):
         # trigger a pre-initial task (pre start-cycle point)
         await run_cmd(force_trigger_tasks(schd, ['^/foo'], []))
+        spawn_ahead(schd.pool)
         assert schd.pool.pre_start_tasks_to_trigger
 
         # configure the workflow to auto-retsart

tests/integration/test_id_match.py

@@ -24,7 +24,9 @@
     from cylc.flow.id import TaskTokens
 
 
-async def test_id_match(flow, scheduler, start, caplog):
+async def test_id_match(
+    flow, scheduler, start, caplog, spawn_ahead
+):
     id_ = flow({
         'scheduling': {
             'cycling mode': 'integer',
@@ -69,6 +71,7 @@ def match(*ids: str) -> Tuple[Set[str], Set[str]]:
                 schd, ['1/b2'], ['1'], ['failed'], None
             )
         )
+        spawn_ahead(schd.pool)
 
         # task pool state:
         # * cycle 1

tests/integration/test_optional_outputs.py

@@ -425,6 +425,9 @@ async def test_clock_expiry(
         one = schd.pool.get_task(ISO8601Point('20000101T0000Z'), 'x')
         assert one
 
+        # spawn the second one
+        schd.pool.spawn_next_parentless(one)
+
         # the second task (preparing)
         two = schd.pool.get_task(ISO8601Point('20010101T0000Z'), 'x')
         assert two