# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for running fuzzers."""
import json
import os
import sys
import shutil
import tempfile
import unittest
from unittest import mock

import parameterized
from pyfakefs import fake_filesystem_unittest

import build_fuzzers
import fuzz_target
import run_fuzzers

# pylint: disable=wrong-import-position
INFRA_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(INFRA_DIR)

import helper
import test_helpers

# NOTE: This integration test relies on
# https://github.com/google/oss-fuzz/tree/master/projects/example project.
EXAMPLE_PROJECT = 'example'

# Location of files used for testing.
TEST_DATA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                              'test_data')

MEMORY_FUZZER_DIR = os.path.join(TEST_DATA_PATH, 'memory')
MEMORY_FUZZER = 'curl_fuzzer_memory'

UNDEFINED_FUZZER_DIR = os.path.join(TEST_DATA_PATH, 'undefined')
UNDEFINED_FUZZER = 'curl_fuzzer_undefined'

FUZZ_SECONDS = 10


class RunFuzzerIntegrationTestMixin:  # pylint: disable=too-few-public-methods,invalid-name
  """Mixin for integration test classes that runbuild_fuzzers on builds of a
  specific sanitizer."""
  # These must be defined by children.
  FUZZER_DIR = None
  FUZZER = None

  def setUp(self):
    """Patch the environ so that we can execute runner scripts."""
    test_helpers.patch_environ(self, runner=True)

  def _test_run_with_sanitizer(self, fuzzer_dir, sanitizer):
    """Calls run_fuzzers on fuzzer_dir and |sanitizer| and asserts
    the run succeeded and that no bug was found."""
    with test_helpers.temp_dir_copy(fuzzer_dir) as fuzzer_dir_copy:
      config = test_helpers.create_run_config(fuzz_seconds=FUZZ_SECONDS,
                                              workspace=fuzzer_dir_copy,
                                              oss_fuzz_project_name='curl',
                                              sanitizer=sanitizer)
      result = run_fuzzers.run_fuzzers(config)
    self.assertEqual(result, run_fuzzers.RunFuzzersResult.NO_BUG_FOUND)


@unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                 'INTEGRATION_TESTS=1 not set')
class RunMemoryFuzzerIntegrationTest(RunFuzzerIntegrationTestMixin,
                                     unittest.TestCase):
  """Integration test for build_fuzzers with an MSAN build."""
  FUZZER_DIR = MEMORY_FUZZER_DIR
  FUZZER = MEMORY_FUZZER

  def test_run_with_memory_sanitizer(self):
    """Tests run_fuzzers with a valid MSAN build."""
    self._test_run_with_sanitizer(self.FUZZER_DIR, 'memory')


@unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                 'INTEGRATION_TESTS=1 not set')
class RunUndefinedFuzzerIntegrationTest(RunFuzzerIntegrationTestMixin,
                                        unittest.TestCase):
  """Integration test for build_fuzzers with an UBSAN build."""
  FUZZER_DIR = UNDEFINED_FUZZER_DIR
  FUZZER = UNDEFINED_FUZZER

  def test_run_with_undefined_sanitizer(self):
    """Tests run_fuzzers with a valid UBSAN build."""
    self._test_run_with_sanitizer(self.FUZZER_DIR, 'undefined')


