1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
|
"""
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"]
|
