diff --git a/test/integration/resubmission_dynamic_job_conf.xml b/test/integration/resubmission_dynamic_job_conf.xml
new file mode 100644
index 00000000000..8a25d200fca
--- /dev/null
+++ b/test/integration/resubmission_dynamic_job_conf.xml
@@ -0,0 +1,35 @@
+<?xml version="1.0"?>
+<!-- 
+    Slimmed down resubmission_job_conf.xml for testing dynamic resubmission rules.
+-->
+<job_conf>
+    <plugins>
+        <plugin id="local" type="runner" load="galaxy.jobs.runners.local:LocalJobRunner" workers="2"/>
+        <plugin id="failure_runner" type="runner" load="integration.resubmission_runners:FailsJobRunner" workers="2">
+        </plugin>
+        <plugin id="dynamic" type="runner">
+            <param id="rules_module">integration.resubmission_rules</param>
+        </plugin>
+    </plugins>
+
+    <handlers>
+        <handler id="main"/>
+    </handlers>
+
+    <destinations default="initial_destination">
+        <destination id="initial_destination" runner="dynamic">
+            <param id="type">python</param>
+            <param id="function">dynamic_resubmit_once</param>
+        </destination>
+
+        <!-- Upload destination. -->
+        <destination id="local" runner="local">
+        </destination>
+
+    </destinations>
+
+    <tools>
+        <tool id="upload1" destination="local" resources="upload" />
+    </tools>
+
+</job_conf>
diff --git a/test/integration/resubmission_rules/rules.py b/test/integration/resubmission_rules/rules.py
index 49c8b10f660..4d82e1d2d08 100644
--- a/test/integration/resubmission_rules/rules.py
+++ b/test/integration/resubmission_rules/rules.py
@@ -1,6 +1,20 @@
+from galaxy.jobs import JobDestination
 
 DEFAULT_INITIAL_DESTINATION = "fail_first_try"
 
 
 def initial_destination(resource_params):
     return resource_params.get("initial_destination", None) or DEFAULT_INITIAL_DESTINATION
+
+
+def dynamic_resubmit_once(resource_params):
+    """Build destination that always fails first time and always re-routes to passing destination."""
+    job_destination = JobDestination()
+    # Always fail on the first attempt.
+    job_destination['runner'] = "failure_runner"
+    # Resubmit to a valid destination.
+    job_destination['resubmit'] = [dict(
+        condition="any_failure",
+        destination="local",
+    )]
+    return job_destination
diff --git a/test/integration/test_job_resubmission.py b/test/integration/test_job_resubmission.py
index cffbdb5e5e8..8401b0be801 100644
--- a/test/integration/test_job_resubmission.py
+++ b/test/integration/test_job_resubmission.py
@@ -7,16 +7,17 @@ from base import integration_util
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 JOB_RESUBMISSION_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_job_conf.xml")
 JOB_RESUBMISSION_DEFAULT_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_default_job_conf.xml")
+JOB_RESUBMISSION_DYNAMIC_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_dynamic_job_conf.xml")
 JOB_RESUBMISSION_JOB_RESOURCES_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "resubmission_job_resource_parameters_conf.xml")
 
 
 class _BaseResubmissionIntegerationTestCase(integration_util.IntegrationTestCase):
     framework_tool_and_types = True
 
-    def _assert_job_passes(self, resource_parameters):
+    def _assert_job_passes(self, resource_parameters={}):
         self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)
 
-    def _assert_job_fails(self, resource_parameters):
+    def _assert_job_fails(self, resource_parameters={}):
         exception_thrown = False
         try:
             self._run_tool_test("simple_constructs", resource_parameters=resource_parameters)
@@ -115,3 +116,15 @@ class JobResubmissionDefaultIntegrationTestCase(_BaseResubmissionIntegerationTes
 
     def test_default_resubmission(self):
         self._assert_job_passes(resource_parameters={"test_name": "test_default_resubmission"})
+
+
+class JobResubmissionDynamicIntegrationTestCase(_BaseResubmissionIntegerationTestCase):
+
+    framework_tool_and_types = True
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["job_config_file"] = JOB_RESUBMISSION_DYNAMIC_JOB_CONFIG_FILE
+
+    def test_dynamic_resubmission(self):
+        self._assert_job_passes()