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
|
"""Tests for admin user-management: model functions and HTTP endpoints."""
import pytest
from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.authorisation.users.admin.models import (
create_verified_user,
grant_sysadmin_role)
from tests.unit.auth import conftest
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _credential_count(conn, user_id: str) -> int:
with db.cursor(conn) as cursor:
cursor.execute(
"SELECT COUNT(*) AS cnt FROM user_credentials WHERE user_id=?",
(user_id,))
return cursor.fetchone()["cnt"]
@pytest.mark.unit_test
def test_create_verified_user_sets_verified_flag(conn_after_auth_migrations):
"""
GIVEN: a database with migrations applied
WHEN: create_verified_user is called with valid email, name and password
THEN: the returned user has verified=True and the flag is persisted in the DB
"""
conn = conn_after_auth_migrations
user = create_verified_user(conn, "new@example.org", "New User", "s3cr3t")
assert user.verified is True
with db.cursor(conn) as cursor:
cursor.execute(
"SELECT verified FROM users WHERE user_id=?", (str(user.user_id),))
row = cursor.fetchone()
assert row is not None
assert bool(row["verified"]) is True
@pytest.mark.unit_test
def test_create_verified_user_has_no_roles(conn_after_auth_migrations):
"""
GIVEN: a database with migrations applied
WHEN: create_verified_user is called with valid email, name and password
THEN: the new user is assigned no roles
"""
conn = conn_after_auth_migrations
user = create_verified_user(conn, "noroles@example.org", "No Roles User", "s3cr3t")
with db.cursor(conn) as cursor:
cursor.execute(
"SELECT COUNT(*) AS cnt FROM user_roles WHERE user_id=?",
(str(user.user_id),))
row = cursor.fetchone()
assert row["cnt"] == 0
@pytest.mark.unit_test
def test_create_verified_user_stores_credentials(conn_after_auth_migrations):
"""
GIVEN: a database with migrations applied
WHEN: create_verified_user is called with valid email, name and password
THEN: a password credential row is stored for the new user
"""
conn = conn_after_auth_migrations
user = create_verified_user(conn, "creds@example.org", "Creds User", "s3cr3t")
assert _credential_count(conn, str(user.user_id)) == 1
@pytest.mark.unit_test
def test_create_verified_user_raises_on_duplicate_email(conn_after_auth_migrations):
"""
GIVEN: a user already exists with a given email
WHEN: create_verified_user is called with the same email
THEN: an exception is raised
"""
conn = conn_after_auth_migrations
create_verified_user(conn, "dupe@example.org", "First User", "s3cr3t")
with pytest.raises(Exception):
create_verified_user(conn, "dupe@example.org", "Second User", "s3cr3t")
# ---------------------------------------------------------------------------
# HTTP endpoint tests: POST /auth/user/create
# ---------------------------------------------------------------------------
_NEW_USER_BODY = {"email": "newbie@example.org", "name": "Newbie", "password": "s3cr3t"}
@pytest.mark.unit_test
def test_create_user_endpoint_no_token_returns_401(fxtr_app):
"""
GIVEN: no Authorization header
WHEN: POST /auth/user/create
THEN: 401 is returned (OAuth2 layer rejects unauthenticated request)
"""
with fxtr_app.test_client() as http:
res = http.post("/auth/user/create", json=_NEW_USER_BODY)
assert res.status_code == 401
@pytest.mark.unit_test
def test_create_user_endpoint_non_admin_returns_403(fxtr_app, mocker, fxtr_oauth2_clients):
"""
GIVEN: a valid token belonging to a non-admin user
WHEN: POST /auth/user/create
THEN: 403 is returned
"""
_conn, clients = fxtr_oauth2_clients
user = conftest.TEST_USERS[3] # unaff@iliated.user — no privileges
mocker.patch(
"gn_auth.auth.authorisation.users.views.require_oauth.acquire",
conftest.get_tokeniser(
user,
tuple(c for c in clients if c.user == user)[0]))
with fxtr_app.test_client() as http:
res = http.post(
"/auth/user/create",
json=_NEW_USER_BODY,
headers={"Authorization": "Bearer some-mocked-token"})
assert res.status_code == 403
def _setup_admin_mock(conn, clients, mocker):
"""Grant sysadmin role and mock the token for sys@admin.user."""
admin = conftest.TEST_USERS[4]
with db.cursor(conn) as cursor:
grant_sysadmin_role(cursor, admin)
mocker.patch(
"gn_auth.auth.authorisation.users.views.require_oauth.acquire",
conftest.get_tokeniser(
admin,
tuple(c for c in clients if c.user == admin)[0]))
return admin
@pytest.mark.unit_test
def test_create_user_endpoint_admin_returns_201(fxtr_app, mocker, fxtr_oauth2_clients):
"""
GIVEN: a valid system-admin token and a valid request body
WHEN: POST /auth/user/create
THEN: 201 is returned
"""
conn, clients = fxtr_oauth2_clients
_setup_admin_mock(conn, clients, mocker)
with fxtr_app.test_client() as http:
res = http.post(
"/auth/user/create",
json={**_NEW_USER_BODY, "password": "s3cr3tP4ssw0rd"},
headers={"Authorization": "Bearer some-mocked-token"})
assert res.status_code == 201
@pytest.mark.unit_test
def test_create_user_endpoint_admin_returns_new_user(fxtr_app, mocker, fxtr_oauth2_clients):
"""
GIVEN: a valid system-admin token and a valid request body
WHEN: POST /auth/user/create
THEN: the response body contains the new user's email and name
"""
conn, clients = fxtr_oauth2_clients
_setup_admin_mock(conn, clients, mocker)
with fxtr_app.test_client() as http:
res = http.post(
"/auth/user/create",
json={**_NEW_USER_BODY, "password": "s3cr3tP4ssw0rd"},
headers={"Authorization": "Bearer some-mocked-token"})
data = res.get_json()
assert data.get("email") == _NEW_USER_BODY["email"]
assert data.get("name") == _NEW_USER_BODY["name"]
assert res.status_code == 201
@pytest.mark.unit_test
def test_create_user_endpoint_short_password_returns_400(
fxtr_app, mocker, fxtr_oauth2_clients):
"""
GIVEN: a valid system-admin token and a request body with a short password
WHEN: POST /auth/user/create
THEN: 400 is returned (password must be at least 8 characters)
"""
conn, clients = fxtr_oauth2_clients
_setup_admin_mock(conn, clients, mocker)
with fxtr_app.test_client() as http:
res = http.post(
"/auth/user/create",
json={**_NEW_USER_BODY, "password": "short"},
headers={"Authorization": "Bearer some-mocked-token"})
assert res.status_code == 400
|
