remove has_any_async_functions

race cond and invalid method

Author: KRRT7

codeflash/code_utils/code_utils.py

@@ -251,14 +251,6 @@ def validate_python_code(code: str) -> str:
     return code
 
 
-def has_any_async_functions(code: str) -> bool:
-    try:
-        module = ast.parse(code)
-    except SyntaxError:
-        return False
-    return any(isinstance(node, ast.AsyncFunctionDef) for node in ast.walk(module))
-
-
 def cleanup_paths(paths: list[Path]) -> None:
     for path in paths:
         if path and path.exists():

codeflash/verification/coverage_utils.py

@@ -38,10 +38,10 @@ def load_from_sqlite_database(
 
         cov = Coverage(data_file=database_path, config_file=config_path, data_suffix=True, auto_data=True, branch=True)
 
-        if not database_path.stat().st_size or not database_path.exists():
+        if not database_path.exists() or not database_path.stat().st_size:
             logger.debug(f"Coverage database {database_path} is empty or does not exist")
             sentry_sdk.capture_message(f"Coverage database {database_path} is empty or does not exist")
-            return CoverageUtils.create_empty(source_code_path, function_name, code_context)
+            return CoverageData.create_empty(source_code_path, function_name, code_context)
         cov.load()
 
         reporter = JsonReporter(cov)
@@ -51,7 +51,7 @@ def load_from_sqlite_database(
                 reporter.report(morfs=[source_code_path.as_posix()], outfile=f)
             except NoDataError:
                 sentry_sdk.capture_message(f"No coverage data found for {function_name} in {source_code_path}")
-                return CoverageUtils.create_empty(source_code_path, function_name, code_context)
+                return CoverageData.create_empty(source_code_path, function_name, code_context)
         with temp_json_file.open() as f:
             original_coverage_data = json.load(f)
 

tests/test_code_utils.py

@@ -17,7 +17,7 @@
     is_class_defined_in_file,
     module_name_from_file_path,
     path_belongs_to_site_packages,
-    has_any_async_functions,
+    validate_python_code,
 )
 from codeflash.code_utils.concolic_utils import clean_concolic_tests
 from codeflash.code_utils.coverage_utils import generate_candidates, prepare_coverage_files
@@ -445,25 +445,41 @@ def test_Grammar_copy():
     assert cleaned_code == expected_cleaned_code.strip()
 
 
-def test_has_any_async_functions_with_async_code() -> None:
-    code = """
-def normal_function():
-    pass
+def test_validate_python_code_valid() -> None:
+    code = "def hello():\n    return 'world'"
+    result = validate_python_code(code)
+    assert result == code
 
-async def async_function():
-    pass
-"""
-    result = has_any_async_functions(code)
-    assert result is True
 
+def test_validate_python_code_invalid() -> None:
+    code = "def hello(:\n    return 'world'"
+    with pytest.raises(ValueError, match="Invalid Python code"):
+        validate_python_code(code)
 
-def test_has_any_async_functions_without_async_code() -> None:
-    code = """
-def normal_function():
-    pass
 
-def another_function():
-    pass
+def test_validate_python_code_empty() -> None:
+    code = ""
+    result = validate_python_code(code)
+    assert result == code
+
+
+def test_validate_python_code_complex_invalid() -> None:
+    code = "if True\n    print('missing colon')"
+    with pytest.raises(ValueError, match="Invalid Python code.*line 1.*column 8"):
+        validate_python_code(code)
+
+
+def test_validate_python_code_valid_complex() -> None:
+    code = """
+def calculate(a, b):
+    if a > b:
+        return a + b
+    else:
+        return a * b
+        
+class MyClass:
+    def __init__(self):
+        self.value = 42
 """
-    result = has_any_async_functions(code)
-    assert result is False
+    result = validate_python_code(code)
+    assert result == code