diff --git a/common/src/buttercup/common/maps.py b/common/src/buttercup/common/maps.py index 7af89f6f..058584da 100644 --- a/common/src/buttercup/common/maps.py +++ b/common/src/buttercup/common/maps.py @@ -121,9 +121,10 @@ class CoverageMap: def set_function_coverage(self, function_coverage: FunctionCoverage) -> None: # function paths should be sorted and unique + function_paths_list = list(function_coverage.function_paths) key = [ function_coverage.function_name, - function_coverage.function_paths, + function_paths_list, ] key_str = dumps(key, json_options=CANONICAL_JSON_OPTIONS) self.mp.set(key_str, function_coverage) diff --git a/common/tests/test_coverage_map.py b/common/tests/test_coverage_map.py new file mode 100644 index 00000000..5f3a141a --- /dev/null +++ b/common/tests/test_coverage_map.py @@ -0,0 +1,98 @@ +import pytest +from redis import Redis +from buttercup.common.maps import CoverageMap +from buttercup.common.datastructures.msg_pb2 import FunctionCoverage + + +@pytest.fixture +def redis_client(): + res = Redis(host="localhost", port=6379, db=15) + yield res + res.flushdb() + + +@pytest.fixture +def coverage_map(redis_client): + return CoverageMap(redis_client, harness_name="test_harness", package_name="test_package", task_id="test_task_id") + + +def test_coverage_map_set_and_get(coverage_map): + # Create a FunctionCoverage instance + function_coverage = FunctionCoverage() + function_coverage.function_name = "test_function" + function_coverage.function_paths.extend(["path1", "path2"]) + function_coverage.total_lines = 100 + function_coverage.covered_lines = 75 + + # Set the function coverage + coverage_map.set_function_coverage(function_coverage) + + # Retrieve the function coverage + function_paths_list = list(function_coverage.function_paths) + retrieved_coverage = coverage_map.get_function_coverage(function_coverage.function_name, function_paths_list) + + # Verify the retrieved coverage matches the original + assert retrieved_coverage.function_name == function_coverage.function_name + assert retrieved_coverage.function_paths == function_coverage.function_paths + assert retrieved_coverage.total_lines == function_coverage.total_lines + assert retrieved_coverage.covered_lines == function_coverage.covered_lines + + +def test_coverage_map_iteration(coverage_map): + # Create a FunctionCoverage instance + function_coverage = FunctionCoverage() + function_coverage.function_name = "test_function" + function_coverage.function_paths.extend(["path1", "path2"]) + function_coverage.total_lines = 100 + function_coverage.covered_lines = 75 + + # Set the function coverage + coverage_map.set_function_coverage(function_coverage) + + # Get all function coverages + coverages = coverage_map.list_function_coverage() + + # Verify we got exactly one coverage entry + assert len(coverages) == 1 + + # Verify the retrieved coverage matches the original + retrieved_coverage = coverages[0] + assert retrieved_coverage.function_name == function_coverage.function_name + assert retrieved_coverage.function_paths == function_coverage.function_paths + assert retrieved_coverage.total_lines == function_coverage.total_lines + assert retrieved_coverage.covered_lines == function_coverage.covered_lines + + +def test_coverage_map_multiple_functions(coverage_map): + # Create multiple FunctionCoverage instances + function1 = FunctionCoverage() + function1.function_name = "function1" + function1.function_paths.extend(["path1"]) + function1.total_lines = 100 + function1.covered_lines = 80 + + function2 = FunctionCoverage() + function2.function_name = "function2" + function2.function_paths.extend(["path2"]) + function2.total_lines = 200 + function2.covered_lines = 180 + + # Set both function coverages + coverage_map.set_function_coverage(function1) + coverage_map.set_function_coverage(function2) + + # Get all function coverages + coverages = coverage_map.list_function_coverage() + + # Verify we got exactly two coverage entries + assert len(coverages) == 2 + + # Create a set of function names for easier verification + function_names = {coverage.function_name for coverage in coverages} + assert function_names == {"function1", "function2"} + + +def test_coverage_map_nonexistent_function(coverage_map): + # Try to get coverage for a nonexistent function + coverage = coverage_map.get_function_coverage("nonexistent_function", ["nonexistent_path"]) + assert coverage is None diff --git a/fuzzer/src/buttercup/fuzzing_infra/coverage_bot.py b/fuzzer/src/buttercup/fuzzing_infra/coverage_bot.py index e8546e53..10795039 100644 --- a/fuzzer/src/buttercup/fuzzing_infra/coverage_bot.py +++ b/fuzzer/src/buttercup/fuzzing_infra/coverage_bot.py @@ -71,9 +71,8 @@ class CoverageBot(TaskLoop): if not (function_coverage.total_lines > 0 and function_coverage.covered_lines > 0): return False - old_function_coverage = coverage_map.get_function_coverage( - function_coverage.function_name, function_coverage.function_paths - ) + function_paths_list = list(function_coverage.function_paths) + old_function_coverage = coverage_map.get_function_coverage(function_coverage.function_name, function_paths_list) if old_function_coverage is None: return True return function_coverage.covered_lines > old_function_coverage.covered_lines diff --git a/fuzzer/tests/test_coverage_bot.py b/fuzzer/tests/test_coverage_bot.py new file mode 100644 index 00000000..441e7201 --- /dev/null +++ b/fuzzer/tests/test_coverage_bot.py @@ -0,0 +1,142 @@ +import pytest +from redis import Redis +from buttercup.fuzzing_infra.coverage_bot import CoverageBot +from buttercup.fuzzing_infra.coverage_runner import CoveredFunction +from buttercup.common.maps import CoverageMap +from buttercup.common.datastructures.msg_pb2 import FunctionCoverage + + +@pytest.fixture +def redis_client(): + res = Redis(host="localhost", port=6379, db=13) + yield res + res.flushdb() + + +@pytest.fixture +def coverage_bot(redis_client): + return CoverageBot( + redis=redis_client, + timer_seconds=1, + wdir="/tmp", + python="python3", + allow_pull=True, + base_image_url="test_image", + llvm_cov_tool="llvm-cov", + ) + + +def test_should_update_function_coverage_zero_coverage(redis_client): + coverage_map = CoverageMap(redis_client, "test_harness", "test_package", "test_task") + function_coverage = FunctionCoverage() + function_coverage.function_name = "test_function" + function_coverage.total_lines = 0 + function_coverage.covered_lines = 0 + function_coverage.function_paths.extend(["path1"]) + + result = CoverageBot._should_update_function_coverage(coverage_map, function_coverage) + assert result is False + + +def test_should_update_function_coverage_new_function(redis_client): + coverage_map = CoverageMap(redis_client, "test_harness", "test_package", "test_task") + + function_coverage = FunctionCoverage() + function_coverage.function_name = "test_function" + function_coverage.total_lines = 100 + function_coverage.covered_lines = 50 + function_coverage.function_paths.extend(["path1"]) + + result = CoverageBot._should_update_function_coverage(coverage_map, function_coverage) + assert result is True + + +def test_should_update_function_coverage_better_coverage(redis_client): + coverage_map = CoverageMap(redis_client, "test_harness", "test_package", "test_task") + + # First set some initial coverage + old_coverage = FunctionCoverage() + old_coverage.function_name = "test_function" + old_coverage.total_lines = 100 + old_coverage.covered_lines = 50 + old_coverage.function_paths.extend(["path1"]) + coverage_map.set_function_coverage(old_coverage) + + # Now test with better coverage + function_coverage = FunctionCoverage() + function_coverage.function_name = "test_function" + function_coverage.total_lines = 100 + function_coverage.covered_lines = 75 + function_coverage.function_paths.extend(["path1"]) + + result = CoverageBot._should_update_function_coverage(coverage_map, function_coverage) + assert result is True + + +def test_should_update_function_coverage_worse_coverage(redis_client): + coverage_map = CoverageMap(redis_client, "test_harness", "test_package", "test_task") + + # First set some initial coverage + old_coverage = FunctionCoverage() + old_coverage.function_name = "test_function" + old_coverage.total_lines = 100 + old_coverage.covered_lines = 75 + old_coverage.function_paths.extend(["path1"]) + coverage_map.set_function_coverage(old_coverage) + + # Now test with worse coverage + function_coverage = FunctionCoverage() + function_coverage.function_name = "test_function" + function_coverage.total_lines = 100 + function_coverage.covered_lines = 50 + function_coverage.function_paths.extend(["path1"]) + + result = CoverageBot._should_update_function_coverage(coverage_map, function_coverage) + assert result is False + + +def test_submit_function_coverage(coverage_bot, redis_client): + # Create test data + func_coverage = [ + CoveredFunction(names="test_function", total_lines=100, covered_lines=75, function_paths=["path1", "path2"]) + ] + harness_name = "test_harness" + package_name = "test_package" + task_id = "test_task_id" + + # Create a real CoverageMap instance + coverage_map = CoverageMap(redis_client, harness_name, package_name, task_id) + + coverage_bot._submit_function_coverage(func_coverage, harness_name, package_name, task_id) + + # Verify the coverage was stored correctly + stored_coverage = coverage_map.get_function_coverage("test_function", ["path1", "path2"]) + assert stored_coverage is not None + assert stored_coverage.function_name == "test_function" + assert stored_coverage.total_lines == 100 + assert stored_coverage.covered_lines == 75 + assert list(stored_coverage.function_paths) == ["path1", "path2"] + + +def test_submit_function_coverage_multiple_functions(coverage_bot, redis_client): + # Create test data with multiple functions + func_coverage = [ + CoveredFunction(names="function1", total_lines=100, covered_lines=75, function_paths=["path1"]), + CoveredFunction(names="function2", total_lines=200, covered_lines=150, function_paths=["path2"]), + ] + harness_name = "test_harness" + package_name = "test_package" + task_id = "test_task_id" + + # Create a real CoverageMap instance + coverage_map = CoverageMap(redis_client, harness_name, package_name, task_id) + + coverage_bot._submit_function_coverage(func_coverage, harness_name, package_name, task_id) + + # Verify both functions were stored correctly + stored_coverages = coverage_map.list_function_coverage() + assert len(stored_coverages) == 2 + + # Create a set of function names for easier verification + function_names = {coverage.function_name for coverage in stored_coverages} + assert function_names == {"function1", "function2"}