diff --git a/compass/ocean/tests/global_convergence/cosine_bell/__init__.py b/compass/ocean/tests/global_convergence/cosine_bell/__init__.py
index 8ac28cb55f..e02da42acd 100644
--- a/compass/ocean/tests/global_convergence/cosine_bell/__init__.py
+++ b/compass/ocean/tests/global_convergence/cosine_bell/__init__.py
@@ -72,25 +72,6 @@ def configure(self):
 
         self.update_cores()
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for resolution in self.resolutions:
-            if self.icosahedral:
-                mesh_name = f'Icos{resolution}'
-            else:
-                mesh_name = f'QU{resolution}'
-            ntasks = config.getint('cosine_bell', f'{mesh_name}_ntasks')
-            min_tasks = config.getint('cosine_bell', f'{mesh_name}_min_tasks')
-            step = self.steps[f'{mesh_name}_forward']
-            step.ntasks = ntasks
-            step.min_tasks = min_tasks
-
-        # run the step
-        super().run()
-
     def update_cores(self):
         """ Update the number of cores and min_tasks for each forward step """
 
diff --git a/compass/ocean/tests/global_convergence/cosine_bell/forward.py b/compass/ocean/tests/global_convergence/cosine_bell/forward.py
index 40ed0e154b..dd9b781ba7 100644
--- a/compass/ocean/tests/global_convergence/cosine_bell/forward.py
+++ b/compass/ocean/tests/global_convergence/cosine_bell/forward.py
@@ -13,6 +13,9 @@ class Forward(Step):
     ----------
     resolution : int
         The resolution of the (uniform) mesh in km
+
+    mesh_name : str
+        The name of the mesh
     """
 
     def __init__(self, test_case, resolution, mesh_name):
@@ -35,6 +38,7 @@ def __init__(self, test_case, resolution, mesh_name):
                          subdir=f'{mesh_name}/forward')
 
         self.resolution = resolution
+        self.mesh_name = mesh_name
 
         # make sure output is double precision
         self.add_streams_file('compass.ocean.streams', 'streams.output')
@@ -61,6 +65,14 @@ def setup(self):
         """
         dt = self.get_dt()
         self.add_namelist_options({'config_dt': dt})
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -92,3 +104,9 @@ def get_dt(self):
         dt = time.strftime('%H:%M:%S', time.gmtime(dt))
 
         return dt
+
+    def _get_resources(self):
+        mesh_name = self.mesh_name
+        config = self.config
+        self.ntasks = config.getint('cosine_bell', f'{mesh_name}_ntasks')
+        self.min_tasks = config.getint('cosine_bell', f'{mesh_name}_min_tasks')
diff --git a/compass/ocean/tests/global_ocean/forward.py b/compass/ocean/tests/global_ocean/forward.py
index 91313bbba8..c8916a48ec 100644
--- a/compass/ocean/tests/global_ocean/forward.py
+++ b/compass/ocean/tests/global_ocean/forward.py
@@ -25,14 +25,9 @@ class ForwardStep(Step):
     time_integrator : {'split_explicit', 'RK4'}
         The time integrator to use for the forward run
 
-    ntasks_from_config : bool
-        Whether to get ``ntasks`` from the config file
-
-    min_tasks_from_config : bool
-        Whether to get ``min_tasks`` from the config file
-
-    threads_from_config : bool
-        Whether to get ``threads`` from the config file
+    resources_from_config : bool
+        Whether to get ``ntasks``, ``min_tasks`` and ``openmp_threads`` from
+        config options
     """
     def __init__(self, test_case, mesh, init, time_integrator, name='forward',
                  subdir=None, ntasks=None, min_tasks=None,
@@ -81,9 +76,10 @@ def __init__(self, test_case, mesh, init, time_integrator, name='forward',
                          ntasks=ntasks, min_tasks=min_tasks,
                          openmp_threads=openmp_threads)
 
-        self.ntasks_from_config = ntasks is None
-        self.min_tasks_from_config = min_tasks is None
-        self.threads_from_config = openmp_threads is None
+        if (ntasks is None) != (openmp_threads is None):
+            raise ValueError('You must specify both ntasks and openmp_threads '
+                             'or neither.')
+        self.resources_from_config = ntasks is None
 
         # make sure output is double precision
         self.add_streams_file('compass.ocean.streams', 'streams.output')
@@ -140,15 +136,14 @@ def setup(self):
         Set up the test case in the work directory, including downloading any
         dependencies
         """
-        if self.ntasks_from_config:
-            self.ntasks = self.config.getint(
-                'global_ocean', 'forward_ntasks')
-        if self.min_tasks_from_config:
-            self.min_tasks = self.config.getint(
-                'global_ocean', 'forward_min_tasks')
-        if self.threads_from_config:
-            self.openmp_threads = self.config.getint(
-                'global_ocean', 'forward_threads')
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -158,6 +153,16 @@ def run(self):
         add_mesh_and_init_metadata(self.outputs, self.config,
                                    init_filename='init.nc')
 
