pyppin.testing.turn_taker

[View Source]

A tool to simplify unittesting operations that involve many threads.

Classes

TurnTaker(game)

A tool to manage "turn taking" in multithreaded unittests.

class pyppin.testing.turn_taker.TurnTaker(game: _Game)

Bases: object

A tool to manage “turn taking” in multithreaded unittests.

Testing code where you want to verify that actors in multiple threads interact with a shared system in an expected way can be hard. Black-box testing (“the right outcome happened”) can miss subtle race conditions that only happen sometimes, but white-box testing (“the system worked the way my mental model does”) is tricky. This class is designed to make that easier.

Imagine that the threads of your unittest are playing a game with a ball. If one thread wants another thread to do something, they ask that thread to do it, and then pass them the ball. When that thread is done, they pass the ball back to the first player. Catching the ball lets the first player know that the thing should now be done, and they can check what’s going on in the world. By passing the ball back and forth, all the threads can take turns doing things.

The easiest way to illustrate how it works is with an example:

def testSomething(self) -> None:
    class Player1(TurnTaker):
        def run(self) -> None:
            # This thread will go first. It does some initial prep work, then
            self.pass_and_wait('Player2')

            # Now it waits until player2 passes back to it, and it does some other things.

            # Finally, it passes back to player2 and doesn't wait for anything else.
            self.pass_and_finish('Player2')

    class Player2(TurnTaker):
        def run(self) -> None:
            # This thread will wait until it gets passed the ball by player 1, then
            # does some initial stuff, and passes back to player 1.

            self.pass_and_wait('Player1')

            # Player 1 will finish up and then pass back to us. We wrap up and don't need
            # to pass to anyone else; when we finish, nobody else is waiting!

    # The unittest then runs the game as follows:
    TestTaker.play(Player1, Player2, first_player=Player1)

The call to play starts up threads for both players, but Player 2 doesn’t actually start yet – only the first player gets to execute. They do some work and then pass_and_wait() to player 2. This blocks player 1 and unblocks player 2, who now does some other stuff and then calls pass_and_wait() to send control back to player 1. Player 1 does some final checking and then calls pass_and_finish(), signalling that it isn’t waiting for the ball to show up anymore. Since nobody is waiting for any more balls, the game ends, and the unittest has passed!

Importantly, any exceptions (including assertion failures) raised by one of the players will be raised by the call to play(). This is important, because Python doesn’t usually propagate exceptions across threads; this way, all your players can make assertions in the ordinary unittest style and know that they’ll lead to test failures in the usual way, too.

You can find several examples of how to use this in pyppin’s own unittests.)

exception PlayerExitedWithoutPassing

Bases: Exception

Raised if a player exited without passing, while other players were waiting.

This usually means a bug in your unittest!

exception PlayerActedWhenNotTheirTurn

Bases: Exception

Raised if a player acts when it isn’t their turn.

run() → None

The basic method which subclasses implement: Actually do this thread’s job in the test!

This method will be called during this player’s first turn, and can control the flow of the game by calling various instance methods.

pass_and_wait(to: Union[str, Type[TurnTaker]]) → None

Pass the ball to another player and wait for my next turn.

pass_and_finish(to: Union[str, Type[TurnTaker]]) → None

Pass the ball to another player and don’t wait for your next turn; you’re done.

Every player, except the last one, must call this immediately before returning from their run method.

pass_without_waiting(to: Union[str, Type[TurnTaker]]) → None

Pass the ball without waiting, but not necessarily finishing my function.

If you call this, you had better call wait_for_my_turn() before trying to do anything else with the ball! This function is primarily useful when you need to start a blocking operation, like trying to grab a mutex that you know won’t be available until another player does something.

wait_for_my_turn() → None

Block until it’s my next turn.

You usually don’t need to call this explicitly, unless you called pass_without_waiting. A player’s run method isn’t called until their first turn, and pass_and_wait calls this internally.

static play(*players: Type[TurnTaker], first_player: Optional[Union[str, Type[TurnTaker]]] = None, final_timeout: Optional[float] = 5) → None

The main function you call to play a game.

Parameters

• players – The set of player classes that you want to play. During the game, you can refer to any player by the name of the class.

• first_player – Who goes first. If not given, the first player in the list does.

• final_timeout – How long, in seconds, to wait for the game to finish before erroring out.

Raises

• KeyError – If first_player is not one of the players.

• TurnTaker.PlayerExitedWithoutPassing – If some player returned from their run method without passing while other players were still waiting.

• TurnTaker.PlayerActedWhenNotTheirTurn – What it says on the tin.

• TimeoutError – If final_timeout expires while there are active threads.

• Any other exception – If raised by the players themselves!

property name: str

Return the name of the player, which is the same as the name of the class.