1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253 |
- # (c) Nelen & Schuurmans
- import asyncio
- import multiprocessing
- import os
- import pytest
- import uvicorn
- def 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()
|