+    def _get_resources(self):
+        # get the these properties from the config options
+        if self.resources_from_config:
+            config = self.config
+            self.ntasks = config.getint(
+                'global_ocean', 'forward_ntasks')
+            self.min_tasks = config.getint(
+                'global_ocean', 'forward_min_tasks')
+            self.openmp_threads = config.getint(
+                'global_ocean', 'forward_threads')
 
 class ForwardTestCase(TestCase):
     """
@@ -208,27 +213,6 @@ def configure(self):
         """
         configure_global_ocean(test_case=self, mesh=self.mesh, init=self.init)
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for step_name in self.steps_to_run:
-            step = self.steps[step_name]
-            if isinstance(step, ForwardStep):
-                if step.ntasks_from_config:
-                    step.ntasks = config.getint('global_ocean',
-                                                'forward_ntasks')
-                if step.min_tasks_from_config:
-                    step.min_tasks = config.getint('global_ocean',
-                                                   'forward_min_tasks')
-                if step.threads_from_config:
-                    step.threads = config.getint('global_ocean',
-                                                 'forward_threads')
-
-        # run the steps
-        super().run()
-
 
 def get_forward_subdir(init_subdir, time_integrator, name):
     """
diff --git a/compass/ocean/tests/global_ocean/init/__init__.py b/compass/ocean/tests/global_ocean/init/__init__.py
index 13afa89531..acf0bac359 100644
--- a/compass/ocean/tests/global_ocean/init/__init__.py
+++ b/compass/ocean/tests/global_ocean/init/__init__.py
@@ -65,7 +65,7 @@ def __init__(self, test_group, mesh, initial_condition, with_bgc):
 
         if mesh.with_ice_shelf_cavities:
             self.add_step(
-                SshAdjustment(test_case=self, ntasks=4))
+                SshAdjustment(test_case=self))
 
     def configure(self):
         """
@@ -73,29 +73,6 @@ def configure(self):
         """
         configure_global_ocean(test_case=self, mesh=self.mesh, init=self)
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        steps = self.steps_to_run
-        if 'initial_state' in steps:
-            step = self.steps['initial_state']
-            # get the these properties from the config options
-            step.ntasks = config.getint('global_ocean', 'init_ntasks')
-            step.min_tasks = config.getint('global_ocean', 'init_min_tasks')
-            step.threads = config.getint('global_ocean', 'init_threads')
-
-        if 'ssh_adjustment' in steps:
-            step = self.steps['ssh_adjustment']
-            # get the these properties from the config options
-            step.ntasks = config.getint('global_ocean', 'forward_ntasks')
-            step.min_tasks = config.getint('global_ocean', 'forward_min_tasks')
-            step.threads = config.getint('global_ocean', 'forward_threads')
-
-        # run the steps
-        super().run()
-
     def validate(self):
         """
         Test cases can override this method to perform validation of variables
diff --git a/compass/ocean/tests/global_ocean/init/initial_state.py b/compass/ocean/tests/global_ocean/init/initial_state.py
index 39c89acc5b..be05db49d8 100644
--- a/compass/ocean/tests/global_ocean/init/initial_state.py
+++ b/compass/ocean/tests/global_ocean/init/initial_state.py
@@ -148,14 +148,17 @@ def __init__(self, test_case, mesh, initial_condition, with_bgc):
 
     def setup(self):
         """
-        Set up the test case in the work directory, including downloading any
-        dependencies.
+        Get resources at setup from config options
         """
-        # get the these properties from the config options
-        config = self.config
-        self.ntasks = config.getint('global_ocean', 'init_ntasks')
-        self.min_tasks = config.getint('global_ocean', 'init_min_tasks')
-        self.openmp_threads = config.getint('global_ocean', 'init_threads')
+        self._get_resources()
+
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -175,3 +178,10 @@ def run(self):
 
         plot_initial_state(input_file_name='initial_state.nc',
                            output_file_name='initial_state.png')
+
+    def _get_resources(self):
+        # get the these properties from the config options
+        config = self.config
+        self.ntasks = config.getint('global_ocean', 'init_ntasks')
+        self.min_tasks = config.getint('global_ocean', 'init_min_tasks')
+        self.openmp_threads = config.getint('global_ocean', 'init_threads')
diff --git a/compass/ocean/tests/global_ocean/init/ssh_adjustment.py b/compass/ocean/tests/global_ocean/init/ssh_adjustment.py
index fab3ee5731..aed9808447 100644
--- a/compass/ocean/tests/global_ocean/init/ssh_adjustment.py
+++ b/compass/ocean/tests/global_ocean/init/ssh_adjustment.py
@@ -7,8 +7,7 @@ class SshAdjustment(Step):
     A step for iteratively adjusting the pressure from the weight of the ice
     shelf to match the sea-surface height as part of ice-shelf 2D test cases
     """
-    def __init__(self, test_case, ntasks=None, min_tasks=None,
-                 openmp_threads=None):
+    def __init__(self, test_case):
         """
         Create the step
 
