[Swarming] Skip unsupported blame and impact tasks for Swarming jobs

src/clusterfuzz/_internal/base/tasks/__init__.py

@@ -1029,6 +1029,10 @@ def redo_testcase(testcase, tasks, user_email):
   progression = 'progression' in tasks
   impact = 'impact' in tasks
   blame = 'blame' in tasks
+  # TODO(crbug.com/582825425): Support blame and impact tasks for Swarming jobs.
+  if swarming.is_swarming_applicable(testcase.job_type):
+    impact = False
+    blame = False
 
   task_list = []
   testcase_id = testcase.key.id()
@@ -1072,6 +1076,9 @@ def redo_testcase(testcase, tasks, user_email):
     testcase.set_metadata('blame_pending', True, update_testcase=False)
     testcase.set_metadata('predator_result', None, update_testcase=False)
 
+  if not task_list:
+    return
+
   for key in metadata_keys_to_clear:
     testcase.delete_metadata(key, update_testcase=False)
 

src/clusterfuzz/_internal/bot/tasks/task_creation.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Common functions for task creation for test cases."""
+from clusterfuzz._internal import swarming
 from clusterfuzz._internal.base import bisection
 from clusterfuzz._internal.base import tasks
 from clusterfuzz._internal.base import utils
@@ -81,6 +82,10 @@ def create_blame_task_if_needed(testcase):
   if not testcase.is_chromium():
     return
 
+  # TODO(crbug.com/582825425): Support blame tasks for Swarming jobs.
+  if swarming.is_swarming_applicable(testcase.job_type):
+    return
+
   # We cannot run blame job for custom binaries since we don't have any context
   # on the crash revision and regression range.
   if build_manager.is_custom_binary():
@@ -123,6 +128,10 @@ def create_impact_task_if_needed(testcase):
   if testcase.project_name != 'chromium':
     return
 
+  # TODO(crbug.com/582825425): Support impact tasks for Swarming jobs.
+  if swarming.is_swarming_applicable(testcase.job_type):
+    return
+
   # We cannot run impact job for custom binaries since we don't have any
   # archived production builds for these.
   if build_manager.is_custom_binary():

src/clusterfuzz/_internal/cron/helpers/tasks_scheduler.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 """Task scheduler used to recreate recurring tasks."""
 
+from clusterfuzz._internal import swarming
 from clusterfuzz._internal.base import tasks
 from clusterfuzz._internal.base import utils
 from clusterfuzz._internal.datastore import data_types
@@ -26,12 +27,21 @@ def schedule(task):
     logs.warning('Not creating impact tasks outside of Chrome.')
     return
 
+  # TODO(crbug.com/582825425): Support blame and impact tasks for Swarming jobs.
+  skip_swarming = task in ('impact', 'blame') and swarming.is_swarming_enabled()
+  swarming_job_cache = {}
   testcase_ids = []
   for status in ['Processed', 'Duplicate']:
     for testcase in data_types.Testcase.query(
         ndb_utils.is_true(data_types.Testcase.open),
         ndb_utils.is_false(data_types.Testcase.one_time_crasher_flag),
         data_types.Testcase.status == status):
+      if skip_swarming:
+        if testcase.job_type not in swarming_job_cache:
+          swarming_job_cache[testcase.job_type] = swarming.is_swarming_job(
+              testcase.job_type)
+        if swarming_job_cache[testcase.job_type]:
+          continue
       testcase_id = testcase.key.id()
       try:
         tasks.add_task(

src/clusterfuzz/_internal/datastore/data_handler.py

@@ -28,6 +28,7 @@
 from google.cloud import ndb
 
 from clusterfuzz._internal import fuzzing
+from clusterfuzz._internal import swarming
 from clusterfuzz._internal.base import dates
 from clusterfuzz._internal.base import errors
 from clusterfuzz._internal.base import memoize
@@ -978,7 +979,10 @@ def critical_tasks_completed(testcase):
 
   # Only require impact if it is feasible to run it. Don't require it if not
   # (when regression fails).
-  impact_satisfied = testcase.is_impact_set_flag or testcase.regression == 'NA'
+  # TODO(crbug.com/582825425): Require impact for Swarming jobs once supported.
+  impact_satisfied = (
+      testcase.is_impact_set_flag or testcase.regression == 'NA' or
+      swarming.is_swarming_applicable(testcase.job_type))
 
   return bool(testcase.minimized_keys and testcase.regression and
               impact_satisfied and not testcase.analyze_pending)

src/clusterfuzz/_internal/tests/appengine/common/tasks_test.py

@@ -385,3 +385,56 @@ def test_not_progression(self):
     messages = self.client.pull_from_subscription(
         self.subscription, max_messages=1)
     self.assertEqual(0, len(messages))
+
+
+@test_utils.with_cloud_emulators('datastore', 'pubsub')
+class RedoTestcaseSwarmingTest(unittest.TestCase):
+  """Tests for redo_testcase on Swarming jobs."""
+
+  def setUp(self):
+    helpers.patch_environ(self)
+    helpers.patch(self, [
+        'clusterfuzz._internal.base.tasks.add_task',
+        'clusterfuzz._internal.base.utils.current_date_time',
+    ])
+    self.mock.current_date_time.return_value = '2026-01-01 00:00:00 UTC'
+    data_types.Job(
+        name='android_emu_swarming',
+        platform='ANDROID_EMULATOR',
+        environment_string='IS_SWARMING_JOB = True\n').put()
+    self.testcase = test_utils.create_generic_testcase()
+    self.testcase.job_type = 'android_emu_swarming'
+    self.testcase.queue = 'jobs-android-emulator'
+    self.testcase.is_impact_set_flag = True
+    self.testcase.set_metadata('potentially_flaky', True, update_testcase=False)
+    self.testcase.set_metadata('blame_pending', False, update_testcase=False)
+    self.testcase.put()
+
+  def test_redo_skips_impact_and_blame_when_swarming_enabled(self):
+    """Tests that redo_testcase skips impact and blame for Swarming jobs."""
+    data_types.FeatureFlag(id='swarming_remote_execution', enabled=True).put()
+
+    tasks.redo_testcase(self.testcase, ['minimize', 'impact', 'blame'],
+                        'user@example.com')
+
+    self.mock.add_task.assert_called_once_with(
+        'minimize',
+        self.testcase.key.id(),
+        'android_emu_swarming',
+        queue='jobs-android-emulator',
+        wait_time=0)
+    testcase = self.testcase.key.get()
+    self.assertTrue(testcase.is_impact_set_flag)
+    self.assertFalse(testcase.get_metadata('blame_pending'))
+
+  def test_redo_only_unsupported_tasks_is_noop(self):
+    """Tests that redoing only impact/blame on a Swarming job makes no changes."""
+    data_types.FeatureFlag(id='swarming_remote_execution', enabled=True).put()
+    original_comments = self.testcase.comments
+
+    tasks.redo_testcase(self.testcase, ['impact', 'blame'], 'user@example.com')
+
+    self.mock.add_task.assert_not_called()
+    testcase = self.testcase.key.get()
+    self.assertTrue(testcase.get_metadata('potentially_flaky'))
+    self.assertEqual(testcase.comments, original_comments)

src/clusterfuzz/_internal/tests/appengine/handlers/cron/recurring_tasks_test.py

@@ -121,3 +121,50 @@ def test_nonchromium(self):
     self.mock.is_chromium.return_value = False
     schedule_impact_tasks.main()
     self.mock.add_task.assert_not_called()
+
+  def _add_swarming_testcase(self, flag_enabled):
+    """Adds an open Swarming testcase and sets the Swarming flag."""
+    data_types.Job(
+        name='job_swarming',
+        environment_string='IS_SWARMING_JOB = True\n',
+        platform='ANDROID_EMULATOR').put()
+    testcase = data_types.Testcase(
+        open=True,
+        one_time_crasher_flag=False,
+        status='Processed',
+        job_type='job_swarming',
+        queue='jobs-android-emulator')
+    testcase.put()
+    data_types.FeatureFlag(
+        id='swarming_remote_execution', enabled=flag_enabled).put()
+    return testcase
+
+  def test_swarming_jobs_skipped_for_impact(self):
+    """Tests that Swarming jobs are skipped when scheduling impact tasks."""
+    self.mock.is_chromium.return_value = True
+    self._add_swarming_testcase(flag_enabled=True)
+
+    schedule_impact_tasks.main()
+    self.mock.add_task.assert_has_calls([
+        mock.call('impact', 1, 'job', queue='jobs-linux'),
+        mock.call('impact', 5, 'job_windows', queue='jobs-windows'),
+    ])
+    self.assertEqual(self.mock.add_task.call_count, 2)
+
+  def test_swarming_jobs_scheduled_for_impact_when_swarming_disabled(self):
+    """Tests that Swarming jobs still get impact tasks when
+    SWARMING_REMOTE_EXECUTION is disabled."""
+    self.mock.is_chromium.return_value = True
+    swarming_testcase = self._add_swarming_testcase(flag_enabled=False)
+
+    schedule_impact_tasks.main()
+    self.mock.add_task.assert_has_calls([
+        mock.call('impact', 1, 'job', queue='jobs-linux'),
+        mock.call('impact', 5, 'job_windows', queue='jobs-windows'),
+        mock.call(
+            'impact',
+            swarming_testcase.key.id(),
+            'job_swarming',
+            queue='jobs-android-emulator'),
+    ])
+    self.assertEqual(self.mock.add_task.call_count, 3)

src/clusterfuzz/_internal/tests/core/bot/tasks/task_creation_test.py

@@ -16,6 +16,7 @@
 import unittest
 
 from clusterfuzz._internal.bot.tasks import task_creation
+from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.tests.test_libs import helpers as test_helpers
 from clusterfuzz._internal.tests.test_libs import test_utils
 
@@ -53,3 +54,89 @@ def test_create_tasks_no_skip_minimization(self):
     self.assertNotEqual(testcase.minimized_keys, 'NA')
     self.assertTrue(self.mock.create_minimize_task_if_needed.called)
     self.assertFalse(self.mock.create_postminimize_tasks.called)
+
+
+@test_utils.with_cloud_emulators('datastore')
+class SwarmingTaskCreationTest(unittest.TestCase):
+  """Tests for blame, impact, and variant task creation on Swarming jobs."""
+
+  def setUp(self):
+    test_helpers.patch_environ(self)
+    test_helpers.patch(self, [
+        'clusterfuzz._internal.base.tasks.add_task',
+        'clusterfuzz._internal.base.tasks.task_utils.is_remotely_executing_utasks',
+        'clusterfuzz._internal.base.utils.is_chromium',
+        'clusterfuzz._internal.build_management.build_manager.has_symbolized_builds',
+        'clusterfuzz._internal.build_management.build_manager.is_custom_binary',
+    ])
+    self.mock.is_chromium.return_value = True
+    self.mock.is_custom_binary.return_value = False
+    self.mock.has_symbolized_builds.return_value = False
+    self.mock.is_remotely_executing_utasks.return_value = True
+
+    data_types.Job(
+        name='linux_asan_chrome',
+        platform='LINUX',
+        project='chromium',
+        environment_string='PROJECT_NAME = chromium\n').put()
+    data_types.Job(
+        name='android_emu_swarming',
+        platform='ANDROID_EMULATOR',
+        project='chromium',
+        environment_string=(
+            'PROJECT_NAME = chromium\nIS_SWARMING_JOB = True\n')).put()
+
+    self.testcase = test_utils.create_generic_testcase()
+    self.testcase.job_type = 'android_emu_swarming'
+    self.testcase.project_name = 'chromium'
+    self.testcase.regression = '100:200'
+    self.testcase.trusted = False
+    self.testcase.put()
+
+  def _set_swarming_flag(self, enabled):
+    data_types.FeatureFlag(
+        id='swarming_remote_execution', enabled=enabled).put()
+
+  def test_skips_blame_and_impact_when_swarming_enabled(self):
+    """Tests that blame and impact tasks are skipped for Swarming jobs when
+    SWARMING_REMOTE_EXECUTION is enabled."""
+    self._set_swarming_flag(True)
+
+    task_creation.create_blame_task_if_needed(self.testcase)
+    task_creation.create_impact_task_if_needed(self.testcase)
+    self.mock.add_task.assert_not_called()
+
+  def test_creates_blame_and_impact_when_swarming_disabled(self):
+    """Tests that blame and impact tasks are still created for Swarming jobs
+    when SWARMING_REMOTE_EXECUTION is disabled."""
+    self._set_swarming_flag(False)
+
+    task_creation.create_blame_task_if_needed(self.testcase)
+    task_creation.create_impact_task_if_needed(self.testcase)
+    self.assertEqual([call[0][0] for call in self.mock.add_task.call_args_list],
+                     ['blame', 'impact'])
+
+  def test_skips_variant_for_untrusted_testcase_on_swarming_job(self):
+    """Tests that untrusted testcases don't get variant tasks on Swarming jobs,
+    since Swarming can't run untrusted workloads."""
+    linux_testcase = test_utils.create_generic_testcase()
+    linux_testcase.job_type = 'linux_asan_chrome'
+    linux_testcase.project_name = 'chromium'
+    linux_testcase.trusted = False
+    linux_testcase.put()
+
+    task_creation.create_variant_tasks_if_needed(linux_testcase)
+    self.mock.add_task.assert_not_called()
+
+  def test_creates_variant_for_trusted_testcase_on_swarming_job(self):
+    """Tests that trusted testcases still get variant tasks on Swarming jobs."""
+    linux_testcase = test_utils.create_generic_testcase()
+    linux_testcase.job_type = 'linux_asan_chrome'
+    linux_testcase.project_name = 'chromium'
+    linux_testcase.trusted = True
+    linux_testcase.put()
+
+    task_creation.create_variant_tasks_if_needed(linux_testcase)
+    self.mock.add_task.assert_called_once_with(
+        'variant', linux_testcase.key.id(), 'android_emu_swarming',
+        'jobs-android-emulator')

