text: import signal
  import pytest
  
  import trio
  from .. import _core
  from .._core.tests.tutil import ignore_coroutine_never_awaited_warnings
  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
  
          @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(asyncio.Future())
          assert "asyncio" in str(excinfo.value)
  
          with pytest.raises(TypeError) as excinfo:
              coroutine_or_error(lambda: asyncio.Future())
          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)
  