Internal

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

provide.foundation.testmode.internal

Functions

reset_circuit_breaker_state

reset_circuit_breaker_state() -> None

Reset all circuit breaker instances to ensure test isolation.

This function resets all circuit breaker instances that were created by the @circuit_breaker decorator and direct instantiation to ensure their state doesn't leak between tests.

Source code in provide/foundation/testmode/internal.py

def reset_circuit_breaker_state() -> None:
    """Reset all circuit breaker instances to ensure test isolation.

    This function resets all circuit breaker instances that were created
    by the @circuit_breaker decorator and direct instantiation to ensure
    their state doesn't leak between tests.
    """
    # Reset all CircuitBreaker instances created directly (not via decorator)
    # Do this FIRST to catch all instances before decorator reset
    _reset_direct_circuit_breaker_instances()

    try:
        import asyncio

        from provide.foundation.resilience.decorators import (
            reset_circuit_breakers_for_testing,
            reset_test_circuit_breakers,
        )

        # Reset both production and test circuit breakers
        # These are now async functions, so we need to run them in an event loop

        # Check if we're in an async context (running event loop)
        try:
            asyncio.get_running_loop()
            # We're in an async context - skip reset to avoid blocking
            # This shouldn't happen in practice since reset is called from sync fixtures
            return
        except RuntimeError:
            # No running loop - we're in sync context, safe to proceed
            pass

        # Use asyncio.run() to create fresh event loop for each call
        # This is more reliable than trying to reuse get_event_loop()
        asyncio.run(reset_circuit_breakers_for_testing())
        asyncio.run(reset_test_circuit_breakers())
    except ImportError:
        # Resilience decorators module not available, skip
        pass

reset_configuration_state

reset_configuration_state() -> None

Reset configuration state to defaults.

This resets all versioned configurations and their managers to ensure clean state between tests.

Source code in provide/foundation/testmode/internal.py

def reset_configuration_state() -> None:
    """Reset configuration state to defaults.

    This resets all versioned configurations and their managers
    to ensure clean state between tests.
    """
    try:
        from provide.foundation.hub.manager import get_hub

        hub = get_hub()

        # Reset config manager if available
        try:
            config_manager = hub.get_component("config_manager")
            if config_manager and hasattr(config_manager, "clear_all"):
                config_manager.clear_all()
        except Exception:
            # Config manager not available or reset failed, skip
            pass

    except ImportError:
        # Hub not available, skip
        pass

reset_coordinator_state

reset_coordinator_state() -> None

Reset setup coordinator state.

This clears cached coordinator state including setup logger cache and other coordinator-managed state.

Source code in provide/foundation/testmode/internal.py

def reset_coordinator_state() -> None:
    """Reset setup coordinator state.

    This clears cached coordinator state including setup logger
    cache and other coordinator-managed state.
    """
    try:
        from provide.foundation.logger.setup.coordinator import reset_coordinator_state

        reset_coordinator_state()
    except ImportError:
        # Coordinator module not available, skip
        pass

    try:
        from provide.foundation.logger.setup.coordinator import reset_setup_logger_cache

        reset_setup_logger_cache()
    except ImportError:
        # Setup logger cache not available, skip
        pass

reset_event_loops

reset_event_loops() -> None

Close any running event loops to prevent worker shutdown hangs.

This is critical for pytest-xdist workers to shut down cleanly after async tests complete.

IMPORTANT: This MUST be called AFTER reset_time_machine_state() to ensure time patches are stopped before creating a new event loop. Otherwise the new loop may cache frozen time.monotonic references.

Source code in provide/foundation/testmode/internal.py

