import signal
|
import sys
|
|
import pytest
|
|
import trio
|
from .. import _core
|
from .._core.tests.tutil import (
|
ignore_coroutine_never_awaited_warnings,
|
create_asyncio_future_in_new_loop,
|
)
|
from .._util import (
|
signal_raise,
|
ConflictDetector,
|
is_main_thread,
|
coroutine_or_error,
|
generic_function,
|
Final,
|
NoPublicConstructor,
|
)
|
from ..testing import wait_all_tasks_blocked
|
|
|
def test_signal_raise():
|
record = []
|
|
def handler(signum, _):
|
record.append(signum)
|
|
old = signal.signal(signal.SIGFPE, handler)
|
try:
|
signal_raise(signal.SIGFPE)
|
finally:
|
signal.signal(signal.SIGFPE, old)
|
assert record == [signal.SIGFPE]
|
|
|
async def test_ConflictDetector():
|
ul1 = ConflictDetector("ul1")
|
ul2 = ConflictDetector("ul2")
|
|
with ul1:
|
with ul2:
|
print("ok")
|
|
with pytest.raises(_core.BusyResourceError) as excinfo:
|
with ul1:
|
with ul1:
|
pass # pragma: no cover
|
assert "ul1" in str(excinfo.value)
|
|
async def wait_with_ul1():
|
with ul1:
|
await wait_all_tasks_blocked()
|
|
with pytest.raises(_core.BusyResourceError) as excinfo:
|
async with _core.open_nursery() as nursery:
|
nursery.start_soon(wait_with_ul1)
|
nursery.start_soon(wait_with_ul1)
|
assert "ul1" in str(excinfo.value)
|
|
|
def test_module_metadata_is_fixed_up():
|
import trio
|
import trio.testing
|
|
assert trio.Cancelled.__module__ == "trio"
|
assert trio.open_nursery.__module__ == "trio"
|
assert trio.abc.Stream.__module__ == "trio.abc"
|
assert trio.lowlevel.wait_task_rescheduled.__module__ == "trio.lowlevel"
|
assert trio.testing.trio_test.__module__ == "trio.testing"
|
|
# Also check methods
|
assert trio.lowlevel.ParkingLot.__init__.__module__ == "trio.lowlevel"
|
assert trio.abc.Stream.send_all.__module__ == "trio.abc"
|
|
# And names
|
assert trio.Cancelled.__name__ == "Cancelled"
|
assert trio.Cancelled.__qualname__ == "Cancelled"
|
assert trio.abc.SendStream.send_all.__name__ == "send_all"
|
assert trio.abc.SendStream.send_all.__qualname__ == "SendStream.send_all"
|
assert trio.to_thread.__name__ == "trio.to_thread"
|
assert trio.to_thread.run_sync.__name__ == "run_sync"
|
assert trio.to_thread.run_sync.__qualname__ == "run_sync"
|
|
|
async def test_is_main_thread():
|
assert is_main_thread()
|
|
def not_main_thread():
|
assert not is_main_thread()
|
|
await trio.to_thread.run_sync(not_main_thread)
|
|
|
# @coroutine is deprecated since python 3.8, which is fine with us.
|
@pytest.mark.filterwarnings("ignore:.*@coroutine.*:DeprecationWarning")
|
def test_coroutine_or_error():
|
class Deferred:
|
"Just kidding"
|
|
with ignore_coroutine_never_awaited_warnings():
|
|
async def f(): # pragma: no cover
|
pass
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(f())
|
assert "expecting an async function" in str(excinfo.value)
|
|
import asyncio
|
|
if sys.version_info < (3, 11):
|
|
@asyncio.coroutine
|
def generator_based_coro(): # pragma: no cover
|
yield from asyncio.sleep(1)
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(generator_based_coro())
|
assert "asyncio" in str(excinfo.value)
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(create_asyncio_future_in_new_loop())
|
assert "asyncio" in str(excinfo.value)
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(create_asyncio_future_in_new_loop)
|
assert "asyncio" in str(excinfo.value)
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(Deferred())
|
assert "twisted" in str(excinfo.value)
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(lambda: Deferred())
|
assert "twisted" in str(excinfo.value)
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(len, [[1, 2, 3]])
|
|
assert "appears to be synchronous" in str(excinfo.value)
|
|
async def async_gen(arg): # pragma: no cover
|
yield
|
|
with pytest.raises(TypeError) as excinfo:
|
coroutine_or_error(async_gen, [0])
|
msg = "expected an async function but got an async generator"
|
assert msg in str(excinfo.value)
|
|
# Make sure no references are kept around to keep anything alive
|
del excinfo
|
|
|
def test_generic_function():
|
@generic_function
|
def test_func(arg):
|
"""Look, a docstring!"""
|
return arg
|
|
assert test_func is test_func[int] is test_func[int, str]
|
assert test_func(42) == test_func[int](42) == 42
|
assert test_func.__doc__ == "Look, a docstring!"
|
assert test_func.__qualname__ == "test_generic_function.<locals>.test_func"
|
assert test_func.__name__ == "test_func"
|
assert test_func.__module__ == __name__
|
|
|
def test_final_metaclass():
|
class FinalClass(metaclass=Final):
|
pass
|
|
with pytest.raises(TypeError):
|
|
class SubClass(FinalClass):
|
pass
|
|
|
def test_no_public_constructor_metaclass():
|
class SpecialClass(metaclass=NoPublicConstructor):
|
pass
|
|
with pytest.raises(TypeError):
|
SpecialClass()
|
|
with pytest.raises(TypeError):
|
|
class SubClass(SpecialClass):
|
pass
|
|
# Private constructor should not raise
|
assert isinstance(SpecialClass._create(), SpecialClass)
|
