diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index b9ee2eade6..b653cb7d1f 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -516,11 +516,30 @@ class RunJob(BaseTask):
             args.extend(['-l', job.limit])
         if job.verbosity:
             args.append('-%s' % ('v' * min(3, job.verbosity)))
-        if job.extra_vars_dict:
-            args.extend(['-e', json.dumps(job.extra_vars_dict)])
         if job.job_tags:
             args.extend(['-t', job.job_tags])
-        args.append(job.playbook) # relative path to project.local_path
+
+        # Define special extra_vars for Tower, combine with job.extra_vars.
+        extra_vars = {
+            'tower_job_id': job.pk,
+            'tower_job_launch_type': job.launch_type,
+        }
+        if job.job_template and job.job_template.active:
+            extra_vars.update({
+                'tower_job_template_id': job.job_template.pk,
+                'tower_job_template_name': job.job_template.name,
+            })
+        if job.created_by and job.created_by.is_active:
+            extra_vars.update({
+                'tower_user_id': job.created_by.pk,
+                'tower_user_name': job.created_by.username,
+            })
+        if job.extra_vars_dict:
+            extra_vars.update(job.extra_vars_dict)
+        args.extend(['-e', json.dumps(extra_vars)])
+
+        # Add path to playbook (relative to project.local_path).
+        args.append(job.playbook)
 
         # If ssh unlock password is needed, run using ssh-agent.
         if ssh_key_path and use_ssh_agent:
diff --git a/awx/main/tests/tasks.py b/awx/main/tests/tasks.py
index f6cb961d68..7df4203880 100644
--- a/awx/main/tests/tasks.py
+++ b/awx/main/tests/tasks.py
@@ -12,6 +12,9 @@ from django.conf import settings
 from django.test.utils import override_settings
 from django.utils.timezone import now
 
+# Django-CRUM
+from crum import impersonate
+
 # AWX
 from awx.main.models import *
 from awx.main.tests.base import BaseLiveServerTest
@@ -33,6 +36,20 @@ TEST_PLAYBOOK2 = '''- hosts: test-group
     command: test 1 = 0
 '''
 
+TEST_EXTRA_VARS_PLAYBOOK = '''- hosts: test-group
+  gather_facts: false
+  tasks:
+  - fail: msg="{{item}} is not defined"
+    when: "{{item}} is not defined"
+    with_items:
+    - tower_job_id
+    - tower_job_launch_type
+    - tower_job_template_id
+    - tower_job_template_name
+    - tower_user_id
+    - tower_user_name
+'''
+
 TEST_ENV_PLAYBOOK = '''- hosts: test-group
   gather_facts: False
   tasks:
@@ -273,23 +290,24 @@ class RunJobTest(BaseCeleryTest):
         return self.job_template
 
     def create_test_job(self, **kwargs):
-        job_template = kwargs.pop('job_template', None)
-        if job_template:
-            self.job = job_template.create_job(**kwargs)
-        else:
-            opts = {
-                'inventory': self.inventory,
-                'project': self.project,
-                'credential': self.credential,
-                'cloud_credential': self.cloud_credential,
-                'job_type': 'run',
-            }
-            try:
-                opts['playbook'] = self.project.playbooks[0]
-            except (AttributeError, IndexError):
-                pass
-            opts.update(kwargs)
-            self.job = Job.objects.create(**opts)
+        with impersonate(self.super_django_user):
+            job_template = kwargs.pop('job_template', None)
+            if job_template:
+                self.job = job_template.create_job(**kwargs)
+            else:
+                opts = {
+                    'inventory': self.inventory,
+                    'project': self.project,
+                    'credential': self.credential,
+                    'cloud_credential': self.cloud_credential,
+                    'job_type': 'run',
+                }
+                try:
+                    opts['playbook'] = self.project.playbooks[0]
+                except (AttributeError, IndexError):
+                    pass
+                opts.update(kwargs)
+                self.job = Job.objects.create(**opts)
         return self.job
 
     def check_job_result(self, job, expected='successful', expect_stdout=True,
@@ -707,7 +725,7 @@ class RunJobTest(BaseCeleryTest):
         self.assertFalse(job.signal_start())
 
     def test_extra_job_options(self):
-        self.create_test_project(TEST_PLAYBOOK)
+        self.create_test_project(TEST_EXTRA_VARS_PLAYBOOK)
         # Test with extra_vars containing misc whitespace.
         job_template = self.create_test_job_template(forks=3, verbosity=2,
                                                      extra_vars=u'{\n\t"abc": 1234\n}')
@@ -736,7 +754,7 @@ class RunJobTest(BaseCeleryTest):
         self.check_job_result(job3, 'successful')
 
     def test_lots_of_extra_vars(self):
-        self.create_test_project(TEST_PLAYBOOK)
+        self.create_test_project(TEST_EXTRA_VARS_PLAYBOOK)
         extra_vars = dict(('var_%d' % x, x) for x in xrange(200))
         job_template = self.create_test_job_template(extra_vars=extra_vars)
         job = self.create_test_job(job_template=job_template)