7

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?

1

2 Answers 2

4

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.

Sign up to request clarification or add additional context in comments.

2 Comments

If anyone has a better solution, please feel free to share. I'll accept my answer in a week or two if no one else leaves a better solution.
I think AssertionError, not AssertError, right?
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.

1 Comment

The import for the unittest-style should be pytest_reraiase not pytest-reraise

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.