@@ -16,25 +15,8 @@ def __init__(self, test_case, ntasks=None, min_tasks=None,
         ----------
         test_case : compass.ocean.tests.global_ocean.init.Init
             The test case this step belongs to
-
-        ntasks : int, optional
-            the number of tasks the step would ideally use.  If fewer tasks
-            are available on the system, the step will run on all available
-            tasks as long as this is not below ``min_tasks``
-
-        min_tasks : int, optional
-            the number of tasks the step requires.  If the system has fewer
-            than this number of tasks, the step will fail
-
-        openmp_threads : int, optional
-            the number of OpenMP threads the step will use
-
         """
-        if min_tasks is None:
-            min_tasks = ntasks
-        super().__init__(test_case=test_case, name='ssh_adjustment',
-                         ntasks=ntasks, min_tasks=min_tasks,
-                         openmp_threads=openmp_threads)
+        super().__init__(test_case=test_case, name='ssh_adjustment')
 
         # make sure output is double precision
         self.add_streams_file('compass.ocean.streams', 'streams.output')
@@ -71,15 +53,14 @@ def setup(self):
         Set up the test case in the work directory, including downloading any
         dependencies
         """
-        if self.ntasks is None:
-            self.ntasks = self.config.getint(
-                'global_ocean', 'forward_ntasks')
-        if self.min_tasks is None:
-            self.min_tasks = self.config.getint(
-                'global_ocean', 'forward_min_tasks')
-        if self.openmp_threads is None:
-            self.openmp_threads = self.config.getint(
-                'global_ocean', 'forward_threads')
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -89,3 +70,10 @@ def run(self):
         iteration_count = config.getint('ssh_adjustment', 'iterations')
         adjust_ssh(variable='landIcePressure', iteration_count=iteration_count,
                    step=self)
+
+    def _get_resources(self):
+        # get the these properties from the config options
+        config = self.config
+        self.ntasks = config.getint('global_ocean', 'forward_ntasks')
+        self.min_tasks = config.getint('global_ocean', 'forward_min_tasks')
+        self.openmp_threads = config.getint('global_ocean', 'forward_threads')
\ No newline at end of file
diff --git a/compass/ocean/tests/hurricane/forward/__init__.py b/compass/ocean/tests/hurricane/forward/__init__.py
index 630cce30bc..5f08a482c5 100644
--- a/compass/ocean/tests/hurricane/forward/__init__.py
+++ b/compass/ocean/tests/hurricane/forward/__init__.py
@@ -56,10 +56,3 @@ def configure(self):
         Modify the configuration options for this test case
         """
         configure_hurricane(test_case=self, mesh=self.mesh)
-
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        # run the steps
-        super().run()
diff --git a/compass/ocean/tests/hurricane/forward/forward.py b/compass/ocean/tests/hurricane/forward/forward.py
index cc2047cce0..ad705a2e3d 100644
--- a/compass/ocean/tests/hurricane/forward/forward.py
+++ b/compass/ocean/tests/hurricane/forward/forward.py
@@ -69,13 +69,24 @@ def setup(self):
         Set up the test case in the work directory, including downloading any
         dependencies
         """
-        self.ntasks = self.config.getint('hurricane', 'forward_ntasks')
-        self.min_tasks = self.config.getint('hurricane', 'forward_min_tasks')
-        self.openmp_threads = self.config.getint('hurricane',
-                                                 'forward_threads')
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
         Run this step of the testcase
         """
         run_model(self)
+
+    def _get_resources(self):
+        # get the these properties from the config options
+        config = self.config
+        self.ntasks = config.getint('hurricane', 'forward_ntasks')
+        self.min_tasks = config.getint('hurricane', 'forward_min_tasks')
+        self.openmp_threads = config.getint('hurricane', 'forward_threads')
diff --git a/compass/ocean/tests/hurricane/init/__init__.py b/compass/ocean/tests/hurricane/init/__init__.py
index 42743b38f4..0a81ca5730 100644
--- a/compass/ocean/tests/hurricane/init/__init__.py
+++ b/compass/ocean/tests/hurricane/init/__init__.py
@@ -51,11 +51,3 @@ def configure(self):
         Modify the configuration options for this test case
         """
         configure_hurricane(test_case=self, mesh=self.mesh)
-
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-
-        # run the steps
-        super().run()
diff --git a/compass/ocean/tests/hurricane/init/initial_state.py b/compass/ocean/tests/hurricane/init/initial_state.py
index 6a9f13c164..461e689387 100644
--- a/compass/ocean/tests/hurricane/init/initial_state.py
+++ b/compass/ocean/tests/hurricane/init/initial_state.py
@@ -56,16 +56,26 @@ def __init__(self, test_case, mesh):
     def setup(self):
         """
         Set up the test case in the work directory, including downloading any
-        dependencies.
+        dependencies
         """
-        # get the these properties from the config options
-        config = self.config
-        self.ntasks = config.getint('hurricane', 'init_ntasks')
-        self.min_tasks = config.getint('hurricane', 'init_min_tasks')
-        self.openmp_threads = config.getint('hurricane', 'init_threads')
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
         Run this step of the testcase
         """
         run_model(self)
+
+    def _get_resources(self):
+        # get the these properties from the config options
+        config = self.config
+        self.ntasks = config.getint('hurricane', 'init_ntasks')
+        self.min_tasks = config.getint('hurricane', 'init_min_tasks')
+        self.openmp_threads = config.getint('hurricane', 'init_threads')
diff --git a/compass/ocean/tests/isomip_plus/forward.py b/compass/ocean/tests/isomip_plus/forward.py
index 4f62a18571..3e9a9ef390 100644
--- a/compass/ocean/tests/isomip_plus/forward.py
+++ b/compass/ocean/tests/isomip_plus/forward.py
@@ -57,7 +57,7 @@ def __init__(self, test_case, resolution, experiment, name='forward',
         self.resolution = resolution
         self.experiment = experiment
         super().__init__(test_case=test_case, name=name, subdir=subdir,
-                         ntasks=None, min_tasks=None, openmp_threads=1)
+                         ntasks=None, min_tasks=None, openmp_threads=None)
 
         # make sure output is double precision
         self.add_streams_file('compass.ocean.streams', 'streams.output')
@@ -126,7 +126,19 @@ def __init__(self, test_case, resolution, experiment, name='forward',
         self.add_output_file('output.nc')
         self.add_output_file('land_ice_fluxes.nc')
 
-    # no setup() is needed
+    def setup(self):
+        """
+        Set up the test case in the work directory, including downloading any
+        dependencies
+        """
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -170,6 +182,15 @@ def run(self):
                             'config_start_time': "'file'"}
             self.update_namelist_at_runtime(replacements)
 
+    def _get_resources(self):
+        """
+        Get resources (ntasks, min_tasks, and openmp_threads) from the config
+        options
+        """
+        config = self.config
+        self.ntasks = config.getint('isomip_plus', 'forward_ntasks')
+        self.min_tasks = config.getint('isomip_plus', 'forward_min_tasks')
+        self.openmp_threads = config.getint('isomip_plus', 'forward_threads')
 
 def get_time_steps(resolution):
     """
diff --git a/compass/ocean/tests/isomip_plus/isomip_plus_test.py b/compass/ocean/tests/isomip_plus/isomip_plus_test.py
index 1cf209bdeb..13f4301ee4 100644
--- a/compass/ocean/tests/isomip_plus/isomip_plus_test.py
+++ b/compass/ocean/tests/isomip_plus/isomip_plus_test.py
@@ -220,32 +220,6 @@ def configure(self):
                 # default to 10 vertical levels instead of 36
                 config.set('vertical_grid', 'vert_levels', '10')
 
-        for step_name in self.steps:
-            if step_name in ['ssh_adjustment', 'performance', 'simulation']:
-                step = self.steps[step_name]
-                step.ntasks = ntasks
-                step.min_tasks = min_tasks
-                step.openmp_threads = 1
-
-    def run(self):
-        """
-        Run each step of the test case
-        """
-        config = self.config
-        # get the these properties from the config options
-        for step_name in self.steps_to_run:
-            if step_name in ['ssh_adjustment', 'performance', 'simulation']:
-                step = self.steps[step_name]
-                # get the these properties from the config options
-                step.ntasks = config.getint('isomip_plus', 'forward_ntasks')
-                step.min_tasks = config.getint('isomip_plus',
-                                               'forward_min_tasks')
-                step.openmp_threads = config.getint('isomip_plus',
-                                                    'forward_threads')
-
-        # run the steps
-        super().run()
-
     def validate(self):
         """
         Perform validation of variables
diff --git a/compass/ocean/tests/isomip_plus/ssh_adjustment.py b/compass/ocean/tests/isomip_plus/ssh_adjustment.py
index 0623a16994..7e35a2f4d9 100644
--- a/compass/ocean/tests/isomip_plus/ssh_adjustment.py
+++ b/compass/ocean/tests/isomip_plus/ssh_adjustment.py
@@ -8,8 +8,7 @@ class SshAdjustment(Step):
     A step for iteratively adjusting the pressure from the weight of the ice
     shelf to match the sea-surface height as part of ice-shelf 2D test cases
     """
-    def __init__(self, test_case, resolution, ntasks=1, min_tasks=None,
-                 openmp_threads=1, vertical_coordinate='z-star',
+    def __init__(self, test_case, resolution, vertical_coordinate='z-star',
                  thin_film_present=False):
         """
         Create the step
@@ -21,25 +20,10 @@ def __init__(self, test_case, resolution, ntasks=1, min_tasks=None,
 
         resolution : float
             The horizontal resolution (km) of the test case
-
-        ntasks : int, optional
-            the number of tasks the step would ideally use.  If fewer tasks
-            are available on the system, the step will run on all available
-            tasks as long as this is not below ``min_tasks``
-
-        min_tasks : int, optional
-            the number of tasks the step requires.  If the system has fewer
-            than this number of tasks, the step will fail
-
-        openmp_threads : int, optional
-            the number of OpenMP threads the step will use
-
         """
-        if min_tasks is None:
-            min_tasks = ntasks
         super().__init__(test_case=test_case, name='ssh_adjustment',
-                         ntasks=ntasks, min_tasks=min_tasks,
-                         openmp_threads=openmp_threads)
+                         ntasks=None, min_tasks=None,
+                         openmp_threads=None)
 
         # generate the namelist, replacing a few default options
         # start with the same namelist settings as the forward run
@@ -76,7 +60,19 @@ def __init__(self, test_case, resolution, ntasks=1, min_tasks=None,
 
         self.add_output_file(filename='adjusted_init.nc')
 
-    # no setup() is needed
+    def setup(self):
+        """
+        Set up the test case in the work directory, including downloading any
+        dependencies
+        """
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -86,3 +82,13 @@ def run(self):
         iteration_count = config.getint('ssh_adjustment', 'iterations')
         adjust_ssh(variable='landIcePressure', iteration_count=iteration_count,
                    step=self)
+
+    def _get_resources(self):
+        """
+        Get resources (ntasks, min_tasks, and openmp_threads) from the config
+        options
+        """
+        config = self.config
+        self.ntasks = config.getint('isomip_plus', 'forward_ntasks')
+        self.min_tasks = config.getint('isomip_plus', 'forward_min_tasks')
+        self.openmp_threads = config.getint('isomip_plus', 'forward_threads')
diff --git a/compass/ocean/tests/planar_convergence/conv_test_case.py b/compass/ocean/tests/planar_convergence/conv_test_case.py
index 2adf9a7d71..abbca5b5b9 100644
--- a/compass/ocean/tests/planar_convergence/conv_test_case.py
+++ b/compass/ocean/tests/planar_convergence/conv_test_case.py
@@ -44,23 +44,6 @@ def configure(self):
 
         self.update_cores()
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for resolution in self.resolutions:
-            ntasks = config.getint(f'planar_convergence',
-                                   f'{resolution}km_ntasks')
-            min_tasks = config.getint(f'planar_convergence',
-                                      f'{resolution}km_min_tasks')
-            step = self.steps['{}km_forward'.format(resolution)]
-            step.ntasks = ntasks
-            step.min_tasks = min_tasks
-
-        # run the step
-        super().run()
-
     def update_cores(self):
         """ Update the number of cores and min_tasks for each forward step """
 
diff --git a/compass/ocean/tests/planar_convergence/forward.py b/compass/ocean/tests/planar_convergence/forward.py
index faae46c256..765f5a1d6d 100644
--- a/compass/ocean/tests/planar_convergence/forward.py
+++ b/compass/ocean/tests/planar_convergence/forward.py
@@ -68,6 +68,14 @@ def setup(self):
         self.add_streams_file('compass.ocean.tests.planar_convergence',
                               'streams.template',
                               template_replacements=stream_replacements)
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -118,3 +126,11 @@ def get_dt_duration(self):
         stream_replacements = {'output_interval': duration}
 
         return namelist_replacements, stream_replacements
+
+    def _get_resources(self):
+        config = self.config
+        resolution = self.resolution
+        self.ntasks = config.getint(f'planar_convergence',
+                                    f'{resolution}km_ntasks')
+        self.min_tasks = config.getint(f'planar_convergence',
+                                       f'{resolution}km_min_tasks')
diff --git a/compass/ocean/tests/sphere_transport/correlated_tracers_2d/__init__.py b/compass/ocean/tests/sphere_transport/correlated_tracers_2d/__init__.py
index 0bc131e773..48032a1ffb 100644
--- a/compass/ocean/tests/sphere_transport/correlated_tracers_2d/__init__.py
+++ b/compass/ocean/tests/sphere_transport/correlated_tracers_2d/__init__.py
@@ -57,23 +57,6 @@ def configure(self):
 
         self.update_cores()
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for resolution in self.resolutions:
-            ntasks = config.getint('correlated_tracers_2d',
-                                   f'QU{resolution}_ntasks')
-            min_tasks = config.getint('correlated_tracers_2d',
-                                      f'QU{resolution}_min_tasks')
-            step = self.steps[f'QU{resolution}_forward']
-            step.ntasks = ntasks
-            step.min_tasks = min_tasks
-
-        # run the step
-        super().run()
-
     def update_cores(self):
         """ Update the number of cores and min_tasks for each forward step """
 
diff --git a/compass/ocean/tests/sphere_transport/correlated_tracers_2d/forward.py b/compass/ocean/tests/sphere_transport/correlated_tracers_2d/forward.py
index 663e0c9a7b..fa3891c9e9 100644
--- a/compass/ocean/tests/sphere_transport/correlated_tracers_2d/forward.py
+++ b/compass/ocean/tests/sphere_transport/correlated_tracers_2d/forward.py
@@ -62,6 +62,14 @@ def setup(self):
                                    config.get(
                                        'correlated_tracers_2d',
                                        'time_integrator')})
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -100,3 +108,11 @@ def get_timestep_str(self):
             dtminutes = dt / timedelta(minutes=1)
             dtstr = "00:" + str(int(dtminutes))[:2].zfill(2) + ":00"
         return dtstr
