PyTest test-suite passing when asserts should fail when testing another thread's results in Python?

Question

I'm currently using pytest to test an existing (unittest test-suite per the documentation). I'm currently writing a thread that waits for an IP address to be assigned and then returns it to a callback function, and I'm writing unittests to accompany it.

Here's the Test Case class I wrote.

class TestGetIpAddressOnNewThread(unittest.TestCase):

    def test_get_existing_ip(self):
        def func(ip):
            assert ip == "192.168.0.1" # Not the real IP
            # Even when I introduce an assert statement that should fail, test still passes
            assert ip == "shouldn't be the ip" 

        ip_check = GetInstanceIpThread(instance, func)
        ip_check.start()
        ip_check.join()

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

And here's the GetInstanceIpThread pseudo-definition:

class GetInstanceIpThread(threading.Thread):
    def __init__(self, instance, callback):
        threading.Thread.__init__(self)
        self.event = threading.Event()
        self.instance = instance
        self.callback = callback

    def run(self):
        self.instance.wait_until_running()

        ip = self.instance.ip_address
        self.callback(ip)

When I run this test case using pytest path_to_file.py::TestGetIpAddressOnNewThread, it passes (yay!) but even when I introduce assert statements that should 100% fail (boo!). What's going wrong, how do I write tests that actually fail?

Answer 1

So I'm answering my own question because while I was able to find the answer on StackOverflow, it wasn't under any useful keywords, as most answers were talking about how to multithread testing using pytest-xdist, not testing multithreading. I ended up using pytest -s path_to_file.py::TestGetIpAddressOnNewThread as mentioned here during debugging that revealed I had a typo-caused exception that was being printed but not causing the test to fail.

That led me to this question, that I had originally dismissed as I didn't realize that asserts were just raising AssertError's.

As a result, I adapted the community wiki answer as below in order to get the asserts to work correctly!

class GetInstanceIpThread(threading.Thread):
    def __init__(self, instance, callback=None):
        threading.Thread.__init__(self)
        self.event = threading.Event()
        self.instance = instance
        self.callback = callback

    def _run(self):
        # The original run code goes here
        self.instance.wait_until_running()

        ip = self.instance.ip_address
        self.callback(ip)

    def run(self):
        self.exc = None
        try:
            self._run()
        except BaseException as e:
            self.exc = e

    def join(self):
        super(GetInstanceIpThread, self).join()
        if self.exc:
            raise self.exc

Note that this will fail your tests as long as there's any exception in the other thread. This may not be what you want, so if you only want to fail if an assert fails or similar, you can change BaseException to AssertError (or whatever you want to fail on) instead.

Overriding join() is necessary, as you must raise the exception back on pytest's main thread in order for it to properly fail the tests.