src/clusterfuzz/_internal/tests/core/datastore/data_handler_test.py

@@ -622,6 +622,29 @@ def test_critical_tasks_pending_chrome_regression_fail(self):
     testcase.regression = 'nots NA'
     self.assertFalse(data_handler.critical_tasks_completed(testcase))
 
+  def test_critical_tasks_completed_swarming_job_skips_impact(self):
+    """Tests that in Chrome, impact only blocks filing for Swarming jobs while
+    SWARMING_REMOTE_EXECUTION is disabled, since impact tasks are not run for
+    Swarming jobs once it is enabled."""
+    self.mock.is_chromium.return_value = True
+    data_types.Job(
+        name='swarming_job',
+        platform='ANDROID_EMULATOR',
+        environment_string='IS_SWARMING_JOB = True\n').put()
+    testcase = data_types.Testcase(job_type='swarming_job')
+    testcase.minimized_keys = 'fake'
+    testcase.regression = '100:200'
+    testcase.is_impact_set_flag = False
+    testcase.analyze_pending = False
+
+    flag = data_types.FeatureFlag(id='swarming_remote_execution', enabled=False)
+    flag.put()
+    self.assertFalse(data_handler.critical_tasks_completed(testcase))
+
+    flag.enabled = True
+    flag.put()
+    self.assertTrue(data_handler.critical_tasks_completed(testcase))
+
 
 class FilterStackTraceTest(fake_filesystem_unittest.TestCase):
   """Tests for data_handler.filter_stacktrace."""