class BaseFuzzTargetRunnerTest(unittest.TestCase):
  """Tests BaseFuzzTargetRunner."""

  def _create_runner(self, **kwargs):  # pylint: disable=no-self-use
    defaults = {
        'fuzz_seconds': FUZZ_SECONDS,
        'oss_fuzz_project_name': EXAMPLE_PROJECT
    }
    for default_key, default_value in defaults.items():
      if default_key not in kwargs:
        kwargs[default_key] = default_value

    config = test_helpers.create_run_config(**kwargs)
    return run_fuzzers.BaseFuzzTargetRunner(config)

  def _test_initialize_fail(self, expected_error_args, **create_runner_kwargs):
    with mock.patch('logging.error') as mock_error:
      runner = self._create_runner(**create_runner_kwargs)
      self.assertFalse(runner.initialize())
      mock_error.assert_called_with(*expected_error_args)

  @parameterized.parameterized.expand([(0,), (None,), (-1,)])
  def test_initialize_invalid_fuzz_seconds(self, fuzz_seconds):
    """Tests initialize fails with an invalid fuzz seconds."""
    expected_error_args = ('Fuzz_seconds argument must be greater than 1, '
                           'but was: %s.', fuzz_seconds)
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.mkdir(out_path)
      with mock.patch('utils.get_fuzz_targets') as mock_get_fuzz_targets:
        mock_get_fuzz_targets.return_value = [
            os.path.join(out_path, 'fuzz_target')
        ]
        self._test_initialize_fail(expected_error_args,
                                   fuzz_seconds=fuzz_seconds,
                                   workspace=tmp_dir)

  def test_initialize_no_out_dir(self):
    """Tests initialize fails with no out dir."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      expected_error_args = ('Out directory: %s does not exist.', out_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  def test_initialize_nonempty_artifacts(self):
    """Tests initialize with a file artifacts path."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.mkdir(out_path)
      os.makedirs(os.path.join(tmp_dir, 'out'))
      artifacts_path = os.path.join(tmp_dir, 'out', 'artifacts')
      with open(artifacts_path, 'w') as artifacts_handle:
        artifacts_handle.write('fake')
      expected_error_args = (
          'Artifacts path: %s exists and is not an empty directory.',
          artifacts_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  def test_initialize_bad_artifacts(self):
    """Tests initialize with a non-empty artifacts path."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.mkdir(out_path)
      artifacts_path = os.path.join(tmp_dir, 'out', 'artifacts')
      os.makedirs(artifacts_path)
      artifact_path = os.path.join(artifacts_path, 'artifact')
      with open(artifact_path, 'w') as artifact_handle:
        artifact_handle.write('fake')
      expected_error_args = (
          'Artifacts path: %s exists and is not an empty directory.',
          artifacts_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('logging.error')
  def test_initialize_empty_artifacts(self, mock_log_error,
                                      mock_get_fuzz_targets):
    """Tests initialize with an empty artifacts dir."""
    mock_get_fuzz_targets.return_value = ['fuzz-target']
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.mkdir(out_path)
      artifacts_path = os.path.join(tmp_dir, 'out', 'artifacts')
      os.makedirs(artifacts_path)
      runner = self._create_runner(workspace=tmp_dir)
      self.assertTrue(runner.initialize())
      mock_log_error.assert_not_called()
      self.assertTrue(os.path.isdir(artifacts_path))

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('logging.error')
  def test_initialize_no_artifacts(self, mock_log_error, mock_get_fuzz_targets):
    """Tests initialize with no artifacts dir (the expected setting)."""
    mock_get_fuzz_targets.return_value = ['fuzz-target']
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.mkdir(out_path)
      runner = self._create_runner(workspace=tmp_dir)
      self.assertTrue(runner.initialize())
      mock_log_error.assert_not_called()
      self.assertTrue(os.path.isdir(os.path.join(tmp_dir, 'out', 'artifacts')))

  def test_initialize_no_fuzz_targets(self):
    """Tests initialize with no fuzz targets."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.makedirs(out_path)
      expected_error_args = ('No fuzz targets were found in out directory: %s.',
                             out_path)
      self._test_initialize_fail(expected_error_args, workspace=tmp_dir)

  def test_get_fuzz_target_artifact(self):
    """Tests that get_fuzz_target_artifact works as intended."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      runner = self._create_runner(workspace=tmp_dir)
      crashes_dir = 'crashes-dir'
      runner.crashes_dir = crashes_dir
      artifact_name = 'artifact-name'
      target = mock.MagicMock()
      target_name = 'target_name'
      target.target_name = target_name

      fuzz_target_artifact = runner.get_fuzz_target_artifact(
          target, artifact_name)
      expected_fuzz_target_artifact = os.path.join(
          tmp_dir, 'out', 'artifacts', 'target_name-address-artifact-name')

      self.assertEqual(fuzz_target_artifact, expected_fuzz_target_artifact)


class CiFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase):
  """Tests that CiFuzzTargetRunner works as intended."""

  def setUp(self):
    self.setUpPyfakefs()

  @mock.patch('utils.get_fuzz_targets')
  @mock.patch('run_fuzzers.CiFuzzTargetRunner.run_fuzz_target')
  @mock.patch('run_fuzzers.CiFuzzTargetRunner.create_fuzz_target_obj')
  def test_run_fuzz_targets_quits(self, mock_create_fuzz_target_obj,
                                  mock_run_fuzz_target, mock_get_fuzz_targets):
    """Tests that run_fuzz_targets quits on the first crash it finds."""
    workspace = 'workspace'
    out_path = os.path.join(workspace, 'build-out')
    self.fs.create_dir(out_path)
    config = test_helpers.create_run_config(
        fuzz_seconds=FUZZ_SECONDS,
        workspace=workspace,
        oss_fuzz_project_name=EXAMPLE_PROJECT)
    runner = run_fuzzers.CiFuzzTargetRunner(config)

    mock_get_fuzz_targets.return_value = ['target1', 'target2']
    runner.initialize()
    testcase = os.path.join(workspace, 'testcase')
    self.fs.create_file(testcase)
    stacktrace = 'stacktrace'
    corpus_dir = 'corpus'
    self.fs.create_dir(corpus_dir)
    mock_run_fuzz_target.return_value = fuzz_target.FuzzResult(
        testcase, stacktrace, corpus_dir)
    magic_mock = mock.MagicMock()
    magic_mock.target_name = 'target1'
    mock_create_fuzz_target_obj.return_value = magic_mock
    self.assertTrue(runner.run_fuzz_targets())
    self.assertIn('target1-address-testcase',
                  os.listdir(runner.workspace.artifacts))
    self.assertEqual(mock_run_fuzz_target.call_count, 1)


class BatchFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase):
  """Tests that BatchFuzzTargetRunnerTest works as intended."""
  WORKSPACE = 'workspace'
  STACKTRACE = 'stacktrace'
  CORPUS_DIR = 'corpus'

  def setUp(self):
    self.setUpPyfakefs()
    out_dir = os.path.join(self.WORKSPACE, 'build-out')
    self.fs.create_dir(out_dir)
    self.testcase1 = os.path.join(out_dir, 'testcase-aaa')
    self.fs.create_file(self.testcase1)
    self.testcase2 = os.path.join(out_dir, 'testcase-bbb')
    self.fs.create_file(self.testcase2)
    self.config = test_helpers.create_run_config(fuzz_seconds=FUZZ_SECONDS,
                                                 workspace=self.WORKSPACE,
                                                 is_github=True)

  @mock.patch('utils.get_fuzz_targets', return_value=['target1', 'target2'])
  @mock.patch('clusterfuzz_deployment.ClusterFuzzLite.upload_build',
              return_value=True)
  @mock.patch('run_fuzzers.BatchFuzzTargetRunner.run_fuzz_target')
  @mock.patch('run_fuzzers.BatchFuzzTargetRunner.create_fuzz_target_obj')
  def test_run_fuzz_targets_quits(self, mock_create_fuzz_target_obj,
                                  mock_run_fuzz_target, _, __):
    """Tests that run_fuzz_targets doesn't quit on the first crash it finds."""
    runner = run_fuzzers.BatchFuzzTargetRunner(self.config)
    runner.initialize()

    call_count = 0

    def mock_run_fuzz_target_impl(_):
      nonlocal call_count
      if call_count == 0:
        testcase = self.testcase1
      elif call_count == 1:
        testcase = self.testcase2
      assert call_count != 2
      call_count += 1
      if not os.path.exists(self.CORPUS_DIR):
        self.fs.create_dir(self.CORPUS_DIR)
      return fuzz_target.FuzzResult(testcase, self.STACKTRACE, self.CORPUS_DIR)

    mock_run_fuzz_target.side_effect = mock_run_fuzz_target_impl
    magic_mock = mock.MagicMock()
    magic_mock.target_name = 'target1'
    mock_create_fuzz_target_obj.return_value = magic_mock
    self.assertTrue(runner.run_fuzz_targets())
    self.assertEqual(mock_run_fuzz_target.call_count, 2)

  @mock.patch('run_fuzzers.BaseFuzzTargetRunner.run_fuzz_targets',
              return_value=False)
  @mock.patch('clusterfuzz_deployment.ClusterFuzzLite.upload_crashes')
  def test_run_fuzz_targets_upload_crashes_and_builds(self, mock_upload_crashes,
                                                      _):
    """Tests that run_fuzz_targets uploads crashes and builds correctly."""
    runner = run_fuzzers.BatchFuzzTargetRunner(self.config)
    # TODO(metzman): Don't rely on this failing gracefully.
    runner.initialize()

    self.assertFalse(runner.run_fuzz_targets())
    self.assertEqual(mock_upload_crashes.call_count, 1)


@unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                 'INTEGRATION_TESTS=1 not set')
class CoverageReportIntegrationTest(unittest.TestCase):
  """Integration tests for coverage reports."""
  SANITIZER = 'coverage'

  def setUp(self):
    test_helpers.patch_environ(self, runner=True)

  @mock.patch('filestore.github_actions._upload_artifact_with_upload_js')
  def test_coverage_report(self, _):
    """Tests generation of coverage reports end-to-end, from building to
    generation."""

    with test_helpers.docker_temp_dir() as temp_dir:
      shared = os.path.join(temp_dir, 'shared')
      os.mkdir(shared)
      copy_command = ('cp -r /opt/code_coverage /shared && '
                      'cp $(which llvm-profdata) /shared && '
                      'cp $(which llvm-cov) /shared')
      assert helper.docker_run([
          '-v', f'{shared}:/shared', 'gcr.io/oss-fuzz-base/base-runner', 'bash',
          '-c', copy_command
      ])

      os.environ['CODE_COVERAGE_SRC'] = os.path.join(shared, 'code_coverage')
      os.environ['PATH'] += os.pathsep + shared
      # Do coverage build.
      build_config = test_helpers.create_build_config(
          oss_fuzz_project_name=EXAMPLE_PROJECT,
          project_repo_name='oss-fuzz',
          workspace=temp_dir,
          commit_sha='0b95fe1039ed7c38fea1f97078316bfc1030c523',
          base_commit='da0746452433dc18bae699e355a9821285d863c8',
          sanitizer=self.SANITIZER,
          is_github=True,
          # Needed for test not to fail because of permissions issues.
          bad_build_check=False)
      self.assertTrue(build_fuzzers.build_fuzzers(build_config))

      # TODO(metzman): Get rid of this here and make 'compile' do this.
      chmod_command = ('chmod -R +r /out && '
                       'find /out -type d -exec chmod +x {} +')

      assert helper.docker_run([
          '-v', f'{os.path.join(temp_dir, "build-out")}:/out',
          'gcr.io/oss-fuzz-base/base-builder', 'bash', '-c', chmod_command
      ])

      # Generate report.
      run_config = test_helpers.create_run_config(fuzz_seconds=FUZZ_SECONDS,
                                                  workspace=temp_dir,
                                                  sanitizer=self.SANITIZER,
                                                  run_fuzzers_mode='coverage',
                                                  is_github=True)
      result = run_fuzzers.run_fuzzers(run_config)
      self.assertEqual(result, run_fuzzers.RunFuzzersResult.NO_BUG_FOUND)
      expected_summary_path = os.path.join(
          TEST_DATA_PATH, 'example_coverage_report_summary.json')
      with open(expected_summary_path) as file_handle:
        expected_summary = json.loads(file_handle.read())
        actual_summary_path = os.path.join(temp_dir, 'cifuzz-coverage',
                                           'report', 'linux', 'summary.json')
      with open(actual_summary_path) as file_handle:
        actual_summary = json.loads(file_handle.read())
      self.assertEqual(expected_summary, actual_summary)


