Karen on Python mocks: threading.Timer

Meet Karen, a software engineer. A coder. A Pythonista. Well, among other things – because she has a life.

She loves her profession, and she loves coding. While working, she does quite a few things: she designs and architects; thinks and plans; she has discussions with her peers. She codes. Oh, she codes! And she writes tests – lots of them. She does so because she believes quality is a must.

Part of her job, too, is paying some technical debt left over by coworkers and previous coders who were not as cautious as they should have been or were in a hurry.

This time, short and nasty time.sleep calls in a good number of tests catch her attention. She quickly identifies them as dirty workarounds for unit tests involving threading.Timer and decides to fix them, thinking to herself that writing tests for asynchronous things can, indeed, be a bit tricky.

Karen starts by understanding what the unit is about. The filename is hellotimer.py.

from threading import Timer


def hello(name):
    print(f"Hello, {name}!")


def set_timer(name):
    timer = Timer(1.0, hello, [name])
    timer.start()
    timer.join()

“There’s nothing too special about it,” she mumbles and glances at one of the tests.

import time
from unittest.mock import patch

import hellotimer


class HellotimerTestCase(unittest.TestCase):
    # ...
    def test_set_timer_with_sleep(self):
        hellotimer.set_timer("Neo")
        time.sleep(1)
        self.hello_mock.assert_called_once_with("Neo")

She thinks for a moment and opens threading’s documentation chapter on docs.python.org where she spots a particular segment that details threading.Timer’s workings.

class threading.Timer(interval, function, args=None, kwargs=None)

Create a timer that will run function with arguments args and keyword arguments kwargs, after interval seconds have passed. If args is None (the default) then an empty list will be used. If kwargs is None (the default) then an empty dict will be used.

She even skims over threading.Timer’s source code to make sure she fully understands the matter at hand.

