1 22
import signal
2 22
import pytest
3

4 22
import trio
5 22
from .. import _core
6 22
from .._core.tests.tutil import ignore_coroutine_never_awaited_warnings
7 22
from .._util import (
8
    signal_raise,
9
    ConflictDetector,
10
    is_main_thread,
11
    coroutine_or_error,
12
    generic_function,
13
    Final,
14
    NoPublicConstructor,
15
    SubclassingDeprecatedIn_v0_15_0,
16
)
17 22
from ..testing import wait_all_tasks_blocked
18

19

20 22
def test_signal_raise():
21 22
    record = []
22

23 22
    def handler(signum, _):
24 22
        record.append(signum)
25

26 22
    old = signal.signal(signal.SIGFPE, handler)
27 22
    try:
28 22
        signal_raise(signal.SIGFPE)
29
    finally:
30 22
        signal.signal(signal.SIGFPE, old)
31 22
    assert record == [signal.SIGFPE]
32

33

34 22
async def test_ConflictDetector():
35 22
    ul1 = ConflictDetector("ul1")
36 22
    ul2 = ConflictDetector("ul2")
37

38 22
    with ul1:
39 22
        with ul2:
40 22
            print("ok")
41

42 22
    with pytest.raises(_core.BusyResourceError) as excinfo:
43 22
        with ul1:
44 22
            with ul1:
45
                pass  # pragma: no cover
46 22
    assert "ul1" in str(excinfo.value)
47

48 22
    async def wait_with_ul1():
49 22
        with ul1:
50 22
            await wait_all_tasks_blocked()
51

52 22
    with pytest.raises(_core.BusyResourceError) as excinfo:
53 22
        async with _core.open_nursery() as nursery:
54 22
            nursery.start_soon(wait_with_ul1)
55 22
            nursery.start_soon(wait_with_ul1)
56 22
    assert "ul1" in str(excinfo.value)
57

58

59 22
def test_module_metadata_is_fixed_up():
60 22
    import trio
61 22
    import trio.testing
62

63 22
    assert trio.Cancelled.__module__ == "trio"
64 22
    assert trio.open_nursery.__module__ == "trio"
65 22
    assert trio.abc.Stream.__module__ == "trio.abc"
66 22
    assert trio.lowlevel.wait_task_rescheduled.__module__ == "trio.lowlevel"
67 22
    assert trio.testing.trio_test.__module__ == "trio.testing"
68

69
    # Also check methods
70 22
    assert trio.lowlevel.ParkingLot.__init__.__module__ == "trio.lowlevel"
71 22
    assert trio.abc.Stream.send_all.__module__ == "trio.abc"
72

73
    # And names
74 22
    assert trio.Cancelled.__name__ == "Cancelled"
75 22
    assert trio.Cancelled.__qualname__ == "Cancelled"
76 22
    assert trio.abc.SendStream.send_all.__name__ == "send_all"
77 22
    assert trio.abc.SendStream.send_all.__qualname__ == "SendStream.send_all"
78 22
    assert trio.to_thread.__name__ == "trio.to_thread"
79 22
    assert trio.to_thread.run_sync.__name__ == "run_sync"
80 22
    assert trio.to_thread.run_sync.__qualname__ == "run_sync"
81

82

83 22
async def test_is_main_thread():
84 22
    assert is_main_thread()
85

86 22
    def not_main_thread():
87 22
        assert not is_main_thread()
88

89 22
    await trio.to_thread.run_sync(not_main_thread)
90

91

92
# @coroutine is deprecated since python 3.8, which is fine with us.
93 22
@pytest.mark.filterwarnings("ignore:.*@coroutine.*:DeprecationWarning")
94 9
def test_coroutine_or_error():
95 22
    class Deferred:
96
        "Just kidding"
97

98 22
    with ignore_coroutine_never_awaited_warnings():
99

100
        async def f():  # pragma: no cover
101
            pass
102

103 22
        with pytest.raises(TypeError) as excinfo:
104 22
            coroutine_or_error(f())
105 22
        assert "expecting an async function" in str(excinfo.value)
106

107 22
        import asyncio
108

109 22
        @asyncio.coroutine
110
        def generator_based_coro():  # pragma: no cover
111
            yield from asyncio.sleep(1)
112

113 22
        with pytest.raises(TypeError) as excinfo:
114 22
            coroutine_or_error(generator_based_coro())
115 22
        assert "asyncio" in str(excinfo.value)
116

117 22
        with pytest.raises(TypeError) as excinfo:
118 22
            coroutine_or_error(asyncio.Future())
119 22
        assert "asyncio" in str(excinfo.value)
120

121 22
        with pytest.raises(TypeError) as excinfo:
122 22
            coroutine_or_error(lambda: asyncio.Future())
123 22
        assert "asyncio" in str(excinfo.value)
124

125 22
        with pytest.raises(TypeError) as excinfo:
126 22
            coroutine_or_error(Deferred())
127 22
        assert "twisted" in str(excinfo.value)
128

129 22
        with pytest.raises(TypeError) as excinfo:
130 22
            coroutine_or_error(lambda: Deferred())
131 22
        assert "twisted" in str(excinfo.value)
132

133 22
        with pytest.raises(TypeError) as excinfo:
134 22
            coroutine_or_error(len, [[1, 2, 3]])
135

136 22
        assert "appears to be synchronous" in str(excinfo.value)
137

138
        async def async_gen(arg):  # pragma: no cover
139
            yield
140

141 22
        with pytest.raises(TypeError) as excinfo:
142 22
            coroutine_or_error(async_gen, [0])
143 22
        msg = "expected an async function but got an async generator"
144 22
        assert msg in str(excinfo.value)
145

146
        # Make sure no references are kept around to keep anything alive
147 22
        del excinfo
148

149

150 22
def test_generic_function():
151 22
    @generic_function
152 9
    def test_func(arg):
153
        """Look, a docstring!"""
154 22
        return arg
155

156 22
    assert test_func is test_func[int] is test_func[int, str]
157 22
    assert test_func(42) == test_func[int](42) == 42
158 22
    assert test_func.__doc__ == "Look, a docstring!"
159 22
    assert test_func.__qualname__ == "test_generic_function.<locals>.test_func"
160 22
    assert test_func.__name__ == "test_func"
161 22
    assert test_func.__module__ == __name__
162

163

164 22
def test_final_metaclass():
165 22
    class FinalClass(metaclass=Final):
166 22
        pass
167

168 22
    with pytest.raises(TypeError):
169

170 22
        class SubClass(FinalClass):
171 22
            pass
172

173

174 22
def test_subclassing_deprecated_metaclass():
175 22
    class Blah(metaclass=SubclassingDeprecatedIn_v0_15_0):
176 22
        pass
177

178 22
    with pytest.warns(trio.TrioDeprecationWarning):
179

180 22
        class Blah2(Blah):
181 22
            pass
182

183

184 22
def test_no_public_constructor_metaclass():
185 22
    class SpecialClass(metaclass=NoPublicConstructor):
186 22
        pass
187

188 22
    with pytest.raises(TypeError):
189 22
        SpecialClass()
190

191 22
    with pytest.raises(TypeError):
192

193 22
        class SubClass(SpecialClass):
194 22
            pass
195

196
    # Private constructor should not raise
197 22
    assert isinstance(SpecialClass._create(), SpecialClass)

Read our documentation on viewing source code .

Loading