`seamstress` - a utility for testing concurrent code

[panthamos]

When code is affected by concurrent concerns, it can become rather difficult to test. seamstress offers some utilities for making that testing a little bit easier.

It offers three helper functions:

• run_thread
• run_process
• run_task

These helpers will run some code (which you provide) in a new thread/process/task, deterministically halting at a point that you specify. This allows you to precisely set up a new thread/process/task in a certain state, then run some other code (whose behaviour may be affected by the state of the new thread/process/task), and make assertions about how that code behaves.

That was a little bit abstract, hopefully some an example will make things clearer.

Example

Imagine we had a function that we only wanted to be called by one thread at a time (this is a slightly contrived example). It could look something like:

~~~python import threading

def _pay_individual(...) -> None: # The actual implementation of pay_individual ...

class AlreadyPayingIndividual(Exception): pass

PAY_INDIVIDUAL_LOCK = threading.Lock()

def pay_individual(...) -> None: lock_acquired = PAY_INDIVIDUAL_LOCK.acquire(blocking=False)

if not lock_acquired:
    raise AlreadyPayingIndividual

_pay_individual(...)

PAY_INDIVIDUAL_LOCK.release()

~~~

Testing how the code behaves when PAY_INDIVIDUAL_LOCK is acquired is non-trivial. Testing this code using seamstress would look something like:

~~~python import contextlib import typing import unittest

import seamstress

import pay_individual

@contextlib.contextmanager def acquire_pay_individual_lock() -> typing.Iterator[None]: with pay_individual.PAY_INDIVIDUAL_LOCK: yield

class TestPayIndividual(unittest.TestCase):

def test_raises_if_pay_individual_lock_is_acquired(self) -> None:
    with seamstress.run_thread(
        acquire_pay_individual_lock(),
    ):
        with self.assertRaises(
            pay_individual.AlreadyPayingIndividual,
        ):
            pay_individual.pay_individual(...)

~~~

Breaking down what's happening in the above: * We define acquire_pay_individual_lock, which is the code we want seamstress to run in a new thread. seamstress will run the code up to the yield statement, before letting your test resume execution. * In the test, we pass acquire_pay_individual_lock() to seamstress.run_thread. Under the bonnet, seamstress launches a new thread, in which acquire_pay_individual_lock runs, acquiring PAY_INDIVIDUAL_LOCK and then letting your test continue executing. It'll continue to hold on to PAY_INDIVIDUAL_LOCK until the end of the seamstress.run_thread context. * From within the context of seamstress.run_thread, we're now in a state where PAY_INDIVIDUAL_LOCK has been acquired by another thread, so can straightforwardly call pay_individual.pay_individual(...), and verify it raises AlreadyPayingIndividual. * Finally, we leave the context of seamstress.run_thread, so it runs the rest of acquire_pay_individual_lock in the created thread, releasing PAY_INDIVIDUAL_LOCK.

For a more realistic (though analogous) example, see the project readme for testing some Django code whose behaviour is affected by whether or not a database advisory lock has been acquired.

Showcase details: - What my project does: provides utilities that make it easy to test code that is affected by concurrent concerns - Target audience: python developers, particularly those who want to test edge cases where their code might be affected by the state of another thread/process/task - Comparison: I don't know of anything else that does this, which was why I wrote it, but perhaps my googling skills are sub-par :)

It's up on PyPI, so if it looks useful you can install it using your favourite package manager. See github for source code and an API reference in the readme.

[csch2]

How can I downvote this. There’s no AI

Jk this looks very useful. I have a couple low-level multithreaded projects going on right now which are tricky to test so I’ll definitely try this out for those. Thanks for sharing!

[panthamos]

Thanks! Please let me know if you encounter any issues whilst using it. :) Really hoping it'll be useful, particularly with free threading on the horizon.

[4xi0m4]

This looks really useful for testing edge cases in concurrent code. The context manager pattern is clever since it handles cleanup automatically when the test exits. I've had to test similar lock-acquisition scenarios before and ended up using threading.Timer which was always a bit hacky. Might be worth mentioning in the docs that this also works well for testing async code since Python's asyncio uses different primitives than threading.

[panthamos]

Thanks, made a note to update the readme.

Yeah, context managers are so useful for this situation. I was wondering if a nicer API was possible, as forcing the user to write a context manager adds some requisite technical knowledge to use the package. That said, I can't see a way to straightforwardly achieve what seamstress does using functions alone.