I'm running multiple pytest tests.
I have 1 specific test case, which works fine when executed alone, but when large amount of previous tests are being executed, this particular test gets stuck when running run_until_complete(future)
. It gets stuck, but the timeout is working, so after 60 seconds test progresses further. But when no previous tests are being executed, this coroutine is progressed "normally" - without getting to the 60s timeout.
Basically code is more or less this one:
def test_asdf():
# // some code
loop = asyncio.get_event_loop()
job = arq.jobs.Job("job_001", self.redis, "some_job_name") # Arq job
coro = job.result(timeout=10) # wait for arq job to finish.
task = loop.create_task(coro)
future_result = asyncio.wait_for(task, 60) # wait for task with timeout 60 seconds
loop.run_until_complete(future_result)
# /// some code
I see that all the tests share same event loop - I've checked it using:
loop = asyncio.get_event_loop()
id(loop)
So my guess is that previous tests have some futures that are "blocking" event loop, so the current test cannot wait_until_complete
particular Future, as it cannot be progressed for some reason.
My questions:
- How can I list all pending Futures inside event loop when executing pytest. If I knew how many Futures were being executed I could maybe find a solution.
- Could there be something I've missed? I'm using
arq
library, maybe it has some known issues / bugs related to event loops?
I've tried using debug mode, but it didn't seemed to work during test execution. I've also tried to print all tasks like this print(asyncio.all_tasks())
but during tests this command was not working (but it works during "normal" application execution).
event-loop
fixture with a function scope as default scope setting. It means a fresh loop for each test - previous test cannot be blamed for failures. But if your main question is how to debug what kind of "left-overs" make that particular test test failing, I can't help.