diff --git a/test/unit/jobs/test_mapper.py b/test/unit/jobs/test_mapper.py
new file mode 100644
index 00000000000..21d636ffc9c
--- /dev/null
+++ b/test/unit/jobs/test_mapper.py
@@ -0,0 +1,129 @@
+import jobs.test_rules
+
+from galaxy.jobs.mapper import JobRunnerMapper
+from galaxy.jobs import JobDestination
+
+from galaxy.util import bunch
+
+
+TOOL_JOB_DESTINATION = JobDestination()
+DYNAMICALLY_GENERATED_DESTINATION = JobDestination()
+
+
+def test_static_mapping():
+    mapper = __mapper()
+    assert mapper.get_job_destination( {} ) is TOOL_JOB_DESTINATION
+
+
+def test_caching():
+    mapper = __mapper()
+    mapper.get_job_destination( {} )
+    mapper.get_job_destination( {} )
+    assert mapper.job_wrapper.tool.call_count == 1
+
+
+def test_dynamic_mapping():
+    mapper = __mapper( __dynamic_destination( dict( function="upload" ) ) )
+    assert mapper.get_job_destination( {} ) is DYNAMICALLY_GENERATED_DESTINATION
+    assert mapper.job_config.rule_response == "local_runner"
+
+
+def test_dynamic_mapping_priorities():
+    mapper = __mapper( __dynamic_destination( dict( function="tophat" ) ) )
+    assert mapper.get_job_destination( {} ) is DYNAMICALLY_GENERATED_DESTINATION
+    # Next line verifies we using definition in 20_instance.py instead of
+    # 10_site.py.
+    assert mapper.job_config.rule_response == "instance_dest_id"
+
+
+def test_dynamic_mapping_defaults_to_tool_id_as_rule():
+    mapper = __mapper( __dynamic_destination( ) )
+    assert mapper.get_job_destination( {} ) is DYNAMICALLY_GENERATED_DESTINATION
+    assert mapper.job_config.rule_response == "tool1_dest_id"
+
+
+def test_dynamic_mapping_function_parameters():
+    mapper = __mapper( __dynamic_destination( dict( function="check_rule_params" ) ) )
+    assert mapper.get_job_destination( {} ) is DYNAMICALLY_GENERATED_DESTINATION
+    assert mapper.job_config.rule_response == "all_passed"
+
+
+def test_dynamic_mapping_resource_parameters():
+    mapper = __mapper( __dynamic_destination( dict( function="check_resource_params" ) ) )
+    assert mapper.get_job_destination( {} ) is DYNAMICALLY_GENERATED_DESTINATION
+    assert mapper.job_config.rule_response == "have_resource_params"
+
+
+def __mapper( tool_job_destination=TOOL_JOB_DESTINATION ):
+    job_wrapper = MockJobWrapper( tool_job_destination )
+    job_config = MockJobConfig()
+
+    mapper = JobRunnerMapper(
+        job_wrapper,
+        {},
+        job_config
+    )
+    mapper.rules_module = jobs.test_rules
+    return mapper
+
+
+def __dynamic_destination( params={} ):
+    return JobDestination( runner="dynamic", params=params )
+
+
+class MockJobConfig( object ):
+
+    def __init__( self ):
+        self.rule_response = None
+
+    def get_destination( self, rep ):
+        # Called to transform dynamic job destination rule response
+        # from destination id/runner url into a dynamic job destination.
+        self.rule_response = rep
+        return DYNAMICALLY_GENERATED_DESTINATION
+
+
+class MockJobWrapper( object ):
+
+    def __init__( self, tool_job_destination ):
+        self.tool = MockTool( tool_job_destination )
+        self.job_id = 12345
+        self.app = object()
+
+    def is_mock_job_wrapper( self ):
+        return True
+
+    def get_job(self):
+        def get_param_values( app, ignore_errors ):
+            assert app == self.app
+            return {
+                "threshold": 8,
+                "__job_resource": {
+                    "__job_resource__select": "True",
+                    "memory": "8gb"
+                }
+            }
+
+        return bunch.Bunch(
+            user=bunch.Bunch(
+                id=6789,
+                email="test@example.com"
+            ),
+            get_param_values=get_param_values
+        )
+
+
+class MockTool( object ):
+
+    def __init__( self, tool_job_destination ):
+        self.id = "testtoolshed/devteam/tool1/23abcd13123"
+        self.call_count = 0
+        self.tool_job_destination = tool_job_destination
+        self.all_ids = [ "testtoolshed/devteam/tool1/23abcd13123", "tool1" ]
+
+    def get_job_destination( self, params ):
+        self.call_count += 1
+        return self.tool_job_destination
+
+    def is_mock_tool( self ):
+        return True
diff --git a/test/unit/jobs/test_rule_helper.py b/test/unit/jobs/test_rule_helper.py
new file mode 100644
index 00000000000..9a83f811961
--- /dev/null
+++ b/test/unit/jobs/test_rule_helper.py
@@ -0,0 +1,98 @@
+from galaxy.util import bunch
+from galaxy import model
+from galaxy.model import mapping
+
+from galaxy.jobs.rule_helper import RuleHelper
+
+USER_EMAIL_1 = "u1@example.com"
+USER_EMAIL_2 = "u2@example.com"
+USER_EMAIL_3 = "u3@example.com"
+
+
+def test_job_count():
+    rule_helper = __rule_helper()
+    __assert_job_count_is( 0, rule_helper )
+
+    __setup_fixtures( rule_helper.app )
+
+    # Test raw counts for users...
+    __assert_job_count_is( 7, rule_helper, for_user_email=USER_EMAIL_1 )
+    __assert_job_count_is( 2, rule_helper, for_user_email=USER_EMAIL_2 )
+    __assert_job_count_is( 0, rule_helper, for_user_email=USER_EMAIL_3 )
+
+    # Test desitnation counts
+    __assert_job_count_is( 2, rule_helper, for_destination="local" )
+    __assert_job_count_is( 7, rule_helper, for_destination="cluster1" )
+
+    # Test per user destination counts
+    __assert_job_count_is( 5, rule_helper, for_destination="cluster1", for_user_email=USER_EMAIL_1 )
+    __assert_job_count_is( 2, rule_helper, for_destination="local", for_user_email=USER_EMAIL_1 )
+
+    __assert_job_count_is( 2, rule_helper, for_destination="cluster1", for_user_email=USER_EMAIL_2 )
+    __assert_job_count_is( 0, rule_helper, for_destination="local", for_user_email=USER_EMAIL_2 )
+
+    # Test per user, per state destination counts
+    __assert_job_count_is( 3, rule_helper, for_destination="cluster1", for_user_email=USER_EMAIL_1, for_job_states=[ "queued" ] )
+    __assert_job_count_is( 2, rule_helper, for_destination="cluster1", for_user_email=USER_EMAIL_1, for_job_states=[ "running" ] )
+    __assert_job_count_is( 0, rule_helper, for_destination="cluster1", for_user_email=USER_EMAIL_1, for_job_states=[ "error" ] )
+    __assert_job_count_is( 5, rule_helper, for_destination="cluster1", for_user_email=USER_EMAIL_1, for_job_states=[ "queued", "running", "error" ] )
+
+
+def __assert_job_count_is( expected_count, rule_helper, **kwds ):
+    acutal_count = rule_helper.job_count( **kwds )
+
+    if expected_count != acutal_count:
+        template = "Expected job count %d, actual job count %s for params %s"
+        raise AssertionError( template % ( expected_count, acutal_count, kwds ) )
+
+
+def __setup_fixtures( app ):
+    # user1 has 3 jobs queued and 2 jobs running on cluster1 and one queued and
+    # on running job on local. user2 has a queued and running job on the cluster.
+    # user3 has no jobs.
+    user1 = model.User( email=USER_EMAIL_1, password="pass1" )
+    user2 = model.User( email=USER_EMAIL_2, password="pass2" )
+    user3 = model.User( email=USER_EMAIL_2, password="pass2" )
+
+    app.add( user1, user2, user3 )
+
+    app.add( __new_job( user=user1, destination_id="cluster1", state="queued" ) )
+    app.add( __new_job( user=user1, destination_id="cluster1", state="queued" ) )
+    app.add( __new_job( user=user1, destination_id="cluster1", state="queued" ) )
+    app.add( __new_job( user=user1, destination_id="cluster1", state="running" ) )
+    app.add( __new_job( user=user1, destination_id="cluster1", state="running" ) )
+
+    app.add( __new_job( user=user1, destination_id="local", state="queued" ) )
+    app.add( __new_job( user=user1, destination_id="local", state="running" ) )
+
+    app.add( __new_job( user=user2, destination_id="cluster1", state="queued" ) )
+    app.add( __new_job( user=user2, destination_id="cluster1", state="running" ) )
+
+
+def __rule_helper():
+    app = MockApp()
+    rule_helper = RuleHelper( app )
+    return rule_helper
+
+
+def __new_job( **kwds ):
+    job = model.Job()
+    for key, value in kwds.items():
+        setattr( job, key, value )
+    return job
+
+
+class MockApp( object ):
+
+    def __init__( self ):
+        self.config = bunch.Bunch( )
+        self.model = mapping.init(
+            "/tmp",
+            "sqlite:///:memory:",
+            create_tables=True
+        )
+
+    def add( self, *args ):
+        for arg in args:
+            self.model.context.add( arg )
+        self.model.context.flush()
diff --git a/test/unit/jobs/test_rules/10_site.py b/test/unit/jobs/test_rules/10_site.py
new file mode 100644
index 00000000000..5f9d558a7f8
--- /dev/null
+++ b/test/unit/jobs/test_rules/10_site.py
@@ -0,0 +1,45 @@
+
+
+def upload():
+    return 'local_runner'
+
+
+def tophat():
+    return 'site_dest_id'
+
+
+def tool1():
+    # tool1 is id to test tool mocked out in test_mapper.py, without specify
+    # function name in dynamic destination - this function should be used by
+    # default.
+    return 'tool1_dest_id'
+
+
+def check_rule_params(
+    job_id,
+    tool,
+    tool_id,
+    job_wrapper,
+    rule_helper,
+    app,
+    job,
+    user,
+    user_email,
+):
+    assert job_id == 12345
+    assert tool.is_mock_tool()
+    assert tool_id == "testtoolshed/devteam/tool1/23abcd13123"
+    assert job_wrapper.is_mock_job_wrapper()
+    assert app == job_wrapper.app
+    assert rule_helper is not None
+
+    assert job.user == user
+    assert user.id == 6789
+    assert user_email == "test@example.com"
+
+    return "all_passed"
+
+
+def check_resource_params( resource_params ):
+    assert resource_params["memory"] == "8gb"
+    return "have_resource_params"
diff --git a/test/unit/jobs/test_rules/20_instance.py b/test/unit/jobs/test_rules/20_instance.py
new file mode 100644
index 00000000000..916592bb880
--- /dev/null
+++ b/test/unit/jobs/test_rules/20_instance.py
@@ -0,0 +1,6 @@
+
+
+def tophat():
+    # This should override definition in 10_site.py
+    return 'instance_dest_id'
+
diff --git a/test/unit/jobs/test_rules/__init__.py b/test/unit/jobs/test_rules/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d