5

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?

Rob Rose
  • 1,806
  • 22
  • 41
  • Does this answer your question? [pytest failure in a separate thread](https://stackoverflow.com/questions/43791360/pytest-failure-in-a-separate-thread) – Mark Apr 28 '22 at 13:05

2 Answers2

3

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.

Rob Rose
  • 1,806
  • 22
  • 41
2

I ran into the same problem but without having access to the code that creates the thread. I published a little test helper package for that usecase, pytest-reraise:

pip install pytest-reraise

Pytest-style test case:

def test_get_existing_ip(reraise):
    def func(ip):
        with reraise:
            assert ip == "192.168.0.1" # Not the real IP
            assert ip == "shouldn't be the ip" 

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

Unittest-style test case:

from pytest-reraise import Reraise

class TestGetIpAddressOnNewThread(unittest.TestCase):

    def test_get_existing_ip(self):
        reraise = Reraise()

        def func(ip):
            with reraise:
                assert ip == "192.168.0.1" # Not the real IP
                assert ip == "shouldn't be the ip" 

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

        # Re-raise the first exception that the `reraise` context manager captured:
        reraise()

Both test cases fail as expected and report the AssertionError.

bjoluc
  • 51
  • 3