Threadsafe object test with lock

test_threadsaftey.py

from __future__ import annotations
import threading
import pytest



class NoLock:
    INC_NUM = 1000000
    def __init__(self):
        self.value = 0

    def increment(self):
        for i in range(self.INC_NUM):
            self.value += 1


class WithLock:
    INC_NUM = 1000000
    lock = threading.Lock()
    def __init__(self):
        self.value = 0

    def increment(self):
        with self.lock:
            for i in range(self.INC_NUM):
                self.value += 1


@pytest.mark.xfail
def test_generic_threadsafety_test_failure():
    cls = NoLock()
    def inc():
        cls.increment()

    n_threads = 10
    threads: list[threading.Thread] = []
    # Create and start multiple threads.
    for _ in range(n_threads):
        t = threading.Thread(target=inc)
        threads.append(t)
        t.start()

    # Wait for all threads to finish.
    for t in threads:
        t.join()

    # We expect that some of the threads fails, thus the value will be lower than expected
    assert cls.value > NoLock.INC_NUM * n_threads


def test_generic_threadsafety():
    cls = WithLock()
    def inc():
        cls.increment()

    n_threads = 10
    threads: list[threading.Thread] = []
    # Create and start multiple threads.
    for _ in range(n_threads):
        t = threading.Thread(target=inc)
        threads.append(t)
        t.start()

    # Wait for all threads to finish.
    for t in threads:
        t.join()

    assert cls.value == WithLock.INC_NUM * n_threads