<commit_before>"""Integrations tests for SiliFuzz orchestrator."""

import lzma
import os
import re
import subprocess
import time

import absl.logging
from absl.testing import absltest


def get_data_dependency(name: str) -> str:
  return os.path.join(
      absltest.get_default_test_srcdir(),
      name)


_ORCHESTRATOR_PATH = get_data_dependency(
    'silifuzz/orchestrator/silifuzz_orchestrator_main')

_RUNNER_PATH = get_data_dependency('silifuzz/orchestrator/test_runner')


class OrchestratorTest(absltest.TestCase):

  _FAKE_CORPUS = []

  @classmethod
  def setUpClass(cls):
    super(OrchestratorTest, cls).setUpClass()
    temp_dir = absltest.get_default_test_tmpdir()
    for x in ('One', 'Two'):
      path = os.path.join(temp_dir, f'fake_corpus_{x}.xz')
      with open(path, 'wb') as f:
        f.write(lzma.compress(f'Corpus {x}'.encode('ascii')))
      cls._FAKE_CORPUS.append(path)

  def _popen_args(
      self,
      test_dummy_commands: list[str],
      duration_sec: int,
      max_cpus: int,
      corpus_files: list[str],
      extra_args: list[str],
  ) -> dict[str, object]:
    args = [
        _ORCHESTRATOR_PATH,
        '--runner',
        _RUNNER_PATH,
        '--stderrthreshold=0',
        f'--duration={duration_sec}s',
        f'--max_cpus={max_cpus}',
    ] + (extra_args or []) + corpus_files + ['--'] + (
        test_dummy_commands or [])
    absl.logging.info(' '.join(args))
    pass_fds = []
    # Figure out the value of --binary_log_fd. It's either this or plumbing
    # Popen **kwargs all the way to assertOrchestratorExitCode()
    for (f, fn) in zip(args, args[1:]):
      if f == '--binary_log_fd':
        pass_fds.append(int(fn))
    return dict(args=args, stderr=subprocess.PIPE, pass_fds=pass_fds)

  def _run(self, *args, **kwargs):
    popen_args = self._popen_args(*args, **kwargs)
    proc = subprocess.Popen(**popen_args)
    # The timeout value is determined by the max polling interval in the
    # orchestrator which is currently 10s.
    _, stderr = proc.communicate(timeout=kwargs['duration_sec'] + 10)
    return stderr.decode('utf-8').split('\n'), proc.returncode

  def assertStrSeqContainsAll(self, seq: list[str], regexs: list[str]):
    """Assert that the regular expression matches any string in `seq`."""

    for r in regexs:
      for s in seq:
        if re.search(r, s):
          break
      else:
        self.fail(f'Did not find "{r}" in\n========\n' + '\n'.join(seq))

  def run_orchestrator(self,
                       test_dummy_commands: list[str],
                       duration_sec: int = 3,
                       max_cpus: int = 1,
                       multicorpus: bool = False,
                       extra_args: list[str] = None) -> (list[str], int):
    corpus_files = [self._FAKE_CORPUS[0]]
    if multicorpus:
      corpus_files.append(self._FAKE_CORPUS[1])
    (err_log, returncode) = self._run(
        duration_sec=duration_sec,
        max_cpus=max_cpus,
        corpus_files=corpus_files,
        extra_args=extra_args,
        test_dummy_commands=test_dummy_commands)
    return (err_log, returncode)

  def test_basic(self):
    (err_log,
     returncode) = self.run_orchestrator(['short_output', 'short_loop'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(
        err_log,
        ['T0 started', 'ShortOutput', 'T0.*exit_status: ok', 'T0 stopped'])

  def test_multicpu(self):
    (err_log, returncode) = self.run_orchestrator(['short_output'], max_cpus=3)
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'T0 started',
        'ShortOutput',
        'T0.*exit_status: ok',
        'T1.*exit_status: ok',
        'T2.*exit_status: ok',
    ])

  def test_exit7(self):
    (err_log, returncode) = self.run_orchestrator(['short_loop', 'exit7'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'T0.*exit_status: internal_error',
    ])

  def test_timeout(self):
    # If you change --timeout=2 to something else, also change test_runner.cc.
    (err_log,
     returncode) = self.run_orchestrator(['--timeout=2', 'infinite_loop'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'T0.*exit_status: internal_error',
    ])

  def test_sequential_mode(self):
    (err_log,
     returncode) = self.run_orchestrator([],
                                         extra_args=['--sequential_mode'],
                                         multicorpus=True)
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'TEST RUNNER sequential_mode',
        'T0 Reached end of stream in sequential mode',
    ])

  def test_multiple_corpora(self):
    # Check that the uncompressed contents of both fake corpora are present.
    (err_log,
     returncode) = self.run_orchestrator(['print_first_line'],
                                         extra_args=['--sequential_mode'],
                                         multicorpus=True)
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'Corpus One',
        'Corpus Two',
    ])

  def test_snap_failure(self):
    (err_log, returncode) = self.run_orchestrator(
        ['snap_fail'], extra_args=['--enable_v1_compat_logging'])
    self.assertEqual(returncode, 1)
    self.assertStrSeqContainsAll(err_log, [
        'snap_fail: my_snap', 'exit_status: snap_fail',
        'Silifuzz detected issue on CPU.*running snapshot my_snap'
    ])

  def test_duration(self):
    (err_log, returncode) = self.run_orchestrator(['sleep100'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'exit_status: internal_error',
        'Runner killed by signal 14',
    ])

  def test_binary_logging(self):
    (read_fd, write_fd) = os.pipe()
    (err_log, returncode) = self.run_orchestrator(
        ['snap_fail'],
        extra_args=['--sequential_mode', '--binary_log_fd',
                    str(write_fd)],
    )
    self.assertEqual(returncode, 1)
    self.assertStrSeqContainsAll(err_log, [
        'snap_fail: my_snap',
        'exit_status: snap_fail',
    ])
    os.close(write_fd)
    bin_log = os.read(read_fd, 4096)
    os.close(read_fd)
    # TODO(ksteuck): Inspect the contents.
    self.assertNotEmpty(bin_log)

  def test_rlimit_fsize(self):
    (err_log,
     returncode) = self.run_orchestrator(['long_output'],
                                         extra_args=['--sequential_mode'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'exit_status: internal_error',
    ])

  def test_sigint(self):
    popen_args = self._popen_args(
        test_dummy_commands=['sleep100'],
        max_cpus=1,
        corpus_files=[self._FAKE_CORPUS[0]],
        duration_sec=3600,
        extra_args=['--sequential_mode'])
    # start_new_session ensures that the orchestrator process is in its own
    # process group so that we can killpg() it later.
    proc = subprocess.Popen(start_new_session=True, **popen_args)
    time.sleep(5)
    # Send SIGINT to the orchestator's process group. This similates ^C
    # behavior.
    os.killpg(proc.pid, 2)
    # NOTE: the communicate() interface used in all other tests takes care of
    # buffering the stdout/stderr in memory. This test relies on the fact
    # that the child process output will fit into the default kernel buffer.
    proc_err = proc.stderr.read(4096)
    proc.wait(5)
    self.assertEqual(proc.returncode, 0)
    self.assertIn('SIGINT/SIGALRM caught', proc_err.decode('utf-8'))
