mirror of
https://github.com/hasura/graphql-engine.git
synced 2024-12-16 09:51:59 +03:00
3cb9bab9f1
When we run the HGE server inside the test harness, it needs to run with an admin secret for some tests to make sense. This tags each test that requires an admin secret with `pytest.mark.admin_secret`, which then generates a UUID and injects that into both the server and the test case (if required). It also simplifies the way the test harness picks up an existing admin secret, allowing it to use the environment variable instead of requiring it via a parameter. PR-URL: https://github.com/hasura/graphql-engine-mono/pull/6120 GitOrigin-RevId: 55c5b9e8c99bdad9c8304098444ddb9516749a2c
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
import os
|
|
import pytest
|
|
import subprocess
|
|
import threading
|
|
from typing import Optional
|
|
|
|
import ports
|
|
|
|
# These are the names of the environment variables that should be passed through to the HGE binary.
|
|
# Other variables are ignored.
|
|
_PASS_THROUGH_ENV_VARS = set([
|
|
'PATH', # required for basically anything to work
|
|
'HASURA_GRAPHQL_PG_SOURCE_URL_1',
|
|
'HASURA_GRAPHQL_PG_SOURCE_URL_2',
|
|
])
|
|
|
|
|
|
def hge_port() -> int:
|
|
return ports.find_free_port()
|
|
|
|
|
|
def hge_server(
|
|
request: pytest.FixtureRequest,
|
|
hge_bin: str,
|
|
hge_port: int,
|
|
hge_url: str,
|
|
hge_key: Optional[str],
|
|
hge_fixture_env: dict[str, str],
|
|
pg_url: str,
|
|
) -> Optional[str]:
|
|
hge_env: dict[str, str] = {name: value for name, value in os.environ.items() if name in _PASS_THROUGH_ENV_VARS}
|
|
hge_marker_env: dict[str, str] = {marker.args[0]: marker.args[1] for marker in request.node.iter_markers('hge_env') if marker.args[1] is not None}
|
|
env = {
|
|
**hge_env,
|
|
**hge_fixture_env,
|
|
**hge_marker_env,
|
|
}
|
|
|
|
hge_key_args = ['--admin-secret', hge_key] if hge_key else []
|
|
|
|
print(f'Starting GraphQL Engine on {hge_url}...')
|
|
hge_process = subprocess.Popen(
|
|
args = [
|
|
hge_bin,
|
|
'--database-url', pg_url,
|
|
'serve',
|
|
'--server-port', str(hge_port),
|
|
'--stringify-numeric-types',
|
|
*hge_key_args,
|
|
],
|
|
env = env,
|
|
)
|
|
|
|
def stop():
|
|
if hge_process.poll() is None:
|
|
print(f'Stopping GraphQL Engine on {hge_url}...')
|
|
hge_process.terminate()
|
|
try:
|
|
hge_process.wait(timeout = 5)
|
|
print(f'GraphQL Engine on {hge_url} has stopped.')
|
|
except subprocess.TimeoutExpired:
|
|
print(f'Given up waiting; killing GraphQL Engine...')
|
|
hge_process.kill()
|
|
hge_process.wait()
|
|
print(f'GraphQL Engine has been successfully killed.')
|
|
else:
|
|
print(f'GraphQL Engine on {hge_url} has already stopped.')
|
|
|
|
# We can re-enable stopping in the background once tests no longer share a database.
|
|
# Until then, HGE can interfere with other tests.
|
|
request.addfinalizer(stop)
|
|
## Stop in the background so we don't hold up other tests.
|
|
# request.addfinalizer(lambda: threading.Thread(target = stop).start())
|
|
|
|
ports.wait_for_port(hge_port, timeout = 30)
|
|
return hge_url
|