+
+    def _get_resources(self):
+        resolution = self.resolution
+        config = self.config
+        self.ntasks = config.getint('correlated_tracers_2d',
+                                    f'QU{resolution}_ntasks')
+        self.min_tasks = config.getint('correlated_tracers_2d',
+                                       f'QU{resolution}_min_tasks')
diff --git a/compass/ocean/tests/sphere_transport/correlated_tracers_2d/namelist.forward b/compass/ocean/tests/sphere_transport/correlated_tracers_2d/namelist.forward
index 24c9741172..ccd141f4b3 100644
--- a/compass/ocean/tests/sphere_transport/correlated_tracers_2d/namelist.forward
+++ b/compass/ocean/tests/sphere_transport/correlated_tracers_2d/namelist.forward
@@ -1,5 +1,4 @@
 config_ocean_run_mode = 'forward'
-config_init_configuration = 'transport_tests'
 config_transport_tests_flow_id = 4
 config_run_duration = '0012_00:00:00'
 config_vert_coord_movement = 'impermeable_interfaces'
diff --git a/compass/ocean/tests/sphere_transport/divergent_2d/__init__.py b/compass/ocean/tests/sphere_transport/divergent_2d/__init__.py
index dc539d484a..b9e4c26aa9 100644
--- a/compass/ocean/tests/sphere_transport/divergent_2d/__init__.py
+++ b/compass/ocean/tests/sphere_transport/divergent_2d/__init__.py
@@ -53,23 +53,6 @@ def configure(self):
 
         self.update_cores()
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for resolution in self.resolutions:
-            ntasks = config.getint('divergent_2d',
-                                   f'QU{resolution}_ntasks')
-            min_tasks = config.getint('divergent_2d',
-                                      f'QU{resolution}_min_tasks')
-            step = self.steps[f'QU{resolution}_forward']
-            step.ntasks = ntasks
-            step.min_tasks = min_tasks
-
-        # run the step
-        super().run()
-
     def update_cores(self):
         """ Update the number of cores and min_tasks for each forward step """
 