@unittest.skipIf(not os.getenv('INTEGRATION_TESTS'),
                 'INTEGRATION_TESTS=1 not set')
class RunAddressFuzzersIntegrationTest(RunFuzzerIntegrationTestMixin,
                                       unittest.TestCase):
  """Integration tests for build_fuzzers with an ASAN build."""

  BUILD_DIR_NAME = 'cifuzz-latest-build'

  def test_new_bug_found(self):
    """Tests run_fuzzers with a valid ASAN build."""
    # Set the first return value to True, then the second to False to
    # emulate a bug existing in the current PR but not on the downloaded
    # OSS-Fuzz build.
    with mock.patch('fuzz_target.FuzzTarget.is_reproducible',
                    side_effect=[True, False]):
      with tempfile.TemporaryDirectory() as tmp_dir:
        workspace = os.path.join(tmp_dir, 'workspace')
        shutil.copytree(TEST_DATA_PATH, workspace)
        config = test_helpers.create_run_config(
            fuzz_seconds=FUZZ_SECONDS,
            workspace=workspace,
            oss_fuzz_project_name=EXAMPLE_PROJECT)
        result = run_fuzzers.run_fuzzers(config)
        self.assertEqual(result, run_fuzzers.RunFuzzersResult.BUG_FOUND)

  @mock.patch('fuzz_target.FuzzTarget.is_reproducible',
              side_effect=[True, True])
  def test_old_bug_found(self, _):
    """Tests run_fuzzers with a bug found in OSS-Fuzz before."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      workspace = os.path.join(tmp_dir, 'workspace')
      shutil.copytree(TEST_DATA_PATH, workspace)
      config = test_helpers.create_run_config(
          fuzz_seconds=FUZZ_SECONDS,
          workspace=workspace,
          oss_fuzz_project_name=EXAMPLE_PROJECT)
      result = run_fuzzers.run_fuzzers(config)
      self.assertEqual(result, run_fuzzers.RunFuzzersResult.NO_BUG_FOUND)

  def test_invalid_build(self):
    """Tests run_fuzzers with an invalid ASAN build."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      out_path = os.path.join(tmp_dir, 'build-out')
      os.mkdir(out_path)
      config = test_helpers.create_run_config(
          fuzz_seconds=FUZZ_SECONDS,
          workspace=tmp_dir,
          oss_fuzz_project_name=EXAMPLE_PROJECT)
      result = run_fuzzers.run_fuzzers(config)
    self.assertEqual(result, run_fuzzers.RunFuzzersResult.ERROR)


class GetFuzzTargetRunnerTest(unittest.TestCase):
  """Tests for get_fuzz_fuzz_target_runner."""

  @parameterized.parameterized.expand([
      ('batch', run_fuzzers.BatchFuzzTargetRunner),
      ('ci', run_fuzzers.CiFuzzTargetRunner),
      ('coverage', run_fuzzers.CoverageTargetRunner)
  ])
  def test_get_fuzz_target_runner(self, run_fuzzers_mode,
                                  fuzz_target_runner_cls):
    """Tests that get_fuzz_target_runner returns the correct runner based on the
    specified run_fuzzers_mode."""
    with tempfile.TemporaryDirectory() as tmp_dir:
      run_config = test_helpers.create_run_config(
          fuzz_seconds=FUZZ_SECONDS,
          workspace=tmp_dir,
          oss_fuzz_project_name='example',
          run_fuzzers_mode=run_fuzzers_mode)
      runner = run_fuzzers.get_fuzz_target_runner(run_config)
      self.assertTrue(isinstance(runner, fuzz_target_runner_cls))


if __name__ == '__main__':
  unittest.main()