[rearchitecture] Prepare to migrate fuzz_task to a utask. (#3197)

src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py

@@ -23,6 +23,7 @@
 def tworker_preprocess_no_io(utask_module, task_argument, job_type,
                              uworker_env):
   logs.log('Starting utask_preprocess: %s.' % utask_module)
+  set_uworker_env(uworker_env)
   uworker_input = utask_module.utask_preprocess(task_argument, job_type,
                                                 uworker_env)
   if not uworker_input:
@@ -58,6 +59,7 @@ def tworker_postprocess_no_io(utask_module, uworker_output, uworker_input):
   # Do this to simulate out-of-band tamper-proof storage of the input.
   uworker_input = uworker_io.deserialize_uworker_input(uworker_input)
   add_uworker_input_to_output(uworker_output, uworker_input)
+  set_uworker_env(uworker_input.uworker_env)
   utask_module.utask_postprocess(uworker_output)
 
 
@@ -66,6 +68,7 @@ def tworker_preprocess(utask_module, task_argument, job_type, uworker_env):
   signed download URL for the uworker's input and the (unsigned) download URL
   for its output."""
   logs.log('Starting utask_preprocess: %s.' % utask_module)
+  set_uworker_env(uworker_env)
   # Do preprocessing.
   uworker_input = utask_module.utask_preprocess(task_argument, job_type,
                                                 uworker_env)
@@ -129,4 +132,5 @@ def tworker_postprocess(utask_module, output_download_url,
   uworker_input = uworker_io.download_and_deserialize_uworker_input(
       input_download_url)
   add_uworker_input_to_output(uworker_output, uworker_input)
+  set_uworker_env(uworker_input.uworker_env)
   utask_module.utask_postprocess(uworker_output)

src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py

@@ -1889,16 +1889,29 @@ def run(self):
     del testcases_metadata
     utils.python_gc()
 
+  def preprocess(self):
+    """Handles preprocessing."""
+    # TODO(metzman): Finish this.
+
+  def postprocess(self):
+    """Handles postprocessing."""
+    # TODO(metzman): Finish this.
+
 
 def utask_main(uworker_input):
   """Runs the given fuzzer for one round."""
-  test_timeout = environment.get_value('TEST_TIMEOUT')
-  session = FuzzingSession(uworker_input.fuzzer_name, uworker_input.job_type,
-                           test_timeout)
+  session = _make_session(uworker_input.fuzzer_name, uworker_input.job_type)
   session.run()
 
 
+def _make_session(fuzzer_name, job_type):
+  test_timeout = environment.get_value('TEST_TIMEOUT')
+  return FuzzingSession(fuzzer_name, job_type, test_timeout)
+
+
 def utask_preprocess(fuzzer_name, job_type, uworker_env):
+  session = _make_session(fuzzer_name, job_type)
+  session.preprocess()
   return uworker_io.UworkerInput(
       job_type=job_type,
       fuzzer_name=fuzzer_name,
@@ -1907,4 +1920,7 @@ def utask_preprocess(fuzzer_name, job_type, uworker_env):
 
 
 def utask_postprocess(output):
-  del output
+  session = _make_session(output.uworker_input.fuzzer_name,
+                          output.uworker_input.job_type)
+  session.postprocess()
+  del session

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

@@ -81,6 +81,7 @@ def setUp(self):
     os.environ['BOT_NAME'] = 'bot_name'
     os.environ['TASK_LEASE_SECONDS'] = '60'
     os.environ['FAIL_WAIT'] = '60'
+    os.environ['TEST_TIMEOUT'] = '10'
     self.mock.utcnow.return_value = test_utils.CURRENT_TIME
 
   def test_run_command_fuzz(self):