diff --git a/compass/ocean/tests/sphere_transport/divergent_2d/forward.py b/compass/ocean/tests/sphere_transport/divergent_2d/forward.py
index da4fea78ab..c39ab373a2 100644
--- a/compass/ocean/tests/sphere_transport/divergent_2d/forward.py
+++ b/compass/ocean/tests/sphere_transport/divergent_2d/forward.py
@@ -59,6 +59,14 @@ def setup(self):
         self.add_namelist_options({'config_dt': dtstr,
                                    'config_time_integrator': config.get(
                                        'divergent_2d', 'time_integrator')})
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -97,3 +105,11 @@ def get_timestep_str(self):
             dtminutes = dt / timedelta(minutes=1)
             dtstr = "00:" + str(int(dtminutes))[:2].zfill(2) + ":00"
         return dtstr
+
+    def _get_resources(self):
+        resolution = self.resolution
+        config = self.config
+        self.ntasks = config.getint('divergent_2d',
+                                    f'QU{resolution}_ntasks')
+        self.min_tasks = config.getint('divergent_2d',
+                                       f'QU{resolution}_min_tasks')
diff --git a/compass/ocean/tests/sphere_transport/divergent_2d/namelist.forward b/compass/ocean/tests/sphere_transport/divergent_2d/namelist.forward
index ce877af815..0be77b0253 100644
--- a/compass/ocean/tests/sphere_transport/divergent_2d/namelist.forward
+++ b/compass/ocean/tests/sphere_transport/divergent_2d/namelist.forward
@@ -1,5 +1,4 @@
 config_ocean_run_mode = 'forward'
