Testing Python methods call sequence in a multithreaded context

Question

I need to check the sequence of calling of some methods of a class. I had this need while I was developing using TDD (test driven development), so when I was writing the test for method_1() I would like to be sure that it calls some methods in an precise order.

I suppose that my production class A is stored in the following file class_a.py:

class A:
    __lock = None

    def __init__(self, lock):
        self.__lock = lock

    def method_1(self):
        self.__lock.acquire()
        self.__atomic_method_1()
        self.__lock.release()

    def __atomic_method_1(self):
        pass

The __lock attribute is an instance of class threading.Lock and is use to reach a thread safe execution of __atomic_method_1().

I need to write a unit test that checks the sequence of calling of the methods of the class A when it is invoked the method_1().

The check must verify that method_1() calls:

1. as first method: self.__lock.acquire()
2. after that it calls self.__atomic_method_1()
3. the last method called is self.__lock.release()

My need comes from wanting to make sure that the __atomic_method_1() method runs in a multithreaded context without being interrupted.

A useful hint but not enough

This is a very useful link, but it doesn't solve my problem. The solution provided by the link is perfect to verify the calling order of a sequence of functions invoked by an other function. The link shows an example where the function under test and the functions called are all contained in a file called module_under_test.
In my case, however, I have to verify the calling sequence of methods of a class and this difference prevents to use the solution provided by the link.

A trace that starts from the suggestion

However I have tried to develop the unit test referring to the link and in this way I have prepared a trace of the test file that I can show below:

import unittest
from unittest import mock
from class_a import A
import threading

class TestCallOrder(unittest.TestCase):
    def test_call_order(self):
       # create an instance of the SUT (note the parameter threading.Lock())
       sut = A(threading.Lock())

       # prepare the MOCK object
       source_mock = ...
       with patch(...)

           # prepare the expected values
           expected = [...]

           # run the code-under-test (method_1()).
           sut.method_1()

           # Check the order calling
           self.assertEqual(expected, source_mock.mock_calls)

if __name__ == '__main__':
    unittest.main()

But I'm not able to complete the method test_call_order().

Thanks

Answer 1

In the question's comments they said that it is not what unit tests are for. Yes, that makes for a brittle tests. But they serve a real use : do I correctly implement locking. You may want to refactor your class so that it is easier to test (and that would be another interesting question).

But if you really want to test it, as is, I have a solution for you.

What we need is to spy on the three methods self.__lock.acquire, self.__lock.release and self.__atomic_method_1. One way to do it is to wrap a Mock around them, and record the behavior. But just knowing they were called is not sufficient, you want the order between them. So you need multiple spies, which collectively log the actions that took place.

import unittest
from unittest import mock
from class_a import A
import threading

class TestCallOrder(unittest.TestCase):
    def test_call_order(self):
        sut = A(threading.Lock())

        # bypass the "name mangling" due to leading "__"
        sut_lock = sut._A__lock
        sut_method = sut._A__atomic_method_1

        # store what we observed during the test
        observed = []

        # prepare the side effects : they are simply observing the call, then forwarding it to the actual function
        def lock_acquire_side_effect():
            observed.append("lock acquired")
            sut_lock.acquire()
        def lock_release_side_effect():
            observed.append("lock released")
            sut_lock.release()
        def method_side_effect(*args, **kwargs):
            observed.append("method called")
            sut_method(*args, **kwargs)

        # prepare the spies, one on the lock, one on the method
        # (we could also mock the two Lock methods separately)
        lock_spy = mock.Mock(wraps=sut_lock, **{"acquire.side_effect": lock_acquire_side_effect,
                                                "release.side_effect": lock_release_side_effect})
        method_spy = mock.Mock(wraps=sut_method, **{"side_effect": method_side_effect})
        # they are wrapping the actual object, so method calls and attribute access are forwarded to it
        # but we also configure them, for certain calls, to use a special side effect (our spy functions)
        with mock.patch.object(sut, "_A__lock", lock_spy):
            with mock.patch.object(sut, "_A__atomic_method_1", method_spy):
            # we apply both spies (with Python 3.10 you can do it with only one `with`)
                sut.method_1()
                self.assertEqual(["lock acquired", "method called", "lock released"], observed)
                # and we check against a very nice ordered log of observations

if __name__ == '__main__':
    unittest.main()

---

Edit

To explain better what I did, here is a schema of how things are connected without mocks :

Your SUT has two references :

• one named __locked which points to its Lock instance, which itself has (for our concerns) 2 references : acquire and release
• the other named __atomic_method_1 which points to its A.__atomic_method_1 method

What we want is to observe the calls made to __lock.acquire, __lock.release, __atomic_method_1 and their relative order.

The simpler way to do that I could think of is to replace each of these three by "spy functions", which records they were being called (simply by appending in a list) then forward the call to the actual function.

But then we need these functions to be called, so we will have to mock things. Because they are not "importable", we can't mock.patch them. But we have the actual object we want to mock things of, and that is exactly what mock.patch.object is for ! While in the with, the sut.something will get replaced by a mock. So we need two mocks, one for the __atomic_method_1, the other for __lock.

As far as I can tell, we won't use __atomic_method_1 in any other way than calling it. So we just want our mock to call our spy method instead. To do that, we can configure it to call a function when it gets called, indicated by "side_effect".

But there are many other ways we can use our __lock besides acquire-ing and release-ing it. We don't know what the __aotmic_method_1 will do with it. So to be sure, we will set the mock to forward everything to the actual object, which means it wraps it.

Which gives us this :

The calls to __lock.acquire and __lock.release are sort of diverted (thanks to mocking) through our spy, while any other still gets through ordinarily.

_{(We could have done without creating a Mock for __aotmic_method_1, and mock.patch.object with the spy function)}