pytest_test_categories.adapters.threading

Production thread monitor adapter using threading module patching.

This module provides the production implementation of ThreadMonitorPort that actually intercepts thread creation by patching the threading module.

The ThreadPatchingMonitor follows hexagonal architecture principles: - Implements the ThreadMonitorPort interface (port) - Patches threading.Thread and concurrent.futures executors to intercept creation - Emits pytest warnings on thread creation in small tests - Restores original threading behavior on deactivation

Unlike other blockers that RAISE exceptions, this monitor WARNS because: 1. Many libraries use threading internally (logging, garbage collection) 2. Some test frameworks use threading 3. Blocking threading could break legitimate test infrastructure

Example

>>> monitor = ThreadPatchingMonitor()
>>> try:
...     monitor.activate(TestSize.SMALL, EnforcementMode.WARN)
...     # Any threading.Thread() call will now emit a warning
... finally:
...     monitor.deactivate()  # Restore original threading behavior

See also

  • ThreadMonitorPort: The abstract interface in ports/threading.py

  • FakeThreadMonitor: Test adapter in adapters/fake_threading.py

  • SocketPatchingNetworkBlocker: Similar production adapter pattern for networking

Classes

ThreadPatchingMonitor

Production adapter that patches threading modules to monitor thread creation.

Module Contents

class pytest_test_categories.adapters.threading.ThreadPatchingMonitor(/, **data)[source]

Bases: pytest_test_categories.ports.threading.ThreadMonitorPort

Production adapter that patches threading modules to monitor thread creation.

This adapter intercepts thread creation by replacing threading.Thread and concurrent.futures executors with wrapper classes that emit warnings.

The patching is reversible - deactivate() restores the original classes.

Parameters:

data (Any)

state[source]

Current monitor state (inherited from ThreadMonitorPort).

current_test_size[source]

The test size set during activation.

current_enforcement_mode[source]

The enforcement mode set during activation.

current_test_nodeid[source]

The pytest node ID of the current test.

Warning

This adapter modifies global state (threading.Thread). Always use in a try/finally block or context manager to ensure cleanup.

Example

>>> monitor = ThreadPatchingMonitor()
>>> try:
...     monitor.activate(TestSize.SMALL, EnforcementMode.WARN)
...     threading.Thread(target=lambda: None)  # Emits warning
... finally:
...     monitor.deactivate()

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_post_init(context, /)[source]

Initialize post-Pydantic setup, storing reference to original classes.

Parameters:

context (object)

Return type:

None

reset()[source]

Reset monitor to initial state, restoring original threading classes.

This is safe to call regardless of current state.

Return type:

None

current_enforcement_mode: pytest_test_categories.ports.network.EnforcementMode | None = None[source]
current_test_nodeid: str = None[source]
current_test_size: pytest_test_categories.types.TestSize | None = None[source]
property is_monitoring: bool[source]

Return True if actively monitoring for thread creation.

Only small tests are monitored for threading usage.

Returns:

True if monitoring is active and test is SMALL, False otherwise.

Return type:

bool