-config_init_configuration = 'transport_tests'
 config_transport_tests_flow_id = 3
 config_run_duration = '0012_00:00:00'
 config_vert_coord_movement = 'impermeable_interfaces'
diff --git a/compass/ocean/tests/sphere_transport/nondivergent_2d/__init__.py b/compass/ocean/tests/sphere_transport/nondivergent_2d/__init__.py
index 9ec56d86d1..76f9a368a8 100644
--- a/compass/ocean/tests/sphere_transport/nondivergent_2d/__init__.py
+++ b/compass/ocean/tests/sphere_transport/nondivergent_2d/__init__.py
@@ -55,23 +55,6 @@ def configure(self):
 
         self.update_cores()
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for resolution in self.resolutions:
-            ntasks = config.getint('nondivergent_2d',
-                                   f'QU{resolution}_ntasks')
-            min_tasks = config.getint('nondivergent_2d',
-                                      f'QU{resolution}_min_tasks')
-            step = self.steps[f'QU{resolution}_forward']
-            step.ntasks = ntasks
-            step.min_tasks = min_tasks
-
-        # run the step
-        super().run()
-
     def update_cores(self):
         """ Update the number of cores and min_tasks for each forward step """
 
diff --git a/compass/ocean/tests/sphere_transport/nondivergent_2d/forward.py b/compass/ocean/tests/sphere_transport/nondivergent_2d/forward.py
index 6dd7ca4fc1..1e666eb292 100644
--- a/compass/ocean/tests/sphere_transport/nondivergent_2d/forward.py
+++ b/compass/ocean/tests/sphere_transport/nondivergent_2d/forward.py
@@ -60,6 +60,14 @@ def setup(self):
         self.add_namelist_options({'config_dt': dtstr,
                                    'config_time_integrator': config.get(
                                        'nondivergent_2d', 'time_integrator')})
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -98,3 +106,11 @@ def get_timestep_str(self):
             dtminutes = dt / timedelta(minutes=1)
             dtstr = "00:" + str(int(dtminutes))[:2].zfill(2) + ":00"
         return dtstr
