Skip to content

Basic fixtures

provide.testkit.threading.basic_fixtures

Basic threading test fixtures.

Core fixtures for creating threads, thread pools, mocks, and thread-local storage.

Functions

mock_thread 

mock_thread() -> Mock

Create a mock thread for testing without actual threading.

Returns:

Type Description
Mock

Mock thread object.
Source code in provide/testkit/threading/basic_fixtures.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@pytest.fixture
def mock_thread() -> Mock:
    """
    Create a mock thread for testing without actual threading.

    Returns:
        Mock thread object.
    """
    mock = Mock(spec=threading.Thread)
    mock.is_alive.return_value = False
    mock.daemon = False
    mock.name = "MockThread"
    mock.ident = 12345
    mock.start = Mock()
    mock.join = Mock()
    mock.run = Mock()

    return mock

test_thread 

test_thread() -> Callable[
    [
        Callable[..., Any],
        tuple[Any, ...],
        dict[str, Any] | None,
        bool,
    ],
    threading.Thread,
]

Create a test thread with automatic cleanup.

Returns:

Type Description
Callable[[Callable[..., Any], tuple[Any, ...], dict[str, Any] | None, bool], Thread]

Function to create and manage test threads.
Source code in provide/testkit/threading/basic_fixtures.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@pytest.fixture
def test_thread() -> Callable[
    [Callable[..., Any], tuple[Any, ...], dict[str, Any] | None, bool],
    threading.Thread,
]:
    """
    Create a test thread with automatic cleanup.

    Returns:
        Function to create and manage test threads.
    """
    threads: list[threading.Thread] = []

    def _create_thread(
        target: Callable[..., Any],
        args: tuple[Any, ...] = (),
        kwargs: dict[str, Any] | None = None,
        daemon: bool = True,
    ) -> threading.Thread:
        """
        Create a test thread.

        Args:
            target: Function to run in thread
            args: Positional arguments for target
            kwargs: Keyword arguments for target
            daemon: Whether thread should be daemon

        Returns:
            Started thread instance
        """
        kwargs = kwargs or {}
        thread = threading.Thread(target=target, args=args, kwargs=kwargs, daemon=daemon)
        threads.append(thread)
        thread.start()
        return thread

    yield _create_thread

    # Cleanup: wait for all threads to complete
    for thread in threads:
        if thread.is_alive():
            thread.join(timeout=1.0)

thread_local_storage 

thread_local_storage() -> threading.local

Create thread-local storage for testing.

Returns:

Type Description
local

Thread-local storage object.
Source code in provide/testkit/threading/basic_fixtures.py 
100
101
102
103
104
105
106
107
108
@pytest.fixture
def thread_local_storage() -> threading.local:
    """
    Create thread-local storage for testing.

    Returns:
        Thread-local storage object.
    """
    return threading.local()

thread_pool 

thread_pool() -> ThreadPoolExecutor

Create a thread pool executor for testing.

Returns:

Type Description
ThreadPoolExecutor

ThreadPoolExecutor instance with automatic cleanup.
Source code in provide/testkit/threading/basic_fixtures.py 
67
68
69
70
71
72
73
74
75
76
77
@pytest.fixture
def thread_pool() -> ThreadPoolExecutor:
    """
    Create a thread pool executor for testing.

    Returns:
        ThreadPoolExecutor instance with automatic cleanup.
    """
    executor = ThreadPoolExecutor(max_workers=4)
    yield executor
    executor.shutdown(wait=True, cancel_futures=True)