import os import sys import pytest from .. import ( current_async_library, AsyncLibraryNotFoundError, current_async_library_cvar, thread_local ) def test_basics_cvar(): with pytest.raises(AsyncLibraryNotFoundError): current_async_library() token = current_async_library_cvar.set("generic-lib") try: assert current_async_library() == "generic-lib" finally: current_async_library_cvar.reset(token) with pytest.raises(AsyncLibraryNotFoundError): current_async_library() def test_basics_tlocal(): with pytest.raises(AsyncLibraryNotFoundError): current_async_library() old_name, thread_local.name = thread_local.name, "generic-lib" try: assert current_async_library() == "generic-lib" finally: thread_local.name = old_name with pytest.raises(AsyncLibraryNotFoundError): current_async_library() def test_asyncio(): import asyncio with pytest.raises(AsyncLibraryNotFoundError): current_async_library() ran = [] async def this_is_asyncio(): assert current_async_library() == "asyncio" # Call it a second time to exercise the caching logic assert current_async_library() == "asyncio" ran.append(True) asyncio.run(this_is_asyncio()) assert ran == [True] with pytest.raises(AsyncLibraryNotFoundError): current_async_library() @pytest.mark.skipif( sys.version_info >= (3, 12), reason= "curio broken on 3.12 (https://github.com/python-trio/sniffio/pull/42)", ) def test_curio(): import curio with pytest.raises(AsyncLibraryNotFoundError): current_async_library() ran = [] async def this_is_curio(): assert current_async_library() == "curio" # Call it a second time to exercise the caching logic assert current_async_library() == "curio" ran.append(True) curio.run(this_is_curio) assert ran == [True] with pytest.raises(AsyncLibraryNotFoundError): current_async_library()