def reset_event_loops() -> None:
    """Close any running event loops to prevent worker shutdown hangs.

    This is critical for pytest-xdist workers to shut down cleanly after
    async tests complete.

    IMPORTANT: This MUST be called AFTER reset_time_machine_state() to ensure
    time patches are stopped before creating a new event loop. Otherwise the
    new loop may cache frozen time.monotonic references.
    """
    try:
        import asyncio

        # Close the current event loop if it's not running
        try:
            loop = asyncio.get_event_loop()
            # Don't close if it's running (we're inside an async context)
            if not loop.is_running() and not loop.is_closed():
                loop.close()
        except RuntimeError:
            # No event loop in this thread, that's fine
            pass

        # DON'T create a new event loop - let pytest-asyncio manage it
        # If we create one here while time patches are still active (fixture cleanup
        # hasn't run yet), the new loop will cache frozen time.monotonic references
    except Exception:
        # If anything fails, continue - better to leak a loop than crash
        pass

reset_eventsets_state

reset_eventsets_state() -> None

Reset event set registry and discovery state.

This clears the event set registry to ensure clean state between tests.

Source code in provide/foundation/testmode/internal.py

def reset_eventsets_state() -> None:
    """Reset event set registry and discovery state.

    This clears the event set registry to ensure clean state
    between tests.
    """
    global _eventsets_reset_in_progress

    # Prevent recursive resets that trigger event processor initialization
    if _eventsets_reset_in_progress:
        return

    _eventsets_reset_in_progress = True
    try:
        from provide.foundation.eventsets.registry import clear_registry

        clear_registry()
    except ImportError:
        # Event sets may not be available in all test environments
        pass
    finally:
        _eventsets_reset_in_progress = False

reset_global_coordinator

reset_global_coordinator() -> None

Reset the global initialization coordinator state for testing.

This function resets the singleton InitializationCoordinator state to ensure proper test isolation between test runs.

WARNING: This should only be called from test code or test fixtures. Production code should not reset the global coordinator state.

Source code in provide/foundation/testmode/internal.py

def reset_global_coordinator() -> None:
    """Reset the global initialization coordinator state for testing.

    This function resets the singleton InitializationCoordinator state
    to ensure proper test isolation between test runs.

    WARNING: This should only be called from test code or test fixtures.
    Production code should not reset the global coordinator state.
    """
    try:
        from provide.foundation.hub.initialization import _coordinator

        _coordinator.reset_state()
    except ImportError:
        # Initialization module not available, skip
        pass

reset_hub_state

reset_hub_state() -> None

Reset Hub state to defaults.

This clears the Hub registry and resets all Hub components to their initial state.

Source code in provide/foundation/testmode/internal.py

def reset_hub_state() -> None:
    """Reset Hub state to defaults.

    This clears the Hub registry and resets all Hub components
    to their initial state.
    """
    global _hub_reset_in_progress

    # Prevent recursive resets that can trigger re-initialization
    if _hub_reset_in_progress:
        return

    _hub_reset_in_progress = True
    try:
        try:
            from provide.foundation.hub.manager import clear_hub

            clear_hub()
        except ImportError:
            # Hub module not available, skip
            pass

        try:
            # Also reset the initialized components cache
            from provide.foundation.hub.components import _initialized_components

            _initialized_components.clear()
        except ImportError:
            # Components module not available, skip
            pass

        try:
            # Clear the global component registry (where bootstrap_foundation registers components)
            from provide.foundation.hub.components import _component_registry

            _component_registry.clear()
        except ImportError:
            # Components module not available, skip
            pass

        try:
            # Clear the global command registry (where @register_command decorator registers commands)
            from provide.foundation.hub.registry import _command_registry

            _command_registry.clear()
        except ImportError:
            # Registry module not available, skip
            pass
    finally:
        _hub_reset_in_progress = False

reset_logger_state

reset_logger_state() -> None

Reset Foundation logger state to defaults.

This resets the lazy setup state and logger configuration flags without importing the full logger module to avoid circular dependencies.

Source code in provide/foundation/testmode/internal.py

