diff options
-rw-r--r-- | infra/cifuzz/affected_fuzz_targets_test.py | 10 | ||||
-rw-r--r-- | infra/cifuzz/clusterfuzz_deployment_test.py | 74 | ||||
-rw-r--r-- | infra/cifuzz/coverage_test.py | 8 | ||||
-rw-r--r-- | infra/cifuzz/fuzz_target_test.py | 15 | ||||
-rw-r--r-- | infra/cifuzz/run_fuzzers_test.py | 48 | ||||
-rw-r--r-- | infra/cifuzz/stack_parser_test.py | 29 |
6 files changed, 89 insertions, 95 deletions
diff --git a/infra/cifuzz/affected_fuzz_targets_test.py b/infra/cifuzz/affected_fuzz_targets_test.py index d87b696bf..72e6d266c 100644 --- a/infra/cifuzz/affected_fuzz_targets_test.py +++ b/infra/cifuzz/affected_fuzz_targets_test.py @@ -57,11 +57,13 @@ class RemoveUnaffectedFuzzTargets(unittest.TestCase): # yapf: enable def test_remove_unaffected_fuzz_targets(self, side_effect, expected_dir_len): """Tests that remove_unaffected_fuzzers has the intended effect.""" + # We can't use fakefs in this test because this test executes + # utils.is_fuzz_target_local. This function relies on the executable bit + # being set, which doesn't work properly in fakefs. with tempfile.TemporaryDirectory() as tmp_dir, mock.patch( - 'coverage._get_fuzzer_stats_dir_url', return_value=1): - with mock.patch( - 'coverage.OssFuzzCoverageGetter.get_files_covered_by_target' - ) as mocked_get_files: + 'coverage.OssFuzzCoverageGetter.get_files_covered_by_target' + ) as mocked_get_files: + with mock.patch('coverage._get_fuzzer_stats_dir_url', return_value=1): mocked_get_files.side_effect = side_effect shutil.copy(self.TEST_FUZZER_1, tmp_dir) shutil.copy(self.TEST_FUZZER_2, tmp_dir) diff --git a/infra/cifuzz/clusterfuzz_deployment_test.py b/infra/cifuzz/clusterfuzz_deployment_test.py index a7c088126..06ff78476 100644 --- a/infra/cifuzz/clusterfuzz_deployment_test.py +++ b/infra/cifuzz/clusterfuzz_deployment_test.py @@ -14,7 +14,6 @@ """Tests for clusterfuzz_deployment.py""" import os -import tempfile import unittest from unittest import mock import urllib.error @@ -58,47 +57,39 @@ def _create_deployment(**kwargs): return clusterfuzz_deployment.get_clusterfuzz_deployment(config) -class OSSFuzzTest(unittest.TestCase): +class OSSFuzzTest(fake_filesystem_unittest.TestCase): """Tests OSSFuzz.""" - def test_download_corpus(self): + OUT_DIR = '/out' + + def setUp(self): + self.setUpPyfakefs() + self.deployment = _create_deployment() + + @mock.patch('clusterfuzz_deployment.download_and_unpack_zip', + return_value=True) + def test_download_corpus(self, mocked_download_and_unpack_zip): """Tests that we can download a corpus for a valid project.""" - deployment = _create_deployment() - with tempfile.TemporaryDirectory() as tmp_dir: - with mock.patch('clusterfuzz_deployment.download_and_unpack_zip', - return_value=False) as mocked_download_and_unpack_zip: - deployment.download_corpus(EXAMPLE_FUZZER, tmp_dir) - (url, out_dir), _ = mocked_download_and_unpack_zip.call_args - self.assertEqual( - url, 'https://storage.googleapis.com/example-backup.' - 'clusterfuzz-external.appspot.com/corpus/libFuzzer/' - 'example_crash_fuzzer/public.zip') - self.assertEqual(out_dir, - os.path.join(tmp_dir, 'cifuzz-corpus', EXAMPLE_FUZZER)) - - def test_download_fail(self): + result = self.deployment.download_corpus(EXAMPLE_FUZZER, self.OUT_DIR) + self.assertIsNotNone(result) + expected_corpus_dir = os.path.join(self.OUT_DIR, 'cifuzz-corpus', + EXAMPLE_FUZZER) + expected_url = ('https://storage.googleapis.com/example-backup.' + 'clusterfuzz-external.appspot.com/corpus/libFuzzer/' + 'example_crash_fuzzer/public.zip') + call_args, _ = mocked_download_and_unpack_zip.call_args + self.assertEqual(call_args, (expected_url, expected_corpus_dir)) + + @mock.patch('clusterfuzz_deployment.download_and_unpack_zip', + return_value=False) + def test_download_fail(self, _): """Tests that when downloading fails, None is returned.""" - deployment = _create_deployment() - with tempfile.TemporaryDirectory() as tmp_dir: - with mock.patch('clusterfuzz_deployment.download_and_unpack_zip', - return_value=False): - corpus_path = deployment.download_corpus(EXAMPLE_FUZZER, tmp_dir) - self.assertIsNone(corpus_path) - - def test_download_latest_build(self): - """Tests that the build directory is downloaded once and no more.""" - deployment = _create_deployment() - with tempfile.TemporaryDirectory() as tmp_dir: - latest_name = deployment.get_latest_build_name() - with mock.patch('clusterfuzz_deployment.OSSFuzz.get_latest_build_name', - return_value=latest_name): - latest_build_path = deployment.download_latest_build(tmp_dir) - self.assertNotEqual(len(os.listdir(latest_build_path)), 0) + corpus_path = self.deployment.download_corpus(EXAMPLE_FUZZER, self.OUT_DIR) + self.assertIsNone(corpus_path) def test_get_latest_build_name(self): """Tests that the latest build name can be retrieved from GCS.""" - deployment = _create_deployment() - latest_build_name = deployment.get_latest_build_name() + latest_build_name = self.deployment.get_latest_build_name() self.assertTrue(latest_build_name.endswith('.zip')) self.assertTrue('address' in latest_build_name) @@ -147,14 +138,13 @@ class DownloadAndUnpackZipTest(fake_filesystem_unittest.TestCase): def setUp(self): self.setUpPyfakefs() - def test_bad_zip_download(self): + @mock.patch('urllib.request.urlretrieve', return_value=True) + def test_bad_zip_download(self, _): """Tests download_and_unpack_zip returns none when a bad zip is passed.""" - with open('/url_tmp.zip', 'w') as file_handle: - file_handle.write('Test file.') - with mock.patch('urllib.request.urlretrieve', return_value=True): - self.assertFalse( - clusterfuzz_deployment.download_and_unpack_zip( - '/not/a/real/url', '/extract-directory')) + self.fs.create_file('/url_tmp.zip', contents='Test file.') + self.assertFalse( + clusterfuzz_deployment.download_and_unpack_zip('/not/a/real/url', + '/extract-directory')) if __name__ == '__main__': diff --git a/infra/cifuzz/coverage_test.py b/infra/cifuzz/coverage_test.py index e9e9a2ed6..57120f5f5 100644 --- a/infra/cifuzz/coverage_test.py +++ b/infra/cifuzz/coverage_test.py @@ -80,11 +80,11 @@ class GetTargetCoverageReportTest(unittest.TestCase): self.assertIsNone( self.coverage_getter.get_target_coverage_report(INVALID_TARGET)) - def test_invalid_project_json(self): + @mock.patch('coverage._get_latest_cov_report_info', return_value=None) + def test_invalid_project_json(self, _): """Tests an invalid project JSON results in None being returned.""" - with mock.patch('coverage._get_latest_cov_report_info', return_value=None): - coverage_getter = coverage.OssFuzzCoverageGetter(PROJECT_NAME, REPO_PATH) - self.assertIsNone(coverage_getter.get_target_coverage_report(FUZZ_TARGET)) + coverage_getter = coverage.OssFuzzCoverageGetter(PROJECT_NAME, REPO_PATH) + self.assertIsNone(coverage_getter.get_target_coverage_report(FUZZ_TARGET)) class GetFilesCoveredByTargetTest(unittest.TestCase): diff --git a/infra/cifuzz/fuzz_target_test.py b/infra/cifuzz/fuzz_target_test.py index 21fa3b437..8a506fa59 100644 --- a/infra/cifuzz/fuzz_target_test.py +++ b/infra/cifuzz/fuzz_target_test.py @@ -187,16 +187,14 @@ class IsCrashReportableTest(fake_filesystem_unittest.TestCase): self.testcase_path = '/testcase' self.fs.create_file(self.testcase_path, contents='') + @mock.patch('fuzz_target.FuzzTarget.is_reproducible', + side_effect=[True, False]) @mock.patch('logging.info') - def test_new_reproducible_crash(self, mocked_info): + def test_new_reproducible_crash(self, mocked_info, _): """Tests that a new reproducible crash returns True.""" - - with mock.patch('fuzz_target.FuzzTarget.is_reproducible', - side_effect=[True, False]): - with tempfile.TemporaryDirectory() as tmp_dir: - self.test_target.out_dir = tmp_dir - self.assertTrue(self.test_target.is_crash_reportable( - self.testcase_path)) + with tempfile.TemporaryDirectory() as tmp_dir: + self.test_target.out_dir = tmp_dir + self.assertTrue(self.test_target.is_crash_reportable(self.testcase_path)) mocked_info.assert_called_with( 'The crash is reproducible. The crash doesn\'t reproduce ' 'on old builds. This code change probably introduced the ' @@ -219,7 +217,6 @@ class IsCrashReportableTest(fake_filesystem_unittest.TestCase): """Tests that a nonreportable crash causes the method to return False.""" with mock.patch('fuzz_target.FuzzTarget.is_reproducible', side_effect=is_reproducible_retvals): - with mock.patch('clusterfuzz_deployment.OSSFuzz.download_latest_build', return_value=self.oss_fuzz_build_path): self.assertFalse( diff --git a/infra/cifuzz/run_fuzzers_test.py b/infra/cifuzz/run_fuzzers_test.py index 5038e00fc..c41bbe37a 100644 --- a/infra/cifuzz/run_fuzzers_test.py +++ b/infra/cifuzz/run_fuzzers_test.py @@ -237,10 +237,12 @@ class CiFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase): 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, mocked_create_fuzz_target_obj, - mocked_run_fuzz_target): + mocked_run_fuzz_target, + mocked_get_fuzz_targets): """Tests that run_fuzz_targets quits on the first crash it finds.""" workspace = 'workspace' out_path = os.path.join(workspace, 'out') @@ -250,9 +252,8 @@ class CiFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase): project_name=EXAMPLE_PROJECT) runner = run_fuzzers.CiFuzzTargetRunner(config) - with mock.patch('utils.get_fuzz_targets') as mocked_get_fuzz_targets: - mocked_get_fuzz_targets.return_value = ['target1', 'target2'] - runner.initialize() + mocked_get_fuzz_targets.return_value = ['target1', 'target2'] + runner.initialize() testcase = os.path.join(workspace, 'testcase') self.fs.create_file(testcase) stacktrace = b'stacktrace' @@ -271,10 +272,12 @@ class BatchFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase): def setUp(self): self.setUpPyfakefs() + @mock.patch('utils.get_fuzz_targets') @mock.patch('run_fuzzers.BatchFuzzTargetRunner.run_fuzz_target') @mock.patch('run_fuzzers.BatchFuzzTargetRunner.create_fuzz_target_obj') def test_run_fuzz_targets_quits(self, mocked_create_fuzz_target_obj, - mocked_run_fuzz_target): + mocked_run_fuzz_target, + mocked_get_fuzz_targets): """Tests that run_fuzz_targets quits on the first crash it finds.""" workspace = 'workspace' out_path = os.path.join(workspace, 'out') @@ -284,9 +287,8 @@ class BatchFuzzTargetRunnerTest(fake_filesystem_unittest.TestCase): project_name=EXAMPLE_PROJECT) runner = run_fuzzers.BatchFuzzTargetRunner(config) - with mock.patch('utils.get_fuzz_targets') as mocked_get_fuzz_targets: - mocked_get_fuzz_targets.return_value = ['target1', 'target2'] - runner.initialize() + mocked_get_fuzz_targets.return_value = ['target1', 'target2'] + runner.initialize() testcase1 = os.path.join(workspace, 'testcase1') testcase2 = os.path.join(workspace, 'testcase2') self.fs.create_file(testcase1) @@ -342,25 +344,25 @@ class RunAddressFuzzersIntegrationTest(RunFuzzerIntegrationTestMixin, @unittest.skipIf(not os.getenv('INTEGRATION_TESTS'), 'INTEGRATION_TESTS=1 not set') - def test_old_bug_found(self): + @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.""" config = _create_config(fuzz_seconds=FUZZ_SECONDS, workspace=TEST_FILES_PATH, project_name=EXAMPLE_PROJECT) - with mock.patch('fuzz_target.FuzzTarget.is_reproducible', - side_effect=[True, True]): - with tempfile.TemporaryDirectory() as tmp_dir: - workspace = os.path.join(tmp_dir, 'workspace') - shutil.copytree(TEST_FILES_PATH, workspace) - config = _create_config(fuzz_seconds=FUZZ_SECONDS, - workspace=TEST_FILES_PATH, - project_name=EXAMPLE_PROJECT) - run_success, bug_found = run_fuzzers.run_fuzzers(config) - build_dir = os.path.join(TEST_FILES_PATH, 'out', self.BUILD_DIR_NAME) - self.assertTrue(os.path.exists(build_dir)) - self.assertNotEqual(0, len(os.listdir(build_dir))) - self.assertTrue(run_success) - self.assertFalse(bug_found) + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = os.path.join(tmp_dir, 'workspace') + shutil.copytree(TEST_FILES_PATH, workspace) + config = _create_config(fuzz_seconds=FUZZ_SECONDS, + workspace=TEST_FILES_PATH, + project_name=EXAMPLE_PROJECT) + run_success, bug_found = run_fuzzers.run_fuzzers(config) + build_dir = os.path.join(TEST_FILES_PATH, 'out', self.BUILD_DIR_NAME) + self.assertTrue(os.path.exists(build_dir)) + self.assertNotEqual(0, len(os.listdir(build_dir))) + self.assertTrue(run_success) + self.assertFalse(bug_found) def test_invalid_build(self): """Tests run_fuzzers with an invalid ASAN build.""" diff --git a/infra/cifuzz/stack_parser_test.py b/infra/cifuzz/stack_parser_test.py index 0d3969bd3..9b05710fc 100644 --- a/infra/cifuzz/stack_parser_test.py +++ b/infra/cifuzz/stack_parser_test.py @@ -13,9 +13,10 @@ # limitations under the License. """Tests for stack_parser.""" import os -import tempfile import unittest +from pyfakefs import fake_filesystem_unittest + import stack_parser # NOTE: This integration test relies on @@ -27,37 +28,39 @@ TEST_FILES_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'test_files') -class ParseOutputTest(unittest.TestCase): +class ParseOutputTest(fake_filesystem_unittest.TestCase): """Tests parse_fuzzer_output.""" + def setUp(self): + self.setUpPyfakefs() + def test_parse_valid_output(self): """Checks that the parse fuzzer output can correctly parse output.""" # Read the fuzzer output from disk. fuzzer_output_path = os.path.join(TEST_FILES_PATH, 'example_crash_fuzzer_output.txt') + self.fs.add_real_file(fuzzer_output_path) with open(fuzzer_output_path, 'rb') as fuzzer_output_handle: fuzzer_output = fuzzer_output_handle.read() - with tempfile.TemporaryDirectory() as tmp_dir: - bug_summary_filename = 'bug-summary.txt' - bug_summary_path = os.path.join(tmp_dir, bug_summary_filename) - stack_parser.parse_fuzzer_output(fuzzer_output, bug_summary_path) - self.assertEqual(os.listdir(tmp_dir), [bug_summary_filename]) - with open(bug_summary_path) as bug_summary_handle: - bug_summary = bug_summary_handle.read() + bug_summary_path = '/bug-summary.txt' + stack_parser.parse_fuzzer_output(fuzzer_output, bug_summary_path) + with open(bug_summary_path) as bug_summary_handle: + bug_summary = bug_summary_handle.read() # Compare the bug to the expected one. expected_bug_summary_path = os.path.join(TEST_FILES_PATH, 'bug_summary_example.txt') + self.fs.add_real_file(expected_bug_summary_path) with open(expected_bug_summary_path) as expected_bug_summary_handle: expected_bug_summary = expected_bug_summary_handle.read() self.assertEqual(expected_bug_summary, bug_summary) def test_parse_invalid_output(self): """Checks that no files are created when an invalid input was given.""" - with tempfile.TemporaryDirectory() as tmp_dir: - artifact = os.path.join(tmp_dir, 'bug-summary.txt') - stack_parser.parse_fuzzer_output(b'not a valid output_string', artifact) - self.assertEqual(len(os.listdir(tmp_dir)), 0) + artifact_path = '/bug-summary.txt' + stack_parser.parse_fuzzer_output(b'not a valid output_string', + artifact_path) + self.assertFalse(os.path.exists(artifact_path)) if __name__ == '__main__': |