Refactor reproduce (#4988)

This PR is a follow-up to #4976, addressing the feedback and discussions
from the original review. The primary focus is on refactoring and code
quality improvements.

*(Note: This PR involves a significant amount of code being moved to
more appropriate modules.)*

---
Quick summary:
  - Moved local setup functions into the `setup.py` module 
  - Updated imports to be module-level, adhering to style guidelines.
- Added a `--local-logging` flag to `butler` to enable convenient local
log output during development.
  - Reduced the reliance on mocks in tests
  - Implemented logic to correctly handle `minimized_keys` during setup.
  - Improved error handling to be more Pythonic.
- Removed unnecessary type hints and enhanced overall test reliability.

Author: PauloVLB

butler.py

@@ -97,11 +97,13 @@ def _setup_args_for_remote(parser):
 
   subparsers.add_parser('reboot', help='Reboot with `sudo reboot`.')
 
+
 def _add_integration_tests_subparsers(toplevel_subparsers):
   """Adds a parser for the `integration_tests` command."""
   toplevel_subparsers.add_parser(
       'integration_tests', help='Run end-to-end integration tests.')
-  
+
+
 def _add_weights_fuzzer_subparser(weights_subparsers):
   """Adds a parser for the `weights fuzzer` command."""
   parser = weights_subparsers.add_parser(
@@ -236,19 +238,29 @@ def _add_weights_subparser(toplevel_subparsers):
   _add_weights_batches_subparser(subparsers)
   _add_weights_target_subparser(subparsers)
 
+
 def _add_reproduce_subparser(toplevel_subparsers):
   """Adds a parser for the `reproduce` command."""
   parser = toplevel_subparsers.add_parser(
       'reproduce', help='Reproduce a testcase locally.')
   parser.add_argument(
       '-c', '--config-dir', required=True, help='Path to application config.')
   parser.add_argument(
-      '-t', '--testcase-id', required=True, help='The testcase ID to reproduce.')
+      '-t',
+      '--testcase-id',
+      required=True,
+      help='The testcase ID to reproduce.')
+
 
 def main():
   """Parse the command-line args and invoke the right command."""
   parser = _ArgumentParser(
       description='Butler is here to help you with command-line tasks.')
+  parser.add_argument(
+      '--local-logging',
+      action='store_true',
+      default=False,
+      help='Force logs to be local-only.')
   subparsers = parser.add_subparsers(dest='command')
 
   subparsers.add_parser(
@@ -424,12 +436,12 @@ def main():
     parser.print_help()
     return 0
 
-  _setup()
+  _setup(args)
   command = importlib.import_module(f'local.butler.{args.command}')
   return command.execute(args)
 
 
-def _setup():
+def _setup(args):
   """Set up configs and import paths."""
   os.environ['ROOT_DIR'] = os.path.abspath('.')
   os.environ['PYTHONIOENCODING'] = 'UTF-8'
@@ -438,6 +450,10 @@ def _setup():
   from clusterfuzz._internal.base import modules
   modules.fix_module_search_paths()
 
+  if args.local_logging:
+    from clusterfuzz._internal.system import environment
+    environment.set_local_log_only()
+
 
 if __name__ == '__main__':
   sys.exit(main())

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

@@ -49,6 +49,7 @@
 _DATA_BUNDLE_SYNC_INTERVAL_IN_SECONDS = 6 * 60 * 60
 _SYNC_FILENAME = '.sync'
 _TESTCASE_ARCHIVE_EXTENSION = '.zip'
+_EXECUTABLE_PERMISSIONS = 0o750
 
 
 def _set_timeout_value_from_user_upload(testcase_id, uworker_env):
@@ -352,10 +353,12 @@ def _get_testcase_key_and_archive_status(testcase):
 
 
 def _is_testcase_minimized(testcase):
+  """Returns True if the testcase is minimized."""
   return testcase.minimized_keys and testcase.minimized_keys != 'NA'
 
 
 def download_testcase(testcase_download_url, dst):
+  """Downloads a testcase from a signed url"""
   return storage.download_signed_url_to_file(testcase_download_url, dst)
 
 
@@ -651,7 +654,7 @@ def _update_fuzzer(
     return False
 
   # Make fuzzer executable.
-  os.chmod(fuzzer_path, 0o750)
+  os.chmod(fuzzer_path, _EXECUTABLE_PERMISSIONS)
 
   # Cleanup unneeded archive.
   shell.remove_file(archive_path)
@@ -890,3 +893,119 @@ def archive_testcase_and_dependencies_in_gcs(resource_list, testcase_path: str,
     shell.remove_file(zip_path)
 
   return archived, absolute_filename, zip_filename
+
+
+def setup_local_testcase(testcase: data_types.Testcase) -> str | None:
+  """Sets up the testcase file locally.
+
+  Args:
+    testcase: The Testcase object.
+
+  Returns:
+    - str | None : The local file path to the testcase, or None on failure.
+  """
+  try:
+    shell.clear_testcase_directories()
+  except Exception as e:
+    logs.error(f'Error clearing testcase directories: {e}')
+    return None
+
+  try:
+    _, testcase_file_path = _get_testcase_file_and_path(testcase)
+    if _is_testcase_minimized(testcase):
+      blob_key = testcase.minimized_keys
+    else:
+      blob_key = testcase.fuzzed_keys
+    if not blobs.read_blob_to_disk(blob_key, testcase_file_path):
+      logs.error(f'Failed to download testcase from blobstore: {blob_key}')
+      # Returning None for path when download fails
+      return None
+    prepare_environment_for_testcase(testcase)
+  except Exception as e:
+    logs.error(f'Error setting up testcase locally: {e}')
+    return None
+
+  return testcase_file_path
+
+
+def setup_local_fuzzer(fuzzer_name: str) -> bool:
+  """Sets up the fuzzer binaries and environment.
+
+  Args:
+    fuzzer_name: The name of the fuzzer to set up.
+
+  Returns:
+    True if setup was successful, False otherwise.
+  """
+  fuzzer = data_types.Fuzzer.query(data_types.Fuzzer.name == fuzzer_name).get()
+  if not fuzzer:
+    logs.error(f'Fuzzer {fuzzer_name} not found.')
+    return False
+
+  if fuzzer.untrusted_content:
+    logs.warning('You are about to run an untrusted fuzzer locally. '
+                 'This can be dangerous.')
+
+  environment.set_value('UNTRUSTED_CONTENT', fuzzer.untrusted_content)
+
+  if fuzzer.data_bundle_name:
+    logs.info(
+        f'Fuzzer {fuzzer_name} uses data bundle: {fuzzer.data_bundle_name}')
+
+  if fuzzer.launcher_script:
+    logs.error('Fuzzers with launch script not supported yet.')
+    return False
+
+  if fuzzer.builtin:
+    logs.info(f'Fuzzer {fuzzer_name} is builtin, no setup required.')
+    return True
+
+  fuzzer_directory: str = get_fuzzer_directory(fuzzer.name)
+
+  try:
+    if not shell.remove_directory(fuzzer_directory, recreate=True):
+      logs.error(f'Failed to clear fuzzer directory: {fuzzer_directory}')
+      return False
+  except Exception as e:
+    logs.error(f'Error clearing fuzzer directory {fuzzer_directory}: {e}')
+    return False
+
+  archive_path = os.path.join(fuzzer_directory, fuzzer.filename)
+  if not blobs.read_blob_to_disk(fuzzer.blobstore_key, archive_path):
+    logs.error('Failed to download fuzzer archive from blobstore: '
+               f'{fuzzer.blobstore_key}')
+    return False
+
+  try:
+    with archive.open(archive_path) as reader:
+      reader.extract_all(fuzzer_directory)
+  except archive.ArchiveError as e:
+    logs.error(f'Failed to unpack fuzzer archive {fuzzer.filename}: {e}')
+    return False
+  except Exception as e:
+    logs.error(
+        f'Unexpected error unpacking fuzzer archive {fuzzer.filename}: {e}')
+    return False
+  finally:
+    if os.path.exists(archive_path):
+      try:
+        shell.remove_file(archive_path)
+      except Exception as e:
+        logs.warning(
+            f'Failed to remove temporary archive file {archive_path}: {e}')
+
+  fuzzer_path = os.path.join(fuzzer_directory, fuzzer.executable_path)
+  if not os.path.exists(fuzzer_path):
+    logs.error(
+        f'Fuzzer executable {fuzzer.executable_path} not found in archive. '
+        'Check fuzzer configuration.')
+    return False
+
+  try:
+    os.chmod(fuzzer_path, _EXECUTABLE_PERMISSIONS)
+  except OSError as e:
+    logs.error(
+        f'Failed to set permissions on fuzzer executable {fuzzer_path}: {e}')
+    return False
+
+  return True

src/clusterfuzz/_internal/system/environment.py

@@ -1001,6 +1001,16 @@ def set_bot_environment():
   return True
 
 
+def set_local_log_only():
+  """Force logs to be local-only."""
+
+  # We set this to an empty string because currently the logs 
+  # module does not correctly evaluate env vars
+  # (i.e., 'false' is evaluated to true).
+  set_value('LOG_TO_GCP', '')
+  set_value('LOG_TO_CONSOLE', 'True')
+
+
 def set_tsan_max_history_size():
   """Sets maximum history size for TSAN tool."""
   tsan_options = get_value('TSAN_OPTIONS')