Skip to content

Commit

Permalink
trio_test_case: throw an exception if we forget to await a coroutine
Browse files Browse the repository at this point in the history
Forgetting to await a coroutine may make a test do essentially
nothing.  We want to make sure that doesn't happen.

Probably raise_unraisables should be done automatically by
unittest.TestCase.

Ideally, forgetting an await would be an error at the Python level,
see also python-trio/pytest-trio#86
  • Loading branch information
catern committed Feb 7, 2022
1 parent 886e708 commit feffb1d
Showing 1 changed file with 29 additions and 1 deletion.
30 changes: 29 additions & 1 deletion python/rsyscall/tests/trio_test_case.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,25 @@
"A trio-enabled variant of unittest.TestCase"
import trio
import unittest
import contextlib
import functools
import sys
import types
import warnings
from trio._core._run import Nursery
from rsyscall import local_process, Process

@contextlib.contextmanager
def raise_unraisables():
unraisables = []
try:
orig_unraisablehook, sys.unraisablehook = sys.unraisablehook, unraisables.append
yield
finally:
sys.unraisablehook = orig_unraisablehook
if unraisables:
raise trio.MultiError([unr.exc_value for unr in unraisables])

class TrioTestCase(unittest.TestCase):
"A trio-enabled variant of unittest.TestCase"
nursery: Nursery
Expand Down Expand Up @@ -43,7 +57,21 @@ async def test_with_setup() -> None:
nursery.cancel_scope.cancel()
@functools.wraps(test_with_setup)
def sync_test_with_setup(self) -> None:
trio.run(test_with_setup)
# Throw an exception if there were any "coroutine was never awaited" warnings, to fail the test.
# See https://github.com/python-trio/pytest-trio/issues/86
# We also need raise_unraisables, otherwise the exception is suppressed, since it's in __del__
with raise_unraisables():
# Restore the old warning filter after the test.
with warnings.catch_warnings():
warnings.filterwarnings('error', message='.*was never awaited', category=RuntimeWarning)
trio.run(test_with_setup)
setattr(self, methodName, types.MethodType(sync_test_with_setup, self))
super().__init__(methodName)

class Test(unittest.TestCase):
def test_coro_warning(self) -> None:
class Test(TrioTestCase):
async def test(self):
trio.sleep(0)
with self.assertRaises(RuntimeWarning):
Test('test').test()

0 comments on commit feffb1d

Please sign in to comment.