<commit_msg>Fix fd leak in the test.<commit_after>"""Integrations tests for SiliFuzz orchestrator."""

import lzma
import os
import re
import subprocess
import time

import absl.logging
from absl.testing import absltest


def get_data_dependency(name: str) -> str:
  return os.path.join(
      absltest.get_default_test_srcdir(),
      name)


_ORCHESTRATOR_PATH = get_data_dependency(
    'silifuzz/orchestrator/silifuzz_orchestrator_main')

_RUNNER_PATH = get_data_dependency('silifuzz/orchestrator/test_runner')


class OrchestratorTest(absltest.TestCase):

  _FAKE_CORPUS = []

  @classmethod
  def setUpClass(cls):
    super(OrchestratorTest, cls).setUpClass()
    temp_dir = absltest.get_default_test_tmpdir()
    for x in ('One', 'Two'):
      path = os.path.join(temp_dir, f'fake_corpus_{x}.xz')
      with open(path, 'wb') as f:
        f.write(lzma.compress(f'Corpus {x}'.encode('ascii')))
      cls._FAKE_CORPUS.append(path)

  def _popen_args(
      self,
      test_dummy_commands: list[str],
      duration_sec: int,
      max_cpus: int,
      corpus_files: list[str],
      extra_args: list[str],
  ) -> dict[str, object]:
    args = [
        _ORCHESTRATOR_PATH,
        '--runner',
        _RUNNER_PATH,
        '--stderrthreshold=0',
        f'--duration={duration_sec}s',
        f'--max_cpus={max_cpus}',
    ] + (extra_args or []) + corpus_files + ['--'] + (
        test_dummy_commands or [])
    absl.logging.info(' '.join(args))
    pass_fds = []
    # Figure out the value of --binary_log_fd. It's either this or plumbing
    # Popen **kwargs all the way to assertOrchestratorExitCode()
    for (f, fn) in zip(args, args[1:]):
      if f == '--binary_log_fd':
        pass_fds.append(int(fn))
    return dict(args=args, stderr=subprocess.PIPE, pass_fds=pass_fds)

  def _run(self, *args, **kwargs):
    popen_args = self._popen_args(*args, **kwargs)
    proc = subprocess.Popen(**popen_args)
    # The timeout value is determined by the max polling interval in the
    # orchestrator which is currently 10s.
    _, stderr = proc.communicate(timeout=kwargs['duration_sec'] + 10)
    return stderr.decode('utf-8').split('\n'), proc.returncode

  def assertStrSeqContainsAll(self, seq: list[str], regexs: list[str]):
    """Assert that the regular expression matches any string in `seq`."""

    for r in regexs:
      for s in seq:
        if re.search(r, s):
          break
      else:
        self.fail(f'Did not find "{r}" in\n========\n' + '\n'.join(seq))

  def run_orchestrator(self,
                       test_dummy_commands: list[str],
                       duration_sec: int = 3,
                       max_cpus: int = 1,
                       multicorpus: bool = False,
                       extra_args: list[str] = None) -> (list[str], int):
    corpus_files = [self._FAKE_CORPUS[0]]
    if multicorpus:
      corpus_files.append(self._FAKE_CORPUS[1])
    (err_log, returncode) = self._run(
        duration_sec=duration_sec,
        max_cpus=max_cpus,
        corpus_files=corpus_files,
        extra_args=extra_args,
        test_dummy_commands=test_dummy_commands)
    return (err_log, returncode)

  def test_basic(self):
    (err_log,
     returncode) = self.run_orchestrator(['short_output', 'short_loop'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(
        err_log,
        ['T0 started', 'ShortOutput', 'T0.*exit_status: ok', 'T0 stopped'])

  def test_multicpu(self):
    (err_log, returncode) = self.run_orchestrator(['short_output'], max_cpus=3)
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'T0 started',
        'ShortOutput',
        'T0.*exit_status: ok',
        'T1.*exit_status: ok',
        'T2.*exit_status: ok',
    ])

  def test_exit7(self):
    (err_log, returncode) = self.run_orchestrator(['short_loop', 'exit7'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'T0.*exit_status: internal_error',
    ])

  def test_timeout(self):
    # If you change --timeout=2 to something else, also change test_runner.cc.
    (err_log,
     returncode) = self.run_orchestrator(['--timeout=2', 'infinite_loop'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'T0.*exit_status: internal_error',
    ])

  def test_sequential_mode(self):
    (err_log,
     returncode) = self.run_orchestrator([],
                                         extra_args=['--sequential_mode'],
                                         multicorpus=True)
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'TEST RUNNER sequential_mode',
        'T0 Reached end of stream in sequential mode',
    ])

  def test_multiple_corpora(self):
    # Check that the uncompressed contents of both fake corpora are present.
    (err_log,
     returncode) = self.run_orchestrator(['print_first_line'],
                                         extra_args=['--sequential_mode'],
                                         multicorpus=True)
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'Corpus One',
        'Corpus Two',
    ])

  def test_snap_failure(self):
    (err_log, returncode) = self.run_orchestrator(
        ['snap_fail'], extra_args=['--enable_v1_compat_logging'])
    self.assertEqual(returncode, 1)
    self.assertStrSeqContainsAll(err_log, [
        'snap_fail: my_snap', 'exit_status: snap_fail',
        'Silifuzz detected issue on CPU.*running snapshot my_snap'
    ])

  def test_duration(self):
    (err_log, returncode) = self.run_orchestrator(['sleep100'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'exit_status: internal_error',
        'Runner killed by signal 14',
    ])

  def test_binary_logging(self):
    (read_fd, write_fd) = os.pipe()
    (err_log, returncode) = self.run_orchestrator(
        ['snap_fail'],
        extra_args=['--sequential_mode', '--binary_log_fd',
                    str(write_fd)],
    )
    self.assertEqual(returncode, 1)
    self.assertStrSeqContainsAll(err_log, [
        'snap_fail: my_snap',
        'exit_status: snap_fail',
    ])
    os.close(write_fd)
    bin_log = os.read(read_fd, 4096)
    os.close(read_fd)
    # TODO(ksteuck): Inspect the contents.
    self.assertNotEmpty(bin_log)

  def test_rlimit_fsize(self):
    (err_log,
     returncode) = self.run_orchestrator(['long_output'],
                                         extra_args=['--sequential_mode'])
    self.assertEqual(returncode, 0)
    self.assertStrSeqContainsAll(err_log, [
        'exit_status: internal_error',
    ])

  def test_sigint(self):
    popen_args = self._popen_args(
        test_dummy_commands=['sleep100'],
        max_cpus=1,
        corpus_files=[self._FAKE_CORPUS[0]],
        duration_sec=3600,
        extra_args=['--sequential_mode'])
    # start_new_session ensures that the orchestrator process is in its own
    # process group so that we can killpg() it later.
    with subprocess.Popen(start_new_session=True, **popen_args) as proc:
      time.sleep(5)
      # Send SIGINT to the orchestator's process group. This similates ^C
      # behavior.
      os.killpg(proc.pid, 2)
      # NOTE: the communicate() interface used in all other tests takes care of
      # buffering the stdout/stderr in memory. This test relies on the fact
      # that the child process output will fit into the default kernel buffer.
      proc_err = proc.stderr.read(4096)
      self.assertIn('SIGINT/SIGALRM caught', proc_err.decode('utf-8'))
    self.assertEqual(proc.returncode, 0)

{'fraction_non_alphanumeric': 0.12136785926072347, 'fraction_numerical': 0.008367925149897872, 'mean_word_length': 2.84253164556962, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4492092', 'n_tokens_mistral': 5110, 'n_tokens_neox': 4973, 'n_words': 1006}