fix e2e tests

Signed-off-by: Javan Lacerda <javanlacerda@google.com>

Author: javanlacerda

src/clusterfuzz/_internal/batch/service.py

@@ -375,10 +375,8 @@ def create_utask_main_job(self, module: str, job_type: str,
     batch_tasks = [
         remote_task_types.RemoteTask(command, job_type, input_download_url)
     ]
-    result = self.create_utask_main_jobs(batch_tasks)
-    if result is None:
-      return result
-    return result[0]
+    uncreated_tasks = self.create_utask_main_jobs(batch_tasks)
+    return uncreated_tasks
 
   def create_utask_main_jobs(self,
                              remote_tasks: List[remote_task_types.RemoteTask]):

src/clusterfuzz/_internal/tests/core/batch/batch_service_test.py

@@ -254,7 +254,7 @@ def test_create_uworker_main_batch_job(self):
           UUIDS[0], spec1, ['url1'])
       self.mock_batch_client_instance.create_job.assert_called_with(
           expected_create_request)
-      self.assertEqual(result, 'job')
+      self.assertEqual(result, [])
 
 
 @test_utils.with_cloud_emulators('datastore')

src/clusterfuzz/_internal/tests/core/bot/startup/run_bot_test.py

@@ -134,7 +134,7 @@ class ScheduleUtaskMainsTest(unittest.TestCase):
   def setUp(self):
     helpers.patch(self, [
         'clusterfuzz._internal.base.tasks.get_utask_mains',
-        'clusterfuzz._internal.remote_task.remote_task_gate.RemoteTaskGate.create_utask_main_jobs',
+        'clusterfuzz._internal.remote_task.remote_task_gate.RemoteTaskGate',
         'clusterfuzz._internal.base.tasks.bulk_add_tasks',
     ])
 
@@ -150,11 +150,12 @@ def test_schedule_tasks_requeue_uncreated(self):
     self.mock.get_utask_mains.return_value = [mock_task]
 
     # Simulate that the tasks were not created and returned back.
-    self.mock.create_utask_main_jobs.side_effect = lambda x, tasks: tasks
+    self.mock.RemoteTaskGate.return_value.create_utask_main_jobs.side_effect = lambda tasks: tasks
 
     run_bot.schedule_utask_mains()
 
-    self.mock.create_utask_main_jobs.assert_called_once()
+    self.mock.RemoteTaskGate.return_value.create_utask_main_jobs.assert_called_once(
+    )
 
     # Check that bulk_add_tasks was called with the tasks returned by create_utask_main_jobs
     # and the correct queue.
@@ -179,10 +180,11 @@ def test_schedule_tasks_success(self):
     mock_task.lease.return_value.__exit__.return_value = None
 
     self.mock.get_utask_mains.return_value = [mock_task]
-    self.mock.create_utask_main_jobs.return_value = []
+    self.mock.RemoteTaskGate.return_value.create_utask_main_jobs.return_value = []
 
     run_bot.schedule_utask_mains()
 
-    self.mock.create_utask_main_jobs.assert_called_once()
+    self.mock.RemoteTaskGate.return_value.create_utask_main_jobs.assert_called_once(
+    )
     self.mock.bulk_add_tasks.assert_called_once_with(
         [], queue=taskslib.PREPROCESS_QUEUE, eta_now=True)

src/clusterfuzz/_internal/tests/core/k8s/k8s_service_e2e_test.py

@@ -111,7 +111,28 @@ def _wait_for_job_and_delete(self, job_name):
     self.api_client.delete_namespaced_job(
         name=job_name,
         namespace='default',
-        body=k8s_client.V1DeleteOptions(propagation_policy='Foreground'))
+        body=k8s_client.V1DeleteOptions(propagation_policy='Background'))
+
+    # Wait for deletion.
+    for _ in range(120):
+      try:
+        self.api_client.read_namespaced_job(job_name, 'default')
+        time.sleep(1)
+      except k8s_client.rest.ApiException as e:
+        if e.status == 404:
+          return
+        raise
+
+    self.fail(f'Job {job_name} was not deleted.')
+
+  def _get_cf_jobs(self):
+    """Returns a list of ClusterFuzz jobs."""
+    jobs = self.api_client.list_namespaced_job('default')
+    return [
+        j.metadata.name
+        for j in jobs.items
+        if j.metadata.name.startswith('cf-job-')
+    ]
 
   def test_create_job(self):
     """Tests creating a job."""
@@ -143,9 +164,13 @@ def test_create_uworker_main_batch_job(self, mock_get_command_from_module,
         is_kata=False)
     mock_get_k8s_job_configs.return_value = {('fuzz', 'test-job'): config}
 
-    actual_job_name = self.kubernetes_client.create_utask_main_job(
+    result = self.kubernetes_client.create_utask_main_job(
         'module', 'test-job', 'url1')
-    self._wait_for_job_and_delete(actual_job_name)
+    self.assertEqual(result, [])
+
+    jobs = self._get_cf_jobs()
+    self.assertEqual(len(jobs), 1)
+    self._wait_for_job_and_delete(jobs[0])
 
   @mock.patch('clusterfuzz._internal.k8s.service._get_k8s_job_configs')
   @mock.patch(
@@ -181,10 +206,12 @@ def test_create_uworker_main_batch_jobs(self, mock_get_command_from_module,
         remote_task_types.RemoteTask('fuzz', 'test-job2', 'url2'),
     ]
 
-    actual_job_names = self.kubernetes_client.create_utask_main_jobs(tasks)
-    self.assertEqual(len(actual_job_names), 2)
+    result = self.kubernetes_client.create_utask_main_jobs(tasks)
+    self.assertEqual(result, [])
 
-    for job_name in actual_job_names:
+    jobs = self._get_cf_jobs()
+    self.assertEqual(len(jobs), 2)
+    for job_name in jobs:
       self._wait_for_job_and_delete(job_name)