Ensure that old mode_settings objects are cleared when a

simulated job finishes. Test that reloading with a new
simulation config leads to a different outcome.

Author: wxtim

cylc/flow/simulation.py

@@ -90,6 +90,7 @@ def __init__(
 
         self.timeout = started_time + self.simulated_run_length
 
+
 def configure_sim_modes(taskdefs, sim_mode):
     """Adjust task defs for simulation and dummy mode.
 
@@ -219,6 +220,7 @@ def sim_time_check(
     """
     now = time()
     sim_task_state_changed: bool = False
+
     for itask in itasks:
         if (
             itask.state.status != TASK_STATUS_RUNNING
@@ -228,8 +230,6 @@ def sim_time_check(
         ):
             continue
 
-
-
         if itask.mode_settings is None:
             itask.mode_settings = ModeSettings(itask, broadcast_mgr, db_mgr)
 
@@ -251,6 +251,10 @@ def sim_time_check(
                 message_queue.put(
                     TaskMsg(job_d, now_str, 'DEBUG', msg)
                 )
+
+            # We've finished this psuedojob, so delete all the mode settings,
+            # we'll recalculate them for the next "submission"
+            itask.mode_settings = None
             sim_task_state_changed = True
     return sim_task_state_changed
 

tests/integration/test_simulation.py

@@ -13,7 +13,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 pathlib import Path
 import pytest
 from pytest import param
 from queue import Queue
@@ -239,7 +239,7 @@ def test_task_sped_up(sim_time_check_setup, monkeytime):
 
 
 async def test_simulation_mode_settings_restart(
-    monkeytime, flow, scheduler, run, start
+    monkeytime, flow, scheduler, start
 ):
     """Check that simulation mode settings are correctly restored
     upon restart.
@@ -258,8 +258,11 @@ async def test_simulation_mode_settings_restart(
         'runtime': {
             'one': {
                 'execution time limit': 'PT1M',
+                'execution retry delays': 'P0Y',
                 'simulation': {
-                    'speedup factor': 1
+                    'speedup factor': 1,
+                    'fail cycle points': 'all',
+                    'fail try 1 only': True,
                 }
             },
         }
@@ -292,8 +295,9 @@ async def test_simulation_mode_settings_restart(
         monkeytime(12)
         itask.state.status = 'running'
 
-        # Check that we haven't got started time back
+        # Check that we haven't got started time & mode settings back:
         assert itask.summary['started_time'] is None
+        assert itask.mode_settings is None
 
         # Set the start time in the database to 0 to make the
         # test simpler:
@@ -308,9 +312,98 @@ async def test_simulation_mode_settings_restart(
             schd.workflow_db_mgr
         ) is False
 
+        # Check that the itask.mode_settings is now re-created
+        assert itask.mode_settings.__dict__ == {
+            'simulated_run_length': 60.0,
+            'sim_task_fails': False,
+            'timeout': 60.0
+        }
+
         # Set the current time > timeout
         monkeytime(61)
         assert sim_time_check(
             msg_q, [itask], schd.task_events_mgr.broadcast_mgr,
             schd.workflow_db_mgr
         ) is True
+
+        assert itask.mode_settings is None
+
+        schd.task_events_mgr.broadcast_mgr.put_broadcast(
+            ['1066'], ['one'], [{
+                'execution time limit': 'PT1S'}])
+
+        assert itask.mode_settings is None
+
+        schd.task_job_mgr._simulation_submit_task_jobs(
+            [itask], schd.workflow)
+
+        assert itask.submit_num == 2
+        assert itask.mode_settings.__dict__ == {
+            'simulated_run_length': 1.0,
+            'sim_task_fails': False,
+            'timeout': 62.0
+        }
+
+
+async def test_simulation_mode_settings_reload(
+    monkeytime, flow, scheduler, run, start
+):
+    """Check that simulation mode settings are changed for future
+    pseudo jobs on reload.
+
+    We run a task twice
+    """
+    id_ = flow({
+        'scheduler': {'cycle point format': '%Y'},
+        'scheduling': {
+            'initial cycle point': '1066',
+            'graph': {
+                'R1': 'one'
+            }
+        },
+        'runtime': {
+            'one': {
+                'execution time limit': 'PT1M',
+                'execution retry delays': 'P0Y',
+                'simulation': {
+                    'fail cycle points': 'all',
+                    'fail try 1 only': False,
+                }
+            },
+        }
+    })
+    schd = scheduler(id_)
+    async with start(schd) as log:
+        itask = schd.pool.get_tasks()[0]
+        itask.state.status = 'running'
+        itask.state.is_queued = False
+        itask.summary['started_time'] = 0
+
+        # Submit & fail:
+        schd.task_job_mgr._simulation_submit_task_jobs(
+            [itask], schd.workflow)
+        monkeytime(61)
+        itask.mode_settings.timeout = 60
+        assert sim_time_check(
+            schd.message_queue, [itask], schd.task_events_mgr.broadcast_mgr,
+            schd.workflow_db_mgr
+        ) is True
+        schd.process_queued_task_messages()
+
+        # Modify the config in place, reload workflow:
+        conf_file = Path(schd.workflow_run_dir) / 'flow.cylc'
+        conf_file.write_text(conf_file.read_text().replace('= all', '='))
+        await schd.command_reload_workflow()
+
+        # The proxy representing 1066/one will be a different object:
+        itask = schd.pool.get_tasks()[0]
+
+        # Mode settings have been cleared (it's a new task anyway):
+        assert itask.mode_settings is None
+
+        # Submit again.
+        schd.task_job_mgr._simulation_submit_task_jobs(
+            [itask], schd.workflow)
+
+        # Mode settings have changed as per the new config:
+        assert itask.mode_settings.sim_task_fails is False