ClockedTestCase not working maybe

[ja8zyjits]

The problem

I am trying to use the ClockedTestCase. Iam trying to fast forward the asyncio.sleep in my code while testing. But iam not able to make it work.

script.py

import asyncio

class WorkingModel: def init(self, interval=10): self._interval = interval self._status = None self._statrted = False

async def start(self):
    self._statrted = True
    self._sync = asyncio.ensure_future(self.sync_function())

async def sync_function(self):
    while self._statrted:
        await asyncio.sleep(self._interval)
        self._status = 1

async def stop(self):
    self._statrted = False
    await self._sync

> test_script.py

```python
import asynctest
import asyncio
import unittest
import script

class MainTest(asynctest.ClockedTestCase):
    async def test_main(self):
        working_model = script.WorkingModel()
        working_model.start()
        await self.advance(10)
        self.assertEqual(working_model._status, 1)
        working_model.stop()

if __name__=="__main__":
    unittest.main()

When iam running the tests with python -m unittest the script just hangs.

(asynctest_issue_virtual) jzy@jzy:~/work/asynctest_issue_virtual/scripts$ python -m unittest

Nothing much happens

while the normal asynctest.TestCase works but it takes time like 20 seconds (10+10)

test_script.py

import asynctest
import asyncio
import unittest
import script

class MainTest(asynctest.ClockedTestCase):

class MainTest(asynctest.TestCase): async def test_main(self): working_model = script.WorkingModel() await working_model.start()

await self.advance(10)

    await asyncio.sleep(11)
    self.assertEqual(working_model._status, 1)
    await working_model.stop()

if name=="main": unittest.main()

When iam running the tests with `python -m unittest` the script it executes after 20 seconds.

```shell
(asynctest_issue_virtual) jzy@jzy:~/work/asynctest_issue_virtual/scripts$ python -m unittest
.
----------------------------------------------------------------------
Ran 1 test in 20.022s

OK

Support information

• python version 3.6.7
• asynctest version

  Name: asynctest
  Version: 0.12.4
  Summary: Enhance the standard unittest package with features for testing asyncio libraries
  Home-page: https://github.com/Martiusweb/asynctest/
  Author: Martin Richard
  Author-email: martius@martiusweb.net
  License: Apache 2
  Location: /home/jzy/work/asynctest_issue_virtual/lib/python3.6/site-packages

It would be really nice if I could make the self.advance feature work.

[Martiusweb]

Hi,

In your example, there is a race condition between the moment stop() is called and a second iteration of the loop in sync_function() runs:

import asyncio
import asynctest

class WorkingModel:
    def __init__(self, interval=10):
        self._interval = interval
        self._status = None
        self._statrted = False

    async def start(self):
        self._statrted = True
        self._sync = asyncio.ensure_future(self.sync_function())
        print("started")

    async def sync_function(self):
        while self._statrted:
            print(f"going to sleep for {self._interval}")
            await asyncio.sleep(self._interval)
            self._status = 1

    async def stop(self):
        self._statrted = False
        print("stop called, self._statrted set to False")
        print("going to await the bg task self._sync")
        await self._sync
        print("awaited the bg task self._sync")

class MainTest(asynctest.ClockedTestCase):
    async def test_main(self):
        working_model = WorkingModel()
        await working_model.start()
        print("going to advance")
        await self.advance(10)
        print("advanced of 10")
        self.assertEqual(working_model._status, 1)
        await working_model.stop()

"""
output is:

started
going to advance
going to sleep for 10
going to sleep for 10
advanced of 10
stop called, self._statrted set to False
going to await the bg task self._sync
"""

In the output, we can see that a second sleep() is called before _stop is called, however, you never advance the clock anymore, it just waits for this sleep.

When using ClockedTestCase, the clock never moves unless you call advance().

you must thus call self.advance(10) once again after calling working_model.stop(). However, since there's yet another deadlock because stop waits for the second sleep() to finish, it must be executed concurrently.

you can do something like:

    advance = asyncio.ensure_future(self.advance(10))
    await working_model.stop()
    await advance

But there's still a theoretical race condition between stop() and this second advance().

I'm closing the issue since ClockedTestCase works as expected. If you believe we should change something, feel free to comment and reopen the issue.

[ja8zyjits]

@Martiusweb Thanks for your prompt response. May be we need a bit more descriptive documentation regarding the utility of self.advance and its use cases.

[Martiusweb]

True, I added a notice in the doc and tutorial.