first pass at hypothesis integration

Author: KRRT7

codeflash/code_utils/code_utils.py

@@ -254,6 +254,11 @@ def module_name_from_file_path(file_path: Path, project_root_path: Path, *, trav
         raise ValueError(msg)  # noqa: B904
 
 
+def get_qualified_function_path(file_path: Path, project_root_path: Path, qualified_name: str) -> str:
+    module_path = file_path.relative_to(project_root_path).with_suffix("").as_posix().replace("/", ".")
+    return f"{module_path}.{qualified_name}"
+
+
 def file_path_from_module_name(module_name: str, project_root_path: Path) -> Path:
     """Get file path from module path."""
     return project_root_path / (module_name.replace(".", os.sep) + ".py")

codeflash/discovery/discover_unit_tests.py

@@ -497,6 +497,8 @@ def discover_tests_pytest(
             test_type = TestType.REPLAY_TEST
         elif "test_concolic_coverage" in test["test_file"]:
             test_type = TestType.CONCOLIC_COVERAGE_TEST
+        elif "test_hypothesis" in test["test_file"]:
+            test_type = TestType.HYPOTHESIS_TEST
         else:
             test_type = TestType.EXISTING_UNIT_TEST
 
@@ -540,6 +542,8 @@ def get_test_details(_test: unittest.TestCase) -> TestsInFile | None:
             test_type = TestType.REPLAY_TEST
         elif "test_concolic_coverage" in str(_test_module_path):
             test_type = TestType.CONCOLIC_COVERAGE_TEST
+        elif "test_hypothesis" in str(_test_module_path):
+            test_type = TestType.HYPOTHESIS_TEST
         else:
             test_type = TestType.EXISTING_UNIT_TEST
         return TestsInFile(

codeflash/models/test_type.py

@@ -8,6 +8,7 @@ class TestType(Enum):
     REPLAY_TEST = 4
     CONCOLIC_COVERAGE_TEST = 5
     INIT_STATE_TEST = 6
+    HYPOTHESIS_TEST = 7
 
     def to_name(self) -> str:
         if self is TestType.INIT_STATE_TEST:
@@ -18,5 +19,6 @@ def to_name(self) -> str:
             TestType.GENERATED_REGRESSION: "🌀 Generated Regression Tests",
             TestType.REPLAY_TEST: "⏪ Replay Tests",
             TestType.CONCOLIC_COVERAGE_TEST: "🔎 Concolic Coverage Tests",
+            TestType.HYPOTHESIS_TEST: "🔮 Hypothesis Tests",
         }
         return names[self]

codeflash/optimization/function_optimizer.py

@@ -95,6 +95,7 @@
 from codeflash.telemetry.posthog_cf import ph
 from codeflash.verification.concolic_testing import generate_concolic_tests
 from codeflash.verification.equivalence import compare_test_results
+from codeflash.verification.hypothesis_testing import generate_hypothesis_tests
 from codeflash.verification.instrument_codeflash_capture import instrument_codeflash_capture
 from codeflash.verification.parse_line_profile_test_output import parse_line_profile_results
 from codeflash.verification.parse_test_output import calculate_function_throughput_from_test_results, parse_test_results
@@ -281,6 +282,8 @@ def generate_and_instrument_tests(
             GeneratedTestsList,
             dict[str, set[FunctionCalledInTest]],
             str,
+            dict[str, set[FunctionCalledInTest]],
+            str,
             OptimizationSet,
             list[Path],
             list[Path],
@@ -323,9 +326,15 @@ def generate_and_instrument_tests(
 
         generated_tests: GeneratedTestsList
         optimizations_set: OptimizationSet
-        count_tests, generated_tests, function_to_concolic_tests, concolic_test_str, optimizations_set = (
-            generated_results.unwrap()
-        )
+        (
+            count_tests,
+            generated_tests,
+            function_to_concolic_tests,
+            concolic_test_str,
+            function_to_hypothesis_tests,
+            hypothesis_test_str,
+            optimizations_set,
+        ) = generated_results.unwrap()
 
         for i, generated_test in enumerate(generated_tests.generated_tests):
             with generated_test.behavior_file_path.open("w", encoding="utf8") as f:
@@ -345,12 +354,19 @@ def generate_and_instrument_tests(
             logger.info(f"Generated test {i + 1}/{count_tests}:")
             code_print(generated_test.generated_original_test_source, file_name=f"test_{i + 1}.py")
         if concolic_test_str:
-            logger.info(f"Generated test {count_tests}/{count_tests}:")
+            logger.info(f"Generated test {count_tests - (1 if hypothesis_test_str else 0)}/{count_tests}:")
             code_print(concolic_test_str)
+        if hypothesis_test_str:
+            logger.info(f"Generated test {count_tests}/{count_tests}:")
+            code_print(hypothesis_test_str)
 
         function_to_all_tests = {
-            key: self.function_to_tests.get(key, set()) | function_to_concolic_tests.get(key, set())
-            for key in set(self.function_to_tests) | set(function_to_concolic_tests)
+            key: (
+                self.function_to_tests.get(key, set())
+                | function_to_concolic_tests.get(key, set())
+                | function_to_hypothesis_tests.get(key, set())
+            )
+            for key in set(self.function_to_tests) | set(function_to_concolic_tests) | set(function_to_hypothesis_tests)
         }
         instrumented_unittests_created_for_function = self.instrument_existing_tests(function_to_all_tests)
 
@@ -366,6 +382,8 @@ def generate_and_instrument_tests(
                 generated_tests,
                 function_to_concolic_tests,
                 concolic_test_str,
+                function_to_hypothesis_tests,
+                hypothesis_test_str,
                 optimizations_set,
                 generated_test_paths,
                 generated_perf_test_paths,
@@ -398,6 +416,8 @@ def optimize_function(self) -> Result[BestOptimization, str]:
             generated_tests,
             function_to_concolic_tests,
             concolic_test_str,
+            function_to_hypothesis_tests,
+            _hypothesis_test_str,
             optimizations_set,
             generated_test_paths,
             generated_perf_test_paths,
@@ -409,6 +429,7 @@ def optimize_function(self) -> Result[BestOptimization, str]:
             code_context=code_context,
             original_helper_code=original_helper_code,
             function_to_concolic_tests=function_to_concolic_tests,
+            function_to_hypothesis_tests=function_to_hypothesis_tests,
             generated_test_paths=generated_test_paths,
             generated_perf_test_paths=generated_perf_test_paths,
             instrumented_unittests_created_for_function=instrumented_unittests_created_for_function,
@@ -991,6 +1012,7 @@ def instrument_existing_tests(self, function_to_all_tests: dict[str, set[Functio
         existing_test_files_count = 0
         replay_test_files_count = 0
         concolic_coverage_test_files_count = 0
+        hypothesis_test_files_count = 0
         unique_instrumented_test_files = set()
 
         func_qualname = self.function_to_optimize.qualified_name_with_modules_from_root(self.project_root)
@@ -1011,6 +1033,8 @@ def instrument_existing_tests(self, function_to_all_tests: dict[str, set[Functio
                     replay_test_files_count += 1
                 elif test_type == TestType.CONCOLIC_COVERAGE_TEST:
                     concolic_coverage_test_files_count += 1
+                elif test_type == TestType.HYPOTHESIS_TEST:
+                    hypothesis_test_files_count += 1
                 else:
                     msg = f"Unexpected test type: {test_type}"
                     raise ValueError(msg)
@@ -1069,9 +1093,11 @@ def instrument_existing_tests(self, function_to_all_tests: dict[str, set[Functio
             logger.info(
                 f"Discovered {existing_test_files_count} existing unit test file"
                 f"{'s' if existing_test_files_count != 1 else ''}, {replay_test_files_count} replay test file"
-                f"{'s' if replay_test_files_count != 1 else ''}, and "
+                f"{'s' if replay_test_files_count != 1 else ''}, "
                 f"{concolic_coverage_test_files_count} concolic coverage test file"
-                f"{'s' if concolic_coverage_test_files_count != 1 else ''} for {func_qualname}"
+                f"{'s' if concolic_coverage_test_files_count != 1 else ''}, and "
+                f"{hypothesis_test_files_count} hypothesis test file"
+                f"{'s' if hypothesis_test_files_count != 1 else ''} for {func_qualname}"
             )
             console.rule()
         return unique_instrumented_test_files
@@ -1085,7 +1111,15 @@ def generate_tests_and_optimizations(
         generated_test_paths: list[Path],
         generated_perf_test_paths: list[Path],
         run_experiment: bool = False,  # noqa: FBT001, FBT002
-    ) -> Result[tuple[GeneratedTestsList, dict[str, set[FunctionCalledInTest]], OptimizationSet], str]:
+    ) -> Result[
+        tuple[
+            GeneratedTestsList,
+            dict[str, set[FunctionCalledInTest]],
+            dict[str, set[FunctionCalledInTest]],
+            OptimizationSet,
+        ],
+        str,
+    ]:
         n_tests = N_TESTS_TO_GENERATE_EFFECTIVE
         assert len(generated_test_paths) == n_tests
         console.rule()
@@ -1112,7 +1146,10 @@ def generate_tests_and_optimizations(
         future_concolic_tests = self.executor.submit(
             generate_concolic_tests, self.test_cfg, self.args, self.function_to_optimize, self.function_to_optimize_ast
         )
-        futures = [*future_tests, future_optimization_candidates, future_concolic_tests]
+        future_hypothesis_tests = self.executor.submit(
+            generate_hypothesis_tests, self.test_cfg, self.args, self.function_to_optimize, self.function_to_optimize_ast
+        )
+        futures = [*future_tests, future_optimization_candidates, future_concolic_tests, future_hypothesis_tests]
         if run_experiment:
             future_candidates_exp = self.executor.submit(
                 self.local_aiservice_client.optimize_python_code,
@@ -1164,29 +1201,35 @@ def generate_tests_and_optimizations(
             logger.warning(f"Failed to generate and instrument tests for {self.function_to_optimize.function_name}")
             return Failure(f"/!\\ NO TESTS GENERATED for {self.function_to_optimize.function_name}")
         function_to_concolic_tests, concolic_test_str = future_concolic_tests.result()
+        function_to_hypothesis_tests, hypothesis_test_str = future_hypothesis_tests.result()
 
         count_tests = len(tests)
         if concolic_test_str:
             count_tests += 1
+        if hypothesis_test_str:
+            count_tests += 1
 
         logger.info(f"Generated '{count_tests}' tests for {self.function_to_optimize.function_name}")
         console.rule()
         generated_tests = GeneratedTestsList(generated_tests=tests)
-        result = (
+
+        self.generate_and_instrument_tests_results = (
             count_tests,
             generated_tests,
             function_to_concolic_tests,
             concolic_test_str,
+            function_to_hypothesis_tests,
+            hypothesis_test_str,
             OptimizationSet(control=candidates, experiment=candidates_experiment),
         )
-        self.generate_and_instrument_tests_results = result
-        return Success(result)
+        return Success(self.generate_and_instrument_tests_results)
 
     def setup_and_establish_baseline(
         self,
         code_context: CodeOptimizationContext,
         original_helper_code: dict[Path, str],
         function_to_concolic_tests: dict[str, set[FunctionCalledInTest]],
+        function_to_hypothesis_tests: dict[str, set[FunctionCalledInTest]],
         generated_test_paths: list[Path],
         generated_perf_test_paths: list[Path],
         instrumented_unittests_created_for_function: set[Path],
@@ -1197,8 +1240,12 @@ def setup_and_establish_baseline(
         """Set up baseline context and establish original code baseline."""
         function_to_optimize_qualified_name = self.function_to_optimize.qualified_name
         function_to_all_tests = {
-            key: self.function_to_tests.get(key, set()) | function_to_concolic_tests.get(key, set())
-            for key in set(self.function_to_tests) | set(function_to_concolic_tests)
+            key: (
+                self.function_to_tests.get(key, set())
+                | function_to_concolic_tests.get(key, set())
+                | function_to_hypothesis_tests.get(key, set())
+            )
+            for key in set(self.function_to_tests) | set(function_to_concolic_tests) | set(function_to_hypothesis_tests)
         }
 
         # Get a dict of file_path_to_classes of fto and helpers_of_fto

codeflash/verification/concolic_testing.py

@@ -8,6 +8,7 @@
 from typing import TYPE_CHECKING
 
 from codeflash.cli_cmds.console import console, logger
+from codeflash.code_utils.code_utils import get_qualified_function_path
 from codeflash.code_utils.compat import SAFE_SYS_EXECUTABLE
 from codeflash.code_utils.concolic_utils import clean_concolic_tests
 from codeflash.code_utils.static_analysis import has_typed_parameters
@@ -42,6 +43,9 @@ def generate_concolic_tests(
         logger.info("Generating concolic opcode coverage tests for the original code…")
         console.rule()
         try:
+            qualified_function_path = get_qualified_function_path(
+                function_to_optimize.file_path, args.project_root, function_to_optimize.qualified_name
+            )
             cover_result = subprocess.run(
                 [
                     SAFE_SYS_EXECUTABLE,
@@ -50,15 +54,7 @@ def generate_concolic_tests(
                     "cover",
                     "--example_output_format=pytest",
                     "--per_condition_timeout=20",
-                    ".".join(
-                        [
-                            function_to_optimize.file_path.relative_to(args.project_root)
-                            .with_suffix("")
-                            .as_posix()
-                            .replace("/", "."),
-                            function_to_optimize.qualified_name,
-                        ]
-                    ),
+                    qualified_function_path,
                 ],
                 capture_output=True,
                 text=True,