Skip to content

Data fixtures

provide.testkit.threading.data_fixtures

Thread-safe data structure test fixtures.

Fixtures for thread-safe lists, counters, and other data structures for testing.

Classes

ThreadSafeCounter 

ThreadSafeCounter(initial: int = 0)

Thread-safe integer counter.

Source code in provide/testkit/threading/data_fixtures.py 
58
59
60
def __init__(self, initial: int = 0) -> None:
    self._value = initial
    self._lock = threading.Lock()
Attributes
value property 
value: int

Get current value.

Functions
decrement 
decrement(amount: int = 1) -> int

Thread-safe decrement.

Source code in provide/testkit/threading/data_fixtures.py 
68
69
70
71
72
def decrement(self, amount: int = 1) -> int:
    """Thread-safe decrement."""
    with self._lock:
        self._value -= amount
        return self._value
increment 
increment(amount: int = 1) -> int

Thread-safe increment.

Source code in provide/testkit/threading/data_fixtures.py 
62
63
64
65
66
def increment(self, amount: int = 1) -> int:
    """Thread-safe increment."""
    with self._lock:
        self._value += amount
        return self._value
reset 
reset(value: int = 0) -> None

Reset counter.

Source code in provide/testkit/threading/data_fixtures.py 
80
81
82
83
def reset(self, value: int = 0) -> None:
    """Reset counter."""
    with self._lock:
        self._value = value

ThreadSafeList 

ThreadSafeList()

Simple thread-safe list wrapper.

Source code in provide/testkit/threading/data_fixtures.py 
22
23
24
def __init__(self) -> None:
    self._list: list[Any] = []
    self._lock = threading.Lock()
Functions
append 
append(item: Any) -> None

Thread-safe append.

Source code in provide/testkit/threading/data_fixtures.py 
26
27
28
29
def append(self, item: Any) -> None:
    """Thread-safe append."""
    with self._lock:
        self._list.append(item)
clear 
clear() -> None

Clear the list.

Source code in provide/testkit/threading/data_fixtures.py 
41
42
43
44
def clear(self) -> None:
    """Clear the list."""
    with self._lock:
        self._list.clear()
extend 
extend(items: Iterable[Any]) -> None

Thread-safe extend.

Source code in provide/testkit/threading/data_fixtures.py 
31
32
33
34
def extend(self, items: Iterable[Any]) -> None:
    """Thread-safe extend."""
    with self._lock:
        self._list.extend(items)
get_all 
get_all() -> list[Any]

Get copy of all items.

Source code in provide/testkit/threading/data_fixtures.py 
36
37
38
39
def get_all(self) -> list[Any]:
    """Get copy of all items."""
    with self._lock:
        return self._list.copy()

Functions

thread_safe_counter 

thread_safe_counter() -> ThreadSafeCounter

Create a thread-safe counter.

Returns:

Type Description
ThreadSafeCounter

Thread-safe counter implementation.
Source code in provide/testkit/threading/data_fixtures.py 
 97
 98
 99
100
101
102
103
104
105
@pytest.fixture
def thread_safe_counter() -> ThreadSafeCounter:
    """
    Create a thread-safe counter.

    Returns:
        Thread-safe counter implementation.
    """
    return ThreadSafeCounter()

thread_safe_list 

thread_safe_list() -> ThreadSafeList

Create a thread-safe list for collecting results.

Returns:

Type Description
ThreadSafeList

Thread-safe list implementation.
Source code in provide/testkit/threading/data_fixtures.py 
86
87
88
89
90
91
92
93
94
@pytest.fixture
def thread_safe_list() -> ThreadSafeList:
    """
    Create a thread-safe list for collecting results.

    Returns:
        Thread-safe list implementation.
    """
    return ThreadSafeList()