class Timer(Thread):
    # [snip]
    def __init__(self, interval, function, args=None, kwargs=None):
        Thread.__init__(self)
        self.interval = interval
        self.function = function
        self.args = args if args is not None else []
        self.kwargs = kwargs if kwargs is not None else {}
        self.finished = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet."""
        self.finished.set()

    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()

Then switches back to the unit.

from threading import Timer


def hello(name):
    print(f"Hello, {name}!")


def set_timer(name):
    timer = Timer(1.0, hello, [name])
    timer.start()
    timer.join()

Paper, pencil, and thoughts. “I need to emulate the behavior of threading.Timer. To achieve that, I should set it to a mock object. It has to call the passed function, along with the arguments, and then return another mock object just to allow start and join to be called afterwards”.

We know what Karen is doing here. It’s easy and tempting to ignore the importance of a proper analysis when tackling a problem. Mileages may vary, but jumping right into code tends more to result in a solution that is not optimal, over-complicated, and hard to maintain. Give yourself time to think and design your solution before sitting down to write code.

Scribbling down her thoughts, she decides to patch threading.Timer by creating a mock and setting its side_effect attribute, which gives her more flexibility when mocking. It can be used to simulate exceptions being raised, for example. It can also be used to define different return values for multiple calls to the mock. In this case, Karen wants to both control the behavior and define the return value.

def test_set_timer_with_patch(self):
    def side_effect(interval, function, args=None, kwargs=None):
        args = args if args is not None else []
        kwargs = kwargs if kwargs is not None else {}
        function(*args, **kwargs)
        return Mock()

    with patch("hellotimer.Timer", side_effect=side_effect) as timer_mock:
        hellotimer.set_timer("Neo")
        self.hello_mock.assert_called_once_with("Neo")
        timer_mock.assert_called_once_with(1.0, hellotimer.hello, ["Neo"])

“There it is, first test fixed! Let me fix all of them,” she celebrates in her thoughts, sparingly. “But I’m not going to repeat myself over and over. I must create a decorator and use it in all the other tests!” Reflections on what seems to be a better approach.

In fact, not repeating yourself (DRY) is one of the fundamental principles of programming. Recognizing duplication and learning how to eliminate it leads to much cleaner code. Repeating – or copy-pasting – portions of code throughout an application is indeed a bad practice. Repetition increases waste, inflates the amount of code to be maintained, and, therefore, the number of potential bugs. It adds unnecessary complexity to the application and makes it more difficult to understand and change. Not to mention, the smallest of changes may turn into a shotgun surgery. Karen uses a decorator to her advantage.

from functools import wraps


def patch_hellotimer_timer(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        def side_effect(interval, function, args=None, kwargs=None):
            args = args if args is not None else []
            kwargs = kwargs if kwargs is not None else {}
            function(*args, **kwargs)
            return Mock()

        with patch("hellotimer.Timer", side_effect=side_effect) as timer_mock:
            return f(*(*args, timer_mock), **kwargs)

    return wrapper

Satisfied with it, Karen removes the side_effect function and the with expression that defines time_patch from the unit test and uses the new decorator instead. She applies it to the remaining test cases, feeling great about deleting all calls to time.sleep.

@patch_hellotimer_timer
def test_set_timer_with_decorator_intermediate(self, timer_mock):
    hellotimer.set_timer("Neo")
    self.hello_mock.assert_called_once_with("Neo")
    timer_mock.assert_called_once_with(1.0, hellotimer.hello, ["Neo"])

That’s when another thought strikes her. “This mock may be useful in other situations! What if I define a function that creates that decorator, and specify which target I want to patch as an argument? That should work!” She nails it:

def patch_threading_timer(target_timer):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            def side_effect(interval, function, args=None, kwargs=None):
                args = args if args is not None else []
                kwargs = kwargs if kwargs is not None else {}
                function(*args, **kwargs)
                return Mock()

            with patch(target_timer, side_effect=side_effect) as timer_mock:
                return f(*(*args, timer_mock), **kwargs)

        return wrapper

    return decorator

And marvels at how easy it is to use:

@patch_threading_timer("hellotimer.Timer")
def test_set_timer_with_decorator_final(self, timer_mock):
    hellotimer.set_timer("Neo")
    self.hello_mock.assert_called_once_with("Neo")
    timer_mock.assert_called_once_with(1.0, hellotimer.hello, ["Neo"])

Caring about documentation and mindful of good software practices, she describes it and adds comments to a few essential lines. They serve not only her coworkers but her future self too.

def patch_threading_timer(target_timer):
    """patch_threading_timer acts similarly to unittest.mock.patch as a
    function decorator, but specifically for threading.Timer. The function
    passed to threading.Timer is called right away with all given arguments.

    :arg str target_timer: the target Timer (threading.Timer) to be patched
    """

    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            def side_effect(interval, function, args=None, kwargs=None):
                args = args if args is not None else []
                kwargs = kwargs if kwargs is not None else {}
                # Call whatever would be called when interval is reached
                function(*args, **kwargs)
                # Return a mock object to allow function calls on the
                # returned value
                return Mock()

            with patch(target_timer, side_effect=side_effect) as timer_mock:
                # Pass the mock object to the decorated function for further
                # assertions
                return f(*(*args, timer_mock), **kwargs)

        return wrapper

    return decorator

Happy with the solution, she creates a commit. It contains a succinct and sensible subject line with a couple more paragraphs detailing the reasons behind the change and how it addresses the issue and the effects it has, especially the fact that tests now run twice as fast. She submits a pull request for review and adds it as one of the matters for the next standup, when she’ll let her team know she is available to make any detail clearer.

Karen goes for a short break and her last coffee of the day. The skyline shines in beautiful shades of purplish-red while a flock of mockingbirds screeches aloft in the wind. Marvelled, she doesn’t notice her mobile pulse as a teammate approves her pull request.

For a working implementation of what was discussed above, please check the following repl: