Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 29 additions & 49 deletions codeflash/verification/pytest_plugin.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,20 @@
from __future__ import annotations

import builtins
import contextlib
import datetime
import inspect

# System Imports
import logging
import os
import platform
import random
import re
import sys
import time
import time as _time_module
import uuid
import warnings
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable
Expand Down Expand Up @@ -83,105 +88,80 @@ class UnexpectedError(Exception):
# 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
# Store original functions (no need to do this repeatedly if already patched)
if getattr(time, "_is_patched", False):
return

# Fixed deterministic values (move to module-scope speeds up repeated function calls!)
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
# Fast relative perf_counter using attribute for state
def mock_perf_counter() -> float:
"""Return incrementing counter for relative timing."""
mock_perf_counter.calls += 1
return fixed_timestamp + (mock_perf_counter.calls * 0.001)

mock_perf_counter.calls = 0 # type: ignore

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
# Apply deterministic patches
time.time = mock_time_time
time.perf_counter = mock_perf_counter
time._is_patched = True # sentinel to avoid double patching

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 = datetime.datetime.now
builtins._original_datetime_utcnow = datetime.datetime.utcnow
builtins._mock_datetime_now = mock_datetime_now
builtins._mock_datetime_utcnow = mock_datetime_utcnow

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
np.random.seed(42) # Keep legacy seed for compatibility
# Don't call np.random.default_rng(42) each patch!
np._rng = getattr(np, "_rng", None)
if np._rng is None:
np._rng = np.random.default_rng(42)
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
Expand Down
Loading