def reset_logger_state() -> None:
    """Reset Foundation logger state to defaults.

    This resets the lazy setup state and logger configuration flags
    without importing the full logger module to avoid circular dependencies.
    """
    try:
        from provide.foundation.logger.core import _LAZY_SETUP_STATE

        _LAZY_SETUP_STATE.update({"done": False, "error": None, "in_progress": False})
    except ImportError:
        # Logger state not available, skip
        pass

    try:
        from provide.foundation.logger.core import logger as foundation_logger

        # Reset foundation logger state by bypassing the proxy to avoid circular initialization
        # Access the proxy's __dict__ directly to avoid triggering __setattr__
        foundation_logger.__dict__["_is_configured_by_setup"] = False
        foundation_logger.__dict__["_active_config"] = None
        foundation_logger.__dict__["_active_resolved_emoji_config"] = None
    except (ImportError, AttributeError, TypeError):
        # Skip if foundation_logger is a proxy without direct attribute access
        pass

reset_profiling_state

reset_profiling_state() -> None

Reset profiling state to defaults.

This clears profiling metrics and resets profiling components to ensure clean state between tests.

Source code in provide/foundation/testmode/internal.py

def reset_profiling_state() -> None:
    """Reset profiling state to defaults.

    This clears profiling metrics and resets profiling components
    to ensure clean state between tests.
    """
    try:
        from provide.foundation.hub.manager import get_hub

        hub = get_hub()
        profiler = hub.get_component("profiler")
        if profiler:
            profiler.reset()
    except ImportError:
        # Profiling module or Hub not available, skip
        pass

reset_state_managers

reset_state_managers() -> None

Reset all state managers to default state.

This resets logger state managers and other state management components to ensure clean test isolation.

Source code in provide/foundation/testmode/internal.py

def reset_state_managers() -> None:
    """Reset all state managers to default state.

    This resets logger state managers and other state management
    components to ensure clean test isolation.
    """
    try:
        from provide.foundation.hub.manager import get_hub

        hub = get_hub()

        # Reset logger state manager if available
        try:
            logger_state_manager = hub.get_component("logger_state_manager")
            if logger_state_manager and hasattr(logger_state_manager, "reset_to_default"):
                logger_state_manager.reset_to_default()
        except Exception:
            # Logger state manager not available or reset failed, skip
            pass

    except ImportError:
        # Hub not available, skip
        pass

reset_streams_state

reset_streams_state() -> None

Reset stream state to defaults.

This resets file streams and other stream-related state managed by the streams module.

Source code in provide/foundation/testmode/internal.py

def reset_streams_state() -> None:
    """Reset stream state to defaults.

    This resets file streams and other stream-related state
    managed by the streams module.
    """
    try:
        from provide.foundation.streams.file import reset_streams

        reset_streams()
    except ImportError:
        # Streams module not available, skip
        pass

reset_structlog_state

reset_structlog_state() -> None

Reset structlog configuration to a test-safe state.

This resets structlog but configures it with a wrapper class that supports the trace method. Using reset_defaults() alone would result in BoundLoggerFilteringAtNotset which lacks trace().

Source code in provide/foundation/testmode/internal.py

def reset_structlog_state() -> None:
    """Reset structlog configuration to a test-safe state.

    This resets structlog but configures it with a wrapper class that
    supports the trace method. Using reset_defaults() alone would
    result in BoundLoggerFilteringAtNotset which lacks trace().
    """
    import sys

    # Reset first to clear any cached loggers
    structlog.reset_defaults()

    def _strip_foundation_context(
        _logger: object,
        _method_name: str,
        event_dict: dict[str, object],
    ) -> dict[str, object]:
        """Strip Foundation-specific bound context before rendering.

        Foundation binds logger_name and other context that PrintLogger
        doesn't accept as kwargs. This processor removes them.
        """
        # Remove Foundation-specific keys that shouldn't be passed to PrintLogger
        event_dict.pop("logger_name", None)
        event_dict.pop("_foundation_level_hint", None)
        return event_dict

    # Reconfigure with BoundLogger which supports trace via Foundation's patching
    # Using PrintLoggerFactory with stdout for test safety (parallel test compat)
    structlog.configure(
        processors=[
            structlog.processors.TimeStamper(fmt="iso"),
            _strip_foundation_context,
            structlog.dev.ConsoleRenderer(),
        ],
        wrapper_class=structlog.BoundLogger,
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=False,  # Disable caching for test isolation
    )

