CODEWITHSUNDEEP
pythonmultithreadingthreadpoolpytest
Mark
Mark

Reputation: 2158

Pytest intermittenty failing using multiprocessing.pool.ThreadPool

I have the following function:

def parallel_ping(ip_addresses, deadline=1.0, interval=None, attempts=None, pool_size=20):
    try:
        pool = ThreadPool(processes=pool_size)
        args = ((ip, deadline, interval, attempts) for ip in ip_addresses)
        result = pool.map_async(ping_check, args)
        return result.get()
    finally:
        pool.close()

ping_check is a function that actually pings the devices (ip_addresses)

in my test, I have mocked out the ping_check function (replaced it with a function that returns the packet_loss).

sometimes when the test runs it raises an exception:

UnboundLocalError: local variable 'pool' referenced before assignment

My test function is inside a test_class that inherits from unittest.TestCase, and has a 

def setUp(self):
    self.old_function = ping_check

and a 

def tearDown(self):
    ping_check = self.old_function

function.

In the test function, I am assigning ping_check = mock_ping_check and then calling the parallel_ping function in there.

I am not sure why this exception is raised sometimes, and not others.

Upvotes: 1

Views: 1365

Answers (1)

Max Noel
Max Noel

Reputation: 8910

The exception is raised when your ThreadPool(processes=pool_size) fails. When that happens, an exception is raised instead of the pool being created, which means the pool variable is never assigned.

The finally block is then immediately executed, and tries to call pool.close(), which, since pool was never assigned, raises UnboundLocalError.

Move your pool = ThreadPool(pool_size) line out of the try block. It will have the added benefit of telling you exactly why the line fails.

Upvotes: 4

Related Questions