I have a race condition in a unittest I'm trying to fix.
Suppose there's a module spam.py
:
import threading
def foo(*args, **kwargs):
pass
def bar():
t = threading.Timer(0.5, foo, args=('potato',), kwargs={'x': 69, 'y':'spam'})
t.start()
And here is the test for it:
from mock import patch
from spam import bar
from unittest import TestCase
class SpamTest(TestCase):
def test_bar(self):
with patch('spam.foo') as mock:
bar()
mock.assert_called_once_with('potato', y='spam', x=69)
Of course this test fails with AssertionError: Expected to be called once. Called 0 times.
because the call to bar()
is non-blocking, so the assertion happens too early.
The test can be made to pass by putting a time.sleep(1)
in before the assert, but this is obviously hacky and lame - what is the accepted way to mock / unittest asynchronous stuff?
How about modifying the bar
to return a thead object:
def bar():
t = threading.Timer(0.5, foo, args=('potato',), kwargs={'x': 69, 'y':'spam'})
t.start()
return t # <----
Then, join the thread in the test code:
class SpamTest(TestCase):
def test_bar(self):
with patch('spam.foo') as mock:
t = bar()
t.join() # <----
mock.assert_called_once_with('potato', y='spam', x=69)
UPDATE alternative that does not require bar
change.
import threading
import time
...
class SpamTest(TestCase):
def test_bar(self):
foo = threading.Event()
with patch('spam.foo', side_effect=lambda *args, **kwargs: foo.set()) as mock:
# Make the callback `foo` to be called immediately
with patch.object(threading._Event, 'wait', time.sleep(0.000001)):
bar()
foo.wait() # Wait until `spam.foo` is called. (instead of time.sleep)
mock.assert_called_once_with('potato', y='spam', x=69)
UPDATE
In Python 3.x, patch threading.Event
instead of threading._Event
.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With