refactor the job type name into a single property

awx/api/serializers.py

@@ -1697,7 +1697,7 @@ class HostSerializer(BaseSerializerWithVariables):
         d.setdefault('recent_jobs', [{
             'id': j.job.id,
             'name': j.job.job_template.name if j.job.job_template is not None else "",
-            'type': j.job.get_real_instance_class()._meta.verbose_name.replace(' ', '_'),
+            'type': j.job.job_type_name,
             'status': j.job.status,
             'finished': j.job.finished,
         } for j in obj.job_host_summaries.select_related('job__job_template').order_by('-created')[:5]])
@@ -2842,7 +2842,7 @@ class JobTemplateMixin(object):
         return [{
             'id': x.id, 'status': x.status, 'finished': x.finished, 'canceled_on': x.canceled_on,
             # Make type consistent with API top-level key, for instance workflow_job
-            'type': x.get_real_instance_class()._meta.verbose_name.replace(' ', '_')
+            'type': x.job_type_name
         } for x in optimized_qs[:10]]
 
     def get_summary_fields(self, obj):

awx/main/models/unified_jobs.py

@@ -962,6 +962,10 @@ class UnifiedJob(PolymorphicModel, PasswordFieldsModel, CommonModelNameNotUnique
     def event_class(self):
         raise NotImplementedError()
 
+    @property
+    def job_type_name(self):
+        return self.get_real_instance_class()._meta.verbose_name.replace(' ', '_')
+
     @property
     def result_stdout_text(self):
         related = UnifiedJobDeprecatedStdout.objects.get(pk=self.pk)
@@ -1221,7 +1225,7 @@ class UnifiedJob(PolymorphicModel, PasswordFieldsModel, CommonModelNameNotUnique
 
     def websocket_emit_data(self):
         ''' Return extra data that should be included when submitting data to the browser over the websocket connection '''
-        websocket_data = dict(type=self.get_real_instance_class()._meta.verbose_name.replace(' ', '_'))
+        websocket_data = dict(type=self.job_type_name)
         if self.spawned_by_workflow:
             websocket_data.update(dict(workflow_job_id=self.workflow_job_id,
                                        workflow_node_id=self.workflow_node_id))

awx/main/tests/functional/test_jobs.py

@@ -2,7 +2,9 @@ import pytest
 from unittest import mock
 import json
 
-from awx.main.models import Job, Instance, JobHostSummary
+from awx.main.models import (Job, Instance, JobHostSummary, InventoryUpdate,
+                             InventorySource, Project, ProjectUpdate,
+                             SystemJob, AdHocCommand)
 from awx.main.tasks import cluster_node_heartbeat
 from django.test.utils import override_settings
 
@@ -33,6 +35,31 @@ def test_job_capacity_and_with_inactive_node():
         assert i.capacity == 0
 
 
+@pytest.mark.django_db
+def test_job_type_name():
+    job = Job.objects.create()
+    assert job.job_type_name == 'job'
+
+    ahc = AdHocCommand.objects.create()
+    assert ahc.job_type_name == 'ad_hoc_command'
+
+    source = InventorySource.objects.create(source='ec2')
+    source.save()
+    iu = InventoryUpdate.objects.create(
+        inventory_source=source,
+        source='ec2'
+    )
+    assert iu.job_type_name == 'inventory_update'
+
+    proj = Project.objects.create()
+    proj.save()
+    pu = ProjectUpdate.objects.create(project=proj)
+    assert pu.job_type_name == 'project_update'
+
+    sjob = SystemJob.objects.create()
+    assert sjob.job_type_name == 'system_job'
+
+
 @pytest.mark.django_db
 def test_job_notification_data(inventory, machine_credential, project):
     encrypted_str = "$encrypted$"