reset_test_mode_cache

reset_test_mode_cache() -> None

Pin test mode cache to True; preserve click-testing cache.

During a test run, test mode is always True. Previous behavior cleared both caches (setting them to None), which forced expensive sys._getframe() walks (or worse, inspect.stack() + getmodule()) on every subsequent Foundation re-initialization — ~17 GB of allocations across 10K resets.

Now we pin _test_mode_cache = True so the fast-path cache hit fires immediately, and leave _click_testing_cache untouched because Click-testing status also doesn't change mid-run.

Tests that specifically need to exercise detection logic can call _clear_test_mode_cache() directly from detection.py.

Source code in provide/foundation/testmode/internal.py

def reset_test_mode_cache() -> None:
    """Pin test mode cache to True; preserve click-testing cache.

    During a test run, test mode is always True.  Previous behavior
    cleared both caches (setting them to None), which forced expensive
    sys._getframe() walks (or worse, inspect.stack() + getmodule()) on
    every subsequent Foundation re-initialization — ~17 GB of allocations
    across 10K resets.

    Now we pin ``_test_mode_cache = True`` so the fast-path cache hit
    fires immediately, and leave ``_click_testing_cache`` untouched
    because Click-testing status also doesn't change mid-run.

    Tests that specifically need to exercise detection logic can call
    ``_clear_test_mode_cache()`` directly from detection.py.
    """
    import provide.foundation.testmode.detection as _det

    _det._test_mode_cache = True

reset_time_machine_state

reset_time_machine_state() -> None

Reset time_machine state to ensure time is not frozen.

NOTE: The actual cleanup is now handled by the _force_time_machine_cleanup fixture in tests/conftest.py, which runs BEFORE Foundation teardown.

This ensures time patches are stopped before pytest-asyncio creates event loops for the next test. This function remains as a no-op safety fallback.

Source code in provide/foundation/testmode/internal.py

def reset_time_machine_state() -> None:
    """Reset time_machine state to ensure time is not frozen.

    NOTE: The actual cleanup is now handled by the _force_time_machine_cleanup
    fixture in tests/conftest.py, which runs BEFORE Foundation teardown.

    This ensures time patches are stopped before pytest-asyncio creates event loops
    for the next test. This function remains as a no-op safety fallback.
    """

reset_transport_registration_flags

reset_transport_registration_flags() -> None

Reset transport auto-registration flags.

This resets the module-level flags that prevent duplicate transport registration, allowing transports to be re-registered after the registry is cleared during test cleanup.

Source code in provide/foundation/testmode/internal.py

def reset_transport_registration_flags() -> None:
    """Reset transport auto-registration flags.

    This resets the module-level flags that prevent duplicate transport
    registration, allowing transports to be re-registered after the
    registry is cleared during test cleanup.
    """
    try:
        import sys

        # Reset HTTP transport registration flag if module is loaded
        if "provide.foundation.transport.http" in sys.modules:
            http_module = sys.modules["provide.foundation.transport.http"]
            if hasattr(http_module, "_http_transport_registered"):
                http_module._http_transport_registered = False  # type: ignore[attr-defined]
    except Exception:
        # If reset fails, skip - the guard will be bypassed on next import
        pass

reset_version_cache

reset_version_cache() -> None

Reset version cache to defaults.

This clears the cached version to ensure clean state between tests, allowing each test to verify different version resolution scenarios.

Source code in provide/foundation/testmode/internal.py

def reset_version_cache() -> None:
    """Reset version cache to defaults.

    This clears the cached version to ensure clean state
    between tests, allowing each test to verify different
    version resolution scenarios.
    """
    try:
        from provide.foundation._version import (  # type: ignore[import-untyped]
            reset_version_cache as _reset_cache,
        )

        _reset_cache()
    except ImportError:
        # Version module not available, skip
        pass