| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 | # (c) Nelen & Schuurmansimport asyncioimport multiprocessingimport osimport pytestimport uvicorndef pytest_sessionstart(session):    """    Called after the Session object has been created and    before performing collection and entering the run test loop.    """    if os.environ.get("DEBUG") or os.environ.get("DEBUG_WAIT_FOR_CLIENT"):        from clean_python.testing.debugger import setup_debugger        setup_debugger()@pytest.fixture(scope="session")def event_loop(request):    """Create an instance of the default event loop per test session.    Async fixtures need the event loop, and so must have the same or narrower scope than    the event_loop fixture. Since we have async session-scoped fixtures, the default    event_loop fixture, which has function scope, cannot be used. See:    https://github.com/pytest-dev/pytest-asyncio#async-fixtures    """    loop = asyncio.get_event_loop_policy().new_event_loop()    yield loop    loop.close()@pytest.fixture(scope="session")async def postgres_url():    return os.environ.get("POSTGRES_URL", "postgres:postgres@localhost:5432")@pytest.fixture(scope="session")async def s3_url():    return os.environ.get("S3_URL", "http://localhost:9000")@pytest.fixture(scope="session")async def fastapi_example_app():    port = int(os.environ.get("API_PORT", "8005"))    config = uvicorn.Config("fastapi_example:app", host="0.0.0.0", port=port)    p = multiprocessing.Process(target=uvicorn.Server(config).run)    p.start()    yield f"http://localhost:{port}"    p.terminate()
 |