+
+    def _get_resources(self):
+        resolution = self.resolution
+        config = self.config
+        self.ntasks = config.getint('nondivergent_2d',
+                                    f'QU{resolution}_ntasks')
+        self.min_tasks = config.getint('nondivergent_2d',
+                                       f'QU{resolution}_min_tasks')
diff --git a/compass/ocean/tests/sphere_transport/nondivergent_2d/namelist.forward b/compass/ocean/tests/sphere_transport/nondivergent_2d/namelist.forward
index 298672efe7..e4e21612ca 100644
--- a/compass/ocean/tests/sphere_transport/nondivergent_2d/namelist.forward
+++ b/compass/ocean/tests/sphere_transport/nondivergent_2d/namelist.forward
@@ -1,5 +1,4 @@
 config_ocean_run_mode = 'forward'
-config_init_configuration = 'transport_tests'
 config_transport_tests_flow_id = 2
 config_run_duration = '0012_00:00:00'
 config_vert_coord_movement = 'impermeable_interfaces'
diff --git a/compass/ocean/tests/sphere_transport/rotation_2d/__init__.py b/compass/ocean/tests/sphere_transport/rotation_2d/__init__.py
index 71984afbc9..5925c3375a 100644
--- a/compass/ocean/tests/sphere_transport/rotation_2d/__init__.py
+++ b/compass/ocean/tests/sphere_transport/rotation_2d/__init__.py
@@ -52,23 +52,6 @@ def configure(self):
 
         self.update_cores()
 
-    def run(self):
-        """
-        Run each step of the testcase
-        """
-        config = self.config
-        for resolution in self.resolutions:
-            ntasks = config.getint('rotation_2d',
-                                   f'QU{resolution}_ntasks')
-            min_tasks = config.getint('rotation_2d',
-                                      f'QU{resolution}_min_tasks')
-            step = self.steps[f'QU{resolution}_forward']
-            step.ntasks = ntasks
-            step.min_tasks = min_tasks
-
-        # run the step
-        super().run()
-
     def update_cores(self):
         """ Update the number of cores and min_tasks for each forward step """
 
diff --git a/compass/ocean/tests/sphere_transport/rotation_2d/forward.py b/compass/ocean/tests/sphere_transport/rotation_2d/forward.py
index 3c678c7825..5100658ba6 100644
--- a/compass/ocean/tests/sphere_transport/rotation_2d/forward.py
+++ b/compass/ocean/tests/sphere_transport/rotation_2d/forward.py
@@ -59,6 +59,14 @@ def setup(self):
         self.add_namelist_options({'config_dt': dtstr,
                                    'config_time_integrator': config.get(
                                        'rotation_2d', 'time_integrator')})
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -97,3 +105,11 @@ def get_timestep_str(self):
             dtminutes = dt / timedelta(minutes=1)
             dtstr = "00:" + str(int(dtminutes))[:2].zfill(2) + ":00"
         return dtstr
+
+    def _get_resources(self):
+        resolution = self.resolution
+        config = self.config
+        self.ntasks = config.getint('rotation_2d',
+                                    f'QU{resolution}_ntasks')
+        self.min_tasks = config.getint('rotation_2d',
+                                       f'QU{resolution}_min_tasks')
diff --git a/compass/ocean/tests/sphere_transport/rotation_2d/namelist.forward b/compass/ocean/tests/sphere_transport/rotation_2d/namelist.forward
index 55a7904a0b..8de3e5d0d9 100644
--- a/compass/ocean/tests/sphere_transport/rotation_2d/namelist.forward
+++ b/compass/ocean/tests/sphere_transport/rotation_2d/namelist.forward
@@ -1,5 +1,4 @@
 config_ocean_run_mode = 'forward'
-config_init_configuration = 'transport_tests'
 config_transport_tests_flow_id = 1
 config_run_duration = '0012_00:00:00'
 config_vert_coord_movement = 'impermeable_interfaces'
diff --git a/compass/ocean/tests/tides/forward/forward.py b/compass/ocean/tests/tides/forward/forward.py
index f6ce890212..7e540ea406 100644
--- a/compass/ocean/tests/tides/forward/forward.py
+++ b/compass/ocean/tests/tides/forward/forward.py
@@ -72,12 +72,24 @@ def setup(self):
         Set up the test case in the work directory, including downloading any
         dependencies
         """
-        self.ntasks = self.config.getint('tides', 'forward_ntasks')
-        self.min_tasks = self.config.getint('tides', 'forward_min_tasks')
-        self.threads = self.config.getint('tides', 'forward_threads')
+        self._get_resources()
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
         Run this step of the testcase
         """
         run_model(self)
+
+    def _get_resources(self):
+        # get the these properties from the config options
+        config = self.config
+        self.ntasks = config.getint('tides', 'forward_ntasks')
+        self.min_tasks = config.getint('tides', 'forward_min_tasks')
+        self.openmp_threads = config.getint('tides', 'forward_threads')
diff --git a/compass/ocean/tests/tides/init/initial_state.py b/compass/ocean/tests/tides/init/initial_state.py
index e88419dded..6d502bbf6d 100644
--- a/compass/ocean/tests/tides/init/initial_state.py
+++ b/compass/ocean/tests/tides/init/initial_state.py
@@ -85,14 +85,17 @@ def __init__(self, test_case, mesh):
 
     def setup(self):
         """
-        Set up the test case in the work directory, including downloading any
-        dependencies.
+        Get resources at setup from config options
         """
-        # get the these properties from the config options
-        config = self.config
-        self.ntasks = config.getint('tides', 'init_ntasks')
-        self.min_tasks = config.getint('tides', 'init_min_tasks')
-        self.threads = config.getint('tides', 'init_threads')
+        self._get_resources()
+
+
+    def constrain_resources(self, available_cores):
+        """
+        Update resources at runtime from config options
+        """
+        self._get_resources()
+        super().constrain_resources(available_cores)
 
     def run(self):
         """
@@ -119,3 +122,10 @@ def run(self):
         init["layerThickness"][0, :, 0] = init["bottomDepth"][:]
         init["restingThickness"][:, 0] = init["bottomDepth"][:]
         init.close()
+
+    def _get_resources(self):
+        # get the these properties from the config options
+        config = self.config
+        self.ntasks = config.getint('tides', 'init_ntasks')
+        self.min_tasks = config.getint('tides', 'init_min_tasks')
+        self.openmp_threads = config.getint('tides', 'init_threads')
diff --git a/compass/run/serial.py b/compass/run/serial.py
index f2f942089f..cfa97d1416 100644
--- a/compass/run/serial.py
+++ b/compass/run/serial.py
@@ -431,6 +431,15 @@ def _run_step(test_case, step, new_log_file):
                         log_filename=log_filename) as step_logger:
         step.logger = step_logger
         os.chdir(step.work_dir)
+
+        # runtime_setup() will perform small tasks that require knowing the
+        # resources of the task before the step runs (such as creating
+        # graph partitions)
+        step_logger.info('')
+        log_method_call(method=step.runtime_setup, logger=step_logger)
+        step_logger.info('')
+        step.runtime_setup()
+
         step_logger.info('')
         log_method_call(method=step.run, logger=step_logger)
         step_logger.info('')