Implement an option to choose a job type on relaunch (issue #14177)

awx/api/serializers.py

@@ -3352,11 +3352,17 @@ class JobRelaunchSerializer(BaseSerializer):
         choices=[('all', _('No change to job limit')), ('failed', _('All failed and unreachable hosts'))],
         write_only=True,
     )
+    job_type = serializers.ChoiceField(
+        required=False,
+        allow_null=True,
+        choices=NEW_JOB_TYPE_CHOICES,
+        write_only=True,
+    )
     credential_passwords = VerbatimField(required=True, write_only=True)
 
     class Meta:
         model = Job
-        fields = ('passwords_needed_to_start', 'retry_counts', 'hosts', 'credential_passwords')
+        fields = ('passwords_needed_to_start', 'retry_counts', 'hosts', 'job_type', 'credential_passwords')
 
     def validate_credential_passwords(self, value):
         pnts = self.instance.passwords_needed_to_start

awx/api/views/__init__.py

@@ -3435,6 +3435,7 @@ def post(self, request, *args, **kwargs):
 
         copy_kwargs = {}
         retry_hosts = serializer.validated_data.get('hosts', None)
+        job_type = serializer.validated_data.get('job_type', None)
         if retry_hosts and retry_hosts != 'all':
             if obj.status in ACTIVE_STATES:
                 return Response(
@@ -3455,6 +3456,8 @@ def post(self, request, *args, **kwargs):
                 )
             copy_kwargs['limit'] = ','.join(retry_host_list)
 
+        if job_type:
+            copy_kwargs['job_type'] = job_type
         new_job = obj.copy_unified_job(**copy_kwargs)
         result = new_job.signal_start(**serializer.validated_data['credential_passwords'])
         if not result:

awx/main/tests/functional/api/test_job.py

@@ -210,6 +210,39 @@ def test_disallowed_http_update_methods(put, patch, post, inventory, project, ad
     patch(url=reverse('api:job_detail', kwargs={'pk': job.pk}), data={}, user=admin_user, expect=405)
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "job_type",
+    [
+        'run',
+        'check',
+    ],
+)
+def test_job_relaunch_with_job_type(post, inventory, project, machine_credential, admin_user, job_type):
+    # Create a job template
+    jt = JobTemplate.objects.create(name='testjt', inventory=inventory, project=project)
+
+    # Set initial job type
+    init_job_type = 'check' if job_type == 'run' else 'run'
+
+    # Create a job instance
+    job = jt.create_unified_job(_eager_fields={'job_type': init_job_type})
+
+    # Perform the POST request
+    url = reverse('api:job_relaunch', kwargs={'pk': job.pk})
+    r = post(url=url, data={'job_type': job_type}, user=admin_user, expect=201)
+
+    # Assert that the response status code is 201 (Created)
+    assert r.status_code == 201
+
+    # Retrieve the newly created job from the response
+    new_job_id = r.data.get('id')
+    new_job = Job.objects.get(id=new_job_id)
+
+    # Assert that the new job has the correct job type
+    assert new_job.job_type == job_type
+
+
 class TestControllerNode:
     @pytest.fixture
     def project_update(self, project):