Test
Pytest plugin
Turbulette has a built-in pytest plugin to help you write unit tests for your API.
To use it, you'll need both pytest and pytest-asyncio
The Turbulette pytest plugin provides several fixtures to create the test database and setup the ASGI app.
Turbulette's pytest plugin.
blank_conf()
Reload the conf package to clear db, registry and settings.
Scope: function
Source code in turbulette/test/pytest_plugin.py
@pytest.fixture
def blank_conf():
"""Reload the conf package to clear `db`, `registry` and `settings`.
Scope: `function`
"""
app, registry, db, settings = conf.app, conf.registry, conf.db, conf.settings
reload(conf)
yield
conf.app, conf.db, conf.registry, conf.settings = app, db, registry, settings
create_db(db_name, project_settings, request)
Create a test database.
Scope: session
Depends on : db_name
Source code in turbulette/test/pytest_plugin.py
@pytest.fixture(scope="session")
async def create_db(db_name, project_settings, request):
"""Create a test database.
Scope: `session`
Depends on : [db_name][turbulette.test.pytest_plugin.db_name]
"""
# Connect to the default template1 database to create a new one
project_settings.DB_DSN.database = "template1"
# The pool must be able to authorize two connection to drop the test db if needed
engine = await create_engine(str(project_settings.DB_DSN), min_size=1, max_size=2)
async with engine.acquire():
await engine.status(f'CREATE DATABASE "{db_name}"')
project_settings.DB_DSN.database = db_name
yield
# Drop the test db if needed
if not request.config.getoption("--keep-db", default=False):
await engine.status(f'DROP DATABASE "{db_name}"')
await engine.close()
db_name()
Return the name of the test database.
Scope: session
The name is generated as follows : test_YYY-MM-DDThh:mm:ss
Source code in turbulette/test/pytest_plugin.py
@pytest.fixture(scope="session")
def db_name():
"""Return the name of the test database.
Scope: `session`
The name is generated as follows : `test_YYY-MM-DDThh:mm:ss`
"""
return f"test_{datetime.now().replace(microsecond=0).isoformat()}"
tester(turbulette_setup)
Session fixture that return a Tester instance.
Scope: session
Depends on : turbulette_setup
Source code in turbulette/test/pytest_plugin.py
@pytest.fixture(scope="session")
def tester(turbulette_setup):
"""Session fixture that return a [Tester][turbulette.test.tester.Tester] instance.
Scope: `session`
Depends on : [turbulette_setup][turbulette.test.pytest_plugin.turbulette_setup]
"""
return Tester(turbulette_setup.registry.schema)
turbulette_setup(project_settings, create_db)
Create a test database, apply alembic revisions and setup turbulette project.
Scope: session
Depends on : create_db
Source code in turbulette/test/pytest_plugin.py
@pytest.fixture(scope="session")
async def turbulette_setup(project_settings, create_db):
"""Create a test database, apply alembic revisions and setup turbulette project.
Scope: `session`
Depends on : [create_db][turbulette.test.pytest_plugin.create_db]
"""
conf_module = reload(import_module("turbulette.conf"))
setup(project_settings.__name__, database=True)
async with conf.db.with_bind(bind=project_settings.DB_DSN) as engine:
settings_file = Path(find_spec(project_settings.__name__).origin)
alembic_config = (settings_file.parent / "alembic.ini").as_posix()
script_location = (settings_file.parent / "alembic").as_posix()
config = Config(file_=alembic_config)
config.set_main_option("sqlalchemy.url", str(project_settings.DB_DSN))
config.set_main_option("script_location", script_location)
upgrade(config, "heads")
cache = getattr(import_module("turbulette.cache"), "cache")
await cache.connect()
yield conf_module
if cache.is_connected:
await cache.disconnect()
await engine.close()
Tester class
Test helper to make resolver testing more convenient.
Tester
Helper class to test GraphQL queries against the GraphQL schema.
assert_data_in_response(response, data)
classmethod
Assert that the response contains the specified data dict.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
response |
dict |
Response data to check |
required |
data |
Dict[str, Any] |
Data that should be present in |
required |
Source code in turbulette/test/tester.py
@classmethod
def assert_data_in_response(cls, response: dict, data: Dict[str, Any]):
"""Assert that the response contains the specified data dict.
Args:
response (dict): Response data to check
data (Dict[str, Any]): Data that should be present in `response`
"""
for key, value in data.items():
if key in response:
assert response[key] == value
assert_query_failed(self, query, op_name, variables=None, headers=None, jwt=None, errors=True, error_codes=None, raises=None)
async
Assert True if the GraphQL query fails.
With default params, the query fails if:
- Request status is
200(we test the query execution, not the HTTP request) - GraphQL
errorsarray is present - The name of the GraphQL operation is not in the operation response data
Additional options can be used to assert specific results :
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query |
str |
The GraphQL query to execute |
required |
op_name |
str |
The name of the GraphQL operation |
required |
variables |
dict |
Operation variables. |
None |
headers |
dict |
Request headers. |
None |
jwt |
str |
JWT (must be formatted as |
None |
errors |
bool |
Assert that the response data contains the GraphQL |
True |
error_codes |
List[turbulette.errors.ErrorCode] |
List of error codes that must be present in the response. |
None |
raises |
ErrorCode |
Error code that must be present
in |
None |
Returns:
| Type | Description |
|---|---|
Tuple[bool, dict] |
Ariadne's GraphQLResult object response |
Source code in turbulette/test/tester.py
async def assert_query_failed(
self,
query: str,
op_name: str,
variables: dict = None,
headers: dict = None,
jwt: str = None,
errors: bool = True,
error_codes: List[ErrorCode] = None,
raises: ErrorCode = None,
) -> GraphQLResult:
"""Assert True if the GraphQL query fails.
With default params, the query fails if:
- Request status is `200` (we test the query execution, not the HTTP request)
- GraphQL `errors` array **is** present
- The name of the GraphQL operation is **not** in the operation response data
Additional options can be used to assert specific results :
Args:
query: The GraphQL query to execute
op_name: The name of the GraphQL operation
variables: Operation variables.
headers: Request headers.
jwt: JWT (must be formatted as `prefix token`).
errors: Assert that the response data contains the GraphQL `errors` array
error_codes: List of error codes that must be present in the response.
raises ErrorCode: Error code that must be present
in `["extension"]["code"]`.
Returns:
Ariadne's GraphQLResult object response
"""
response = await self.query(query, variables, op_name, headers, jwt)
self.assert_status_200(response)
if errors:
self.assert_errors(response)
if response[1]["data"]:
assert not response[1]["data"][op_name]
if raises:
assert (
raises.name
in response[1]["errors"][0]["extensions"][
conf.settings.TURBULETTE_ERROR_KEY
]
)
if error_codes:
for val in list(response[1]["extensions"].values()):
if isinstance(val, dict):
assert any(code.name in val for code in error_codes)
return response
assert_query_success(self, query, op_name, variables=None, headers=None, jwt=None, op_errors=False, error_codes=None)
async
Assert True if the GraphQL query succeeds.
With default params, the query succeeds if:
- Request status is
200 - GraphQL
errorsarray is not present - The name of the GraphQL operation is in the response at the top level
Additional options can be used to assert specific results :
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query |
str |
The GraphQL query to execute |
required |
op_name |
str |
The name of the GraphQL operation |
required |
variables |
dict |
Operation variables. |
None |
headers |
dict |
Request headers. |
None |
jwt |
str |
JWT (must be formatted as |
None |
op_errors |
|
If |
False |
error_codes |
List[turbulette.errors.ErrorCode] |
List of error codes that must be present in the response. |
None |
Returns:
| Type | Description |
|---|---|
Tuple[bool, dict] |
Ariadne's GraphQLResult object response |
Source code in turbulette/test/tester.py
async def assert_query_success(
self,
query: str,
op_name: str,
variables: dict = None,
headers: dict = None,
jwt: str = None,
op_errors=False,
error_codes: List[ErrorCode] = None,
) -> GraphQLResult:
"""Assert True if the GraphQL query succeeds.
With default params, the query succeeds if:
- Request status is `200`
- GraphQL `errors` array is **not** present
- The name of the GraphQL operation is in the response at the top level
Additional options can be used to assert specific results :
Args:
query: The GraphQL query to execute
op_name: The name of the GraphQL operation
variables: Operation variables.
headers: Request headers.
jwt: JWT (must be formatted as `prefix token`).
op_errors: If `True`, will assert the presence of the error
field (defined by the `ERROR_FIELD` setting) in operation response data.
error_codes: List of error codes that must be present in the response.
Returns:
Ariadne's GraphQLResult object response
"""
response = await self.query(query, variables, op_name, headers, jwt)
self.assert_status_200(response)
self.assert_no_errors(response)
assert response[1]["data"][op_name]
if error_codes:
for val in list(response[1]["extensions"].values()):
if isinstance(val, dict):
assert any(code.name in val for code in error_codes)
if op_errors:
assert response[1]["data"][op_name][self.error_field]
else:
# If no errors, will assert to None
assert not response[1]["data"][op_name].get(self.error_field)
return response