patch randomness while running tests

code_to_optimize/topological_sort.py

@@ -1,3 +1,4 @@
+import uuid
 from collections import defaultdict
 
 
@@ -21,9 +22,10 @@ def topologicalSortUtil(self, v, visited, stack):
     def topologicalSort(self):
         visited = [False] * self.V
         stack = []
+        sorting_id = uuid.uuid4()
 
         for i in range(self.V):
             if visited[i] == False:
                 self.topologicalSortUtil(i, visited, stack)
 
-        return stack
+        return stack, str(sorting_id)

codeflash/verification/pytest_plugin.py

@@ -9,7 +9,7 @@
 import platform
 import re
 import sys
-import time
+import time as _time_module
 import warnings
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable
@@ -74,6 +74,125 @@ class UnexpectedError(Exception):
     resource.setrlimit(resource.RLIMIT_AS, (memory_limit, memory_limit))
 
 
+# Store references to original functions before any patching
+_ORIGINAL_TIME_TIME = _time_module.time
+_ORIGINAL_PERF_COUNTER = _time_module.perf_counter
+_ORIGINAL_TIME_SLEEP = _time_module.sleep
+
+
+# Apply deterministic patches for reproducible test execution
+def _apply_deterministic_patches() -> None:
+    """Apply patches to make all sources of randomness deterministic."""
+    import datetime
+    import random
+    import time
+    import uuid
+
+    # Store original functions (these are already saved globally above)
+    _original_time = time.time
+    _original_perf_counter = time.perf_counter
+    _original_datetime_now = datetime.datetime.now
+    _original_datetime_utcnow = datetime.datetime.utcnow
+    _original_uuid4 = uuid.uuid4
+    _original_uuid1 = uuid.uuid1
+    _original_random = random.random
+
+    # Fixed deterministic values
+    fixed_timestamp = 1609459200.0  # 2021-01-01 00:00:00 UTC
+    fixed_datetime = datetime.datetime(2021, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
+    fixed_uuid = uuid.UUID("12345678-1234-5678-9abc-123456789012")
+
+    # Counter for perf_counter to maintain relative timing
+    _perf_counter_start = fixed_timestamp
+    _perf_counter_calls = 0
+
+    def mock_time_time() -> float:
+        """Return fixed timestamp while preserving performance characteristics."""
+        _original_time()  # Maintain performance characteristics
+        return fixed_timestamp
+
+    def mock_perf_counter() -> float:
+        """Return incrementing counter for relative timing."""
+        nonlocal _perf_counter_calls
+        _original_perf_counter()  # Maintain performance characteristics
+        _perf_counter_calls += 1
+        return _perf_counter_start + (_perf_counter_calls * 0.001)  # Increment by 1ms each call
+
+    def mock_datetime_now(tz: datetime.timezone | None = None) -> datetime.datetime:
+        """Return fixed datetime while preserving performance characteristics."""
+        _original_datetime_now(tz)  # Maintain performance characteristics
+        if tz is None:
+            return fixed_datetime
+        return fixed_datetime.replace(tzinfo=tz)
+
+    def mock_datetime_utcnow() -> datetime.datetime:
+        """Return fixed UTC datetime while preserving performance characteristics."""
+        _original_datetime_utcnow()  # Maintain performance characteristics
+        return fixed_datetime
+
+    def mock_uuid4() -> uuid.UUID:
+        """Return fixed UUID4 while preserving performance characteristics."""
+        _original_uuid4()  # Maintain performance characteristics
+        return fixed_uuid
+
+    def mock_uuid1(node: int | None = None, clock_seq: int | None = None) -> uuid.UUID:
+        """Return fixed UUID1 while preserving performance characteristics."""
+        _original_uuid1(node, clock_seq)  # Maintain performance characteristics
+        return fixed_uuid
+
+    def mock_random() -> float:
+        """Return deterministic random value while preserving performance characteristics."""
+        _original_random()  # Maintain performance characteristics
+        return 0.123456789  # Fixed random value
+
+    # Apply patches
+    time.time = mock_time_time
+    time.perf_counter = mock_perf_counter
+    uuid.uuid4 = mock_uuid4
+    uuid.uuid1 = mock_uuid1
+
+    # Seed random module for other random functions
+    random.seed(42)
+    random.random = mock_random
+
+    # For datetime, we need to use a different approach since we can't patch class methods
+    # Store original methods for potential later use
+    import builtins
+
+    builtins._original_datetime_now = _original_datetime_now  # noqa: SLF001
+    builtins._original_datetime_utcnow = _original_datetime_utcnow  # noqa: SLF001
+    builtins._mock_datetime_now = mock_datetime_now  # noqa: SLF001
+    builtins._mock_datetime_utcnow = mock_datetime_utcnow  # noqa: SLF001
+
+    # Patch numpy.random if available
+    try:
+        import numpy as np
+
+        # Use modern numpy random generator approach
+        np.random.default_rng(42)
+        np.random.seed(42)  # Keep legacy seed for compatibility  # noqa: NPY002
+    except ImportError:
+        pass
+
+    # Patch os.urandom if needed
+    try:
+        import os
+
+        _original_urandom = os.urandom
+
+        def mock_urandom(n: int) -> bytes:
+            _original_urandom(n)  # Maintain performance characteristics
+            return b"\x42" * n  # Fixed bytes
+
+        os.urandom = mock_urandom
+    except (ImportError, AttributeError):
+        pass
+
+
+# Note: Deterministic patches are applied conditionally, not globally
+# They should only be applied when running CodeFlash optimization tests
+
+
 def pytest_addoption(parser: Parser) -> None:
     """Add command line options."""
     pytest_loops = parser.getgroup("loops")
@@ -137,6 +256,9 @@ def pytest_configure(config: Config) -> None:
     config.addinivalue_line("markers", "loops(n): run the given test function `n` times.")
     config.pluginmanager.register(PytestLoops(config), PytestLoops.name)
 
+    # Apply deterministic patches when the plugin is configured
+    _apply_deterministic_patches()
+
 
 class PytestLoops:
     name: str = "pytest-loops"
@@ -157,7 +279,7 @@ def pytest_runtestloop(self, session: Session) -> bool:
         if session.config.option.collectonly:
             return True
 
-        start_time: float = time.time()
+        start_time: float = _ORIGINAL_TIME_TIME()
         total_time: float = self._get_total_time(session)
 
         count: int = 0
@@ -184,7 +306,7 @@ def pytest_runtestloop(self, session: Session) -> bool:
                     raise session.Interrupted(session.shouldstop)
             if self._timed_out(session, start_time, count):
                 break  # exit loop
-            time.sleep(self._get_delay_time(session))
+            _ORIGINAL_TIME_SLEEP(self._get_delay_time(session))
         return True
 
     def _clear_lru_caches(self, item: pytest.Item) -> None:
@@ -283,7 +405,7 @@ def _timed_out(self, session: Session, start_time: float, count: int) -> bool:
         """
         return count >= session.config.option.codeflash_max_loops or (
             count >= session.config.option.codeflash_min_loops
-            and time.time() - start_time > self._get_total_time(session)
+            and _ORIGINAL_TIME_TIME() - start_time > self._get_total_time(session)
         )
 
     @pytest.fixture

tests/scripts/end_to_end_test_topological_sort.py

@@ -1,6 +1,5 @@
 import os
 import pathlib
-import tomlkit
 
 from codeflash.code_utils.code_utils import add_addopts_to_pyproject
 from end_to_end_test_utilities import CoverageExpectation, TestConfig, run_codeflash_command, run_with_retries
@@ -17,7 +16,7 @@ def run_test(expected_improvement_pct: int) -> bool:
                 CoverageExpectation(
                     function_name="Graph.topologicalSort",
                     expected_coverage=100.0,
-                    expected_lines=[24, 25, 26, 27, 28, 29],
+                    expected_lines=[25, 26, 27, 28, 29, 30, 31],
                 )
             ],
         )