Fix CoverageMap and CoverageBot and add test cases (#319)

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)

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

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

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"}