""" Shared fixtures for the GeneNetwork integration test suite. Base URLs default to the CD environment. Override via environment variables: GN2_BASE_URL e.g. https://genenetwork.org (default: CD) GN3_BASE_URL e.g. https://genenetwork.org/api3 (default: CD) GN_AUTH_BASE_URL e.g. https://auth.genenetwork.org (default: CD) For auth-flow tests set either: File-based (CI — populated by create-test-users / create-test-oauth2-client): GN_TEST_USERS_FILE path to the users credentials JSON file GN_TEST_CLIENT_FILE path to the OAuth2 client credentials JSON file Individual env vars (manual runs, admin user only): GN_TEST_EMAIL registered admin test-user e-mail address GN_TEST_PASSWORD password for GN_TEST_EMAIL GN_OAUTH2_CLIENT_ID OAuth2 client UUID GN_OAUTH2_CLIENT_SECRET OAuth2 client secret """ import json import os import time import pytest import requests _GN2_DEFAULT = "https://cd.genenetwork.org" _GN3_DEFAULT = "https://cd.genenetwork.org/api3" _GN_AUTH_DEFAULT = "https://auth-cd.genenetwork.org" def _wait_for(url: str, timeout: int = 120, interval: int = 2) -> None: """Poll url until the upstream service is up or timeout elapses. Retries on ConnectionError (process not yet bound) and on 502/503 (Nginx upstream-not-ready responses). Any other HTTP response is treated as "service is up". """ deadline = time.monotonic() + timeout last_exc = None while time.monotonic() < deadline: try: resp = requests.get(url, timeout=5, allow_redirects=True) if resp.status_code not in (502, 503): return except requests.exceptions.ConnectionError as exc: last_exc = exc time.sleep(interval) raise RuntimeError( f"Service at {url} did not become available within {timeout}s" ) from last_exc def _read_json_file(path: str) -> dict: with open(path, encoding="utf8") as f: return json.load(f) def _user_by_role(users_data: dict, role: str): return next( (u for u in users_data.get("users", []) if u["role"] == role), None, ) @pytest.fixture(scope="session") def gn2_url() -> str: url = os.environ.get("GN2_BASE_URL", _GN2_DEFAULT).rstrip("/") _wait_for(url) return url @pytest.fixture(scope="session") def gn3_url() -> str: url = os.environ.get("GN3_BASE_URL", _GN3_DEFAULT).rstrip("/") _wait_for(url) return url @pytest.fixture(scope="session") def gn_auth_url() -> str: url = os.environ.get("GN_AUTH_BASE_URL", _GN_AUTH_DEFAULT).rstrip("/") _wait_for(url) return url @pytest.fixture(scope="session") def http() -> requests.Session: """Shared requests.Session; sets a conservative timeout for all calls.""" with requests.Session() as session: session.headers.update({"Accept": "application/json"}) yield session # --------------------------------------------------------------------------- # Auth-flow helpers (Phase 2 tests) # --------------------------------------------------------------------------- @pytest.fixture(scope="session") def oauth2_credentials(): """Return (email, password, client_id, client_secret) for the admin test user. Reads from GN_TEST_USERS_FILE + GN_TEST_CLIENT_FILE when set (CI), or falls back to GN_TEST_EMAIL / GN_TEST_PASSWORD / GN_OAUTH2_CLIENT_ID / GN_OAUTH2_CLIENT_SECRET for manual runs. Skips if neither is available. """ users_file = os.environ.get("GN_TEST_USERS_FILE") client_file = os.environ.get("GN_TEST_CLIENT_FILE") if users_file and client_file: users_data = _read_json_file(users_file) client_data = _read_json_file(client_file) admin = _user_by_role(users_data, "system-admin") if admin is None: pytest.fail(f"No system-admin user found in {users_file}") return ( admin["email"], admin["password"], client_data["client"]["client_id"], client_data["client"]["client_secret"], ) email = os.environ.get("GN_TEST_EMAIL") password = os.environ.get("GN_TEST_PASSWORD") client_id = os.environ.get("GN_OAUTH2_CLIENT_ID") client_secret = os.environ.get("GN_OAUTH2_CLIENT_SECRET") if not all([email, password, client_id, client_secret]): pytest.skip( "Set GN_TEST_USERS_FILE + GN_TEST_CLIENT_FILE (CI), or " "GN_TEST_EMAIL, GN_TEST_PASSWORD, GN_OAUTH2_CLIENT_ID, and " "GN_OAUTH2_CLIENT_SECRET (manual) to run auth-flow tests." ) return email, password, client_id, client_secret @pytest.fixture(scope="session") def basic_oauth2_credentials(): """Return (email, password, client_id, client_secret) for the unprivileged test user. Requires GN_TEST_USERS_FILE + GN_TEST_CLIENT_FILE (set by the CI auth-flow job). Skips if not available — basic-user tests only run in CI where the full test-session setup has been performed. """ users_file = os.environ.get("GN_TEST_USERS_FILE") client_file = os.environ.get("GN_TEST_CLIENT_FILE") if not (users_file and client_file): pytest.skip( "Set GN_TEST_USERS_FILE and GN_TEST_CLIENT_FILE to run " "auth-flow tests that require an unprivileged user." ) users_data = _read_json_file(users_file) client_data = _read_json_file(client_file) basic = _user_by_role(users_data, "none") if basic is None: pytest.fail(f"No user with role='none' found in {users_file}") return ( basic["email"], basic["password"], client_data["client"]["client_id"], client_data["client"]["client_secret"], ) @pytest.fixture(scope="session") def access_token(gn_auth_url, oauth2_credentials, http): """Obtains a Bearer token via the password grant and caches it for the session.""" email, password, client_id, client_secret = oauth2_credentials resp = http.post( f"{gn_auth_url}/auth/token", json={ "grant_type": "password", "username": email, "password": password, "scope": "profile group resource", "client_id": client_id, "client_secret": client_secret, }, timeout=30, ) assert resp.status_code == 200, f"Token request failed: {resp.text}" data = resp.json() assert "access_token" in data return data["access_token"] @pytest.fixture(scope="session") def basic_access_token(gn_auth_url, basic_oauth2_credentials, http): """Bearer token for the unprivileged test user.""" email, password, client_id, client_secret = basic_oauth2_credentials resp = http.post( f"{gn_auth_url}/auth/token", json={ "grant_type": "password", "username": email, "password": password, "scope": "profile group resource", "client_id": client_id, "client_secret": client_secret, }, timeout=30, ) assert resp.status_code == 200, f"Basic user token request failed: {resp.text}" data = resp.json() assert "access_token" in data return data["access_token"]