aboutsummaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unit/auth/fixtures/group_fixtures.py21
-rw-r--r--tests/unit/auth/fixtures/resource_fixtures.py48
-rw-r--r--tests/unit/auth/fixtures/role_fixtures.py2
-rw-r--r--tests/unit/auth/fixtures/user_fixtures.py21
-rw-r--r--tests/unit/auth/test_admin_user_roles.py262
-rw-r--r--tests/unit/auth/test_admin_users.py190
-rw-r--r--tests/unit/auth/test_groups.py2
-rw-r--r--tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py2
-rw-r--r--tests/unit/auth/test_resources.py18
-rw-r--r--tests/unit/auth/test_system_admin_resources.py94
10 files changed, 629 insertions, 31 deletions
diff --git a/tests/unit/auth/fixtures/group_fixtures.py b/tests/unit/auth/fixtures/group_fixtures.py
index 2e8cd9a..da1c4cd 100644
--- a/tests/unit/auth/fixtures/group_fixtures.py
+++ b/tests/unit/auth/fixtures/group_fixtures.py
@@ -1,5 +1,6 @@
"""Fixtures and utilities for group-related tests"""
import uuid
+import datetime
import pytest
@@ -7,8 +8,12 @@ from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.authorisation.resources.groups import Group
from gn_auth.auth.authorisation.resources import Resource, ResourceCategory
+from .user_fixtures import TEST_USERS
from .resource_fixtures import TEST_RESOURCES
+
+_created_ = datetime.datetime.now()
+
TEST_GROUP_01 = Group(uuid.UUID("9988c21d-f02f-4d45-8966-22c968ac2fbf"),
"TheTestGroup", {})
TEST_GROUP_02 = Group(uuid.UUID("e37d59d7-c05e-4d67-b479-81e627d8d634"),
@@ -24,16 +29,20 @@ GROUPS_AS_RESOURCES = tuple({
"resource_id": res_id,
"resource_name": group.group_name,
"category_id": str(GROUP_CATEGORY.resource_category_id),
- "public": "0"
+ "public": "0",
+ "created_by": str(TEST_USERS[0].user_id),
+ "created_at": _created_.timestamp()
} for res_id, group in zip(
("38d1807d-105f-44a7-8327-7e2d973b6d8d",
"89458ef6-e090-4b53-8c2c-59eaf2785f11"),
TEST_GROUPS))
GROUP_RESOURCES = tuple(
- Resource(uuid.UUID(row["resource_id"]),
- row["resource_name"],
+ Resource(uuid.UUID(row["resource_id"]),# type: ignore[arg-type]
+ row["resource_name"],# type: ignore[arg-type]
GROUP_CATEGORY,
- False)
+ False,
+ created_by=TEST_USERS[0],
+ created_at=_created_)
for row in GROUPS_AS_RESOURCES)
@@ -46,7 +55,7 @@ def __gtuple__(cursor):
return tuple(dict(row) for row in cursor.fetchall())
@pytest.fixture(scope="function")
-def fxtr_group(conn_after_auth_migrations):# pylint: disable=[redefined-outer-name]
+def fxtr_group(conn_after_auth_migrations, fxtr_users):# pylint: disable=[redefined-outer-name, unused-argument]
"""Fixture: setup a test group."""
with db.cursor(conn_after_auth_migrations) as cursor:
cursor.executemany(
@@ -57,7 +66,7 @@ def fxtr_group(conn_after_auth_migrations):# pylint: disable=[redefined-outer-na
cursor.executemany(
"INSERT INTO resources "
- "VALUES(:resource_id, :resource_name, :category_id, :public)",
+ "VALUES(:resource_id, :resource_name, :category_id, :public, :created_by, :created_at)",
GROUPS_AS_RESOURCES)
cursor.executemany(
diff --git a/tests/unit/auth/fixtures/resource_fixtures.py b/tests/unit/auth/fixtures/resource_fixtures.py
index e06f64e..b570a49 100644
--- a/tests/unit/auth/fixtures/resource_fixtures.py
+++ b/tests/unit/auth/fixtures/resource_fixtures.py
@@ -1,11 +1,15 @@
"""Fixtures and utilities for resource-related tests"""
import uuid
+import datetime
import pytest
from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.authorisation.resources import Resource, ResourceCategory
+from .user_fixtures import TEST_USERS
+
+_created_ = datetime.datetime.now()
SYSTEM_CATEGORY = ResourceCategory(
uuid.UUID("aa3d787f-af6a-44fa-9b0b-c82d40e54ad2"),
@@ -15,48 +19,74 @@ SYSTEM_RESOURCE = Resource(
uuid.UUID("0248b289-b277-4eaa-8c94-88a434d14b6e"),
"GeneNetwork System",
SYSTEM_CATEGORY,
- True)
+ True,
+ resource_data=tuple(),
+ created_by=TEST_USERS[4],
+ created_at=_created_)
TEST_RESOURCES = (
Resource(uuid.UUID("26ad1668-29f5-439d-b905-84d551f85955"),
"ResourceG01R01",
ResourceCategory(uuid.UUID("48056f84-a2a6-41ac-8319-0e1e212cba2a"),
"genotype", "Genotype Dataset"),
- True),
+ True,
+ resource_data=tuple(),
+ created_by=TEST_USERS[0],
+ created_at=_created_),
Resource(uuid.UUID("2130aec0-fefd-434d-92fd-9ca342348b2d"),
"ResourceG01R02",
ResourceCategory(uuid.UUID("548d684b-d4d1-46fb-a6d3-51a56b7da1b3"),
"phenotype", "Phenotype (Publish) Dataset"),
- False),
+ False,
+ resource_data=tuple(),
+ created_by=TEST_USERS[0],
+ created_at=_created_),
Resource(uuid.UUID("e9a1184a-e8b4-49fb-b713-8d9cbeea5b83"),
"ResourceG01R03",
ResourceCategory(uuid.UUID("fad071a3-2fc8-40b8-992b-cdefe7dcac79"),
"mrna", "mRNA Dataset"),
- False),
+ False,
+ resource_data=tuple(),
+ created_by=TEST_USERS[0],
+ created_at=_created_),
Resource(uuid.UUID("14496a1c-c234-49a2-978c-8859ea274054"),
"ResourceG02R01",
ResourceCategory(uuid.UUID("48056f84-a2a6-41ac-8319-0e1e212cba2a"),
"genotype", "Genotype Dataset"),
- False),
+ False,
+ resource_data=tuple(),
+ created_by=TEST_USERS[0],
+ created_at=_created_),
Resource(uuid.UUID("04ad9e09-94ea-4390-8a02-11f92999806b"),
"ResourceG02R02",
ResourceCategory(uuid.UUID("fad071a3-2fc8-40b8-992b-cdefe7dcac79"),
"mrna", "mRNA Dataset"),
- True))
+ True,
+ resource_data=tuple(),
+ created_by=TEST_USERS[0],
+ created_at=_created_))
TEST_RESOURCES_PUBLIC = (SYSTEM_RESOURCE, TEST_RESOURCES[0], TEST_RESOURCES[4])
@pytest.fixture(scope="function")
-def fxtr_resources(conn_after_auth_migrations):
+def fxtr_resources(conn_after_auth_migrations, fxtr_users):# pylint: disable=[unused-argument]
"""fixture: setup test resources in the database"""
conn = conn_after_auth_migrations
with db.cursor(conn) as cursor:
cursor.executemany(
- "INSERT INTO resources VALUES (?,?,?,?)",
+ "INSERT INTO resources VALUES (?,?,?,?,?,?)",
((str(res.resource_id), res.resource_name,
str(res.resource_category.resource_category_id),
- 1 if res.public else 0) for res in TEST_RESOURCES))
+ 1 if res.public else 0,
+ str(res.created_by.user_id),
+ res.created_at.timestamp()) for res in TEST_RESOURCES))
+ cursor.execute(
+ "UPDATE resources SET created_by=?, created_at=? "
+ "WHERE resource_id=?",
+ (str(SYSTEM_RESOURCE.created_by.user_id),
+ SYSTEM_RESOURCE.created_at.timestamp(),
+ str(SYSTEM_RESOURCE.resource_id)))
yield (conn, TEST_RESOURCES)
diff --git a/tests/unit/auth/fixtures/role_fixtures.py b/tests/unit/auth/fixtures/role_fixtures.py
index 63a3fca..24e8e9f 100644
--- a/tests/unit/auth/fixtures/role_fixtures.py
+++ b/tests/unit/auth/fixtures/role_fixtures.py
@@ -108,7 +108,7 @@ def fxtr_resource_roles(fxtr_resources, fxtr_roles):# pylint: disable=[redefined
@pytest.fixture(scope="function")
-def fxtr_setup_group_leaders(fxtr_users):
+def fxtr_setup_group_leaders(fxtr_users, fxtr_group):# pylint: disable=[unused-argument]
"""Define what roles users have that target resources of type 'Group'."""
conn, users = fxtr_users
with db.cursor(conn) as cursor:
diff --git a/tests/unit/auth/fixtures/user_fixtures.py b/tests/unit/auth/fixtures/user_fixtures.py
index 1cf0e20..0872142 100644
--- a/tests/unit/auth/fixtures/user_fixtures.py
+++ b/tests/unit/auth/fixtures/user_fixtures.py
@@ -1,28 +1,35 @@
"""Fixtures and utilities for user-related tests"""
import uuid
+import datetime
import pytest
from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.authentication.users import User, hash_password
+_created_ = datetime.datetime.now()
+
TEST_USERS = (
User(uuid.UUID("ecb52977-3004-469e-9428-2a1856725c7f"), "group@lead.er",
- "Group Leader"),
+ "Group Leader", created=_created_),
User(uuid.UUID("21351b66-8aad-475b-84ac-53ce528451e3"),
- "group@mem.ber01", "Group Member 01"),
+ "group@mem.ber01", "Group Member 01", created=_created_),
User(uuid.UUID("ae9c6245-0966-41a5-9a5e-20885a96bea7"),
- "group@mem.ber02", "Group Member 02"),
+ "group@mem.ber02", "Group Member 02", created=_created_),
User(uuid.UUID("9a0c7ce5-2f40-4e78-979e-bf3527a59579"),
- "unaff@iliated.user", "Unaffiliated User"))
+ "unaff@iliated.user", "Unaffiliated User", created=_created_),
+ User(uuid.UUID("60faf8a7-832b-471e-b6a0-bd4013f1fa0e"),
+ "sys@admin.user", "System Admin User", created=_created_))
@pytest.fixture(scope="function")
-def fxtr_users(conn_after_auth_migrations, fxtr_group):# pylint: disable=[redefined-outer-name, unused-argument]
+def fxtr_users(conn_after_auth_migrations):# pylint: disable=[redefined-outer-name, unused-argument]
"""Fixture: setup test users."""
- query = "INSERT INTO users(user_id, email, name) VALUES (?, ?, ?)"
+ query = (
+ "INSERT INTO users(user_id, email, name, created) VALUES (?, ?, ?, ?)")
with db.cursor(conn_after_auth_migrations) as cursor:
cursor.executemany(query, (
- (str(user.user_id), user.email, user.name) for user in TEST_USERS))
+ (str(user.user_id), user.email, user.name, user.created.timestamp())
+ for user in TEST_USERS))
yield (conn_after_auth_migrations, TEST_USERS)
diff --git a/tests/unit/auth/test_admin_user_roles.py b/tests/unit/auth/test_admin_user_roles.py
new file mode 100644
index 0000000..7ce0b1f
--- /dev/null
+++ b/tests/unit/auth/test_admin_user_roles.py
@@ -0,0 +1,262 @@
+"""Tests for admin role-assignment HTTP endpoints."""
+import pytest
+
+from gn_auth.auth.db import sqlite3 as db
+from gn_auth.auth.authorisation.roles.models import assign_user_role_by_name
+
+from tests.unit.auth import conftest
+from tests.unit.auth.fixtures.resource_fixtures import SYSTEM_RESOURCE
+
+# Body used in all role-assign tests — assigning system-administrator on the
+# system resource is a real, migrations-seeded combination.
+_ASSIGN_BODY = {
+ "role_name": "system-administrator",
+ "resource_id": str(SYSTEM_RESOURCE.resource_id)
+}
+
+# Target user for assignment: unaff@iliated.user (no roles initially)
+_TARGET_USER = conftest.TEST_USERS[3]
+
+
+def _setup_admin_mock(conn, clients, mocker):
+ """Grant resource-owner role on SYSTEM_RESOURCE and mock the token.
+
+ resource-owner carries resource:user:assign-role, which is what the
+ endpoint checks. In production the caller would masquerade as the
+ resource owner; here we grant the role directly for test setup.
+ """
+ admin = conftest.TEST_USERS[4]
+ with db.cursor(conn) as cursor:
+ assign_user_role_by_name(
+ cursor, admin, SYSTEM_RESOURCE.resource_id, "resource-owner")
+ 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_assign_role_no_token_returns_401(fxtr_app):
+ """
+ GIVEN: no Authorization header
+ WHEN: POST /auth/user/<uid>/roles/assign
+ THEN: 401 is returned
+ """
+ with fxtr_app.test_client() as http:
+ res = http.post(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/assign",
+ json=_ASSIGN_BODY)
+ assert res.status_code == 401
+
+
+@pytest.mark.unit_test
+def test_assign_role_non_admin_returns_403(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid token belonging to a non-admin user
+ WHEN: POST /auth/user/<uid>/roles/assign
+ 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(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/assign",
+ json=_ASSIGN_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 403
+
+
+def _revoke_assigned_role(conn):
+ """Remove the role row written by the success tests.
+
+ Keeps the DB in the state the fixtures expect — no user_roles entry for
+ _TARGET_USER — so teardown and any subsequent queries are not surprised.
+ """
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ "DELETE FROM user_roles "
+ "WHERE user_id=? "
+ "AND role_id=(SELECT role_id FROM roles WHERE role_name=?) "
+ "AND resource_id=?",
+ (str(_TARGET_USER.user_id),
+ _ASSIGN_BODY["role_name"],
+ _ASSIGN_BODY["resource_id"]))
+
+
+@pytest.mark.unit_test
+def test_assign_role_admin_returns_200(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid system-admin token and a valid role/resource body
+ WHEN: POST /auth/user/<uid>/roles/assign
+ THEN: 200 is returned
+ """
+ conn, clients = fxtr_oauth2_clients
+ _setup_admin_mock(conn, clients, mocker)
+ try:
+ with fxtr_app.test_client() as http:
+ res = http.post(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/assign",
+ json=_ASSIGN_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 200
+ finally:
+ _revoke_assigned_role(conn)
+
+
+def _assign_target_role(conn):
+ """Pre-assign system-administrator to _TARGET_USER on SYSTEM_RESOURCE.
+
+ Required setup for revoke tests: the endpoint can only revoke what exists.
+ """
+ with db.cursor(conn) as cursor:
+ assign_user_role_by_name(
+ cursor, _TARGET_USER, SYSTEM_RESOURCE.resource_id,
+ _ASSIGN_BODY["role_name"])
+
+
+def _cleanup_target_role(conn):
+ """Remove _TARGET_USER's system-administrator row if still present.
+
+ No-op when the revoke endpoint already deleted it; guards against
+ test failures that leave the DB dirty.
+ """
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ "DELETE FROM user_roles "
+ "WHERE user_id=? "
+ "AND role_id=(SELECT role_id FROM roles WHERE role_name=?) "
+ "AND resource_id=?",
+ (str(_TARGET_USER.user_id),
+ _ASSIGN_BODY["role_name"],
+ _ASSIGN_BODY["resource_id"]))
+
+
+@pytest.mark.unit_test
+def test_assign_role_persists_to_db(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid system-admin token and a valid role/resource body
+ WHEN: POST /auth/user/<uid>/roles/assign
+ THEN: the user_roles row is present in the DB for that user/role/resource
+ """
+ conn, clients = fxtr_oauth2_clients
+ _setup_admin_mock(conn, clients, mocker)
+ try:
+ with fxtr_app.test_client() as http:
+ http.post(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/assign",
+ json=_ASSIGN_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ "SELECT COUNT(*) AS cnt FROM user_roles "
+ "INNER JOIN roles ON user_roles.role_id=roles.role_id "
+ "WHERE user_roles.user_id=? "
+ "AND roles.role_name=? "
+ "AND user_roles.resource_id=?",
+ (str(_TARGET_USER.user_id),
+ _ASSIGN_BODY["role_name"],
+ _ASSIGN_BODY["resource_id"]))
+ assert cursor.fetchone()["cnt"] == 1
+ finally:
+ _revoke_assigned_role(conn)
+
+
+# ---------------------------------------------------------------------------
+# HTTP endpoint tests: POST /auth/user/<uid>/roles/revoke
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit_test
+def test_revoke_role_no_token_returns_401(fxtr_app):
+ """
+ GIVEN: no Authorization header
+ WHEN: POST /auth/user/<uid>/roles/revoke
+ THEN: 401 is returned
+ """
+ with fxtr_app.test_client() as http:
+ res = http.post(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/revoke",
+ json=_ASSIGN_BODY)
+ assert res.status_code == 401
+
+
+@pytest.mark.unit_test
+def test_revoke_role_non_admin_returns_403(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid token belonging to a non-admin user
+ WHEN: POST /auth/user/<uid>/roles/revoke
+ 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(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/revoke",
+ json=_ASSIGN_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 403
+
+
+@pytest.mark.unit_test
+def test_revoke_role_admin_returns_200(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid token with resource:user:assign-role and the target user
+ holds the role on the resource
+ WHEN: POST /auth/user/<uid>/roles/revoke
+ THEN: 200 is returned
+ """
+ conn, clients = fxtr_oauth2_clients
+ _setup_admin_mock(conn, clients, mocker)
+ _assign_target_role(conn)
+ try:
+ with fxtr_app.test_client() as http:
+ res = http.post(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/revoke",
+ json=_ASSIGN_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 200
+ finally:
+ _cleanup_target_role(conn)
+
+
+@pytest.mark.unit_test
+def test_revoke_role_removes_from_db(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid token with resource:user:assign-role and the target user
+ holds the role on the resource
+ WHEN: POST /auth/user/<uid>/roles/revoke
+ THEN: the user_roles row is absent from the DB
+ """
+ conn, clients = fxtr_oauth2_clients
+ _setup_admin_mock(conn, clients, mocker)
+ _assign_target_role(conn)
+ try:
+ with fxtr_app.test_client() as http:
+ http.post(
+ f"/auth/user/{_TARGET_USER.user_id}/roles/revoke",
+ json=_ASSIGN_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ with db.cursor(conn) as cursor:
+ cursor.execute(
+ "SELECT COUNT(*) AS cnt FROM user_roles "
+ "INNER JOIN roles ON user_roles.role_id=roles.role_id "
+ "WHERE user_roles.user_id=? "
+ "AND roles.role_name=? "
+ "AND user_roles.resource_id=?",
+ (str(_TARGET_USER.user_id),
+ _ASSIGN_BODY["role_name"],
+ _ASSIGN_BODY["resource_id"]))
+ assert cursor.fetchone()["cnt"] == 0
+ finally:
+ _cleanup_target_role(conn)
diff --git a/tests/unit/auth/test_admin_users.py b/tests/unit/auth/test_admin_users.py
new file mode 100644
index 0000000..eaa5bcd
--- /dev/null
+++ b/tests/unit/auth/test_admin_users.py
@@ -0,0 +1,190 @@
+"""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/system/administration/users/create
+# ---------------------------------------------------------------------------
+
+_NEW_USER_BODY = {"email": "newbie@example.org", "name": "Newbie", "password": "s3cr3t"}
+_CREATE_URL = "/auth/system/administration/users/create"
+
+
+@pytest.mark.unit_test
+def test_create_user_endpoint_no_token_returns_401(fxtr_app):
+ """
+ GIVEN: no Authorization header
+ WHEN: POST /auth/system/administration/users/create
+ THEN: 401 is returned
+ """
+ with fxtr_app.test_client() as http:
+ res = http.post(_CREATE_URL, 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/system/administration/users/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.system.admin.users.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(
+ _CREATE_URL,
+ 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.system.admin.users.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/system/administration/users/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(
+ _CREATE_URL,
+ 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/system/administration/users/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(
+ _CREATE_URL,
+ 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/system/administration/users/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(
+ _CREATE_URL,
+ json={**_NEW_USER_BODY, "password": "short"},
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 400
diff --git a/tests/unit/auth/test_groups.py b/tests/unit/auth/test_groups.py
index 346beb9..6f1e8cd 100644
--- a/tests/unit/auth/test_groups.py
+++ b/tests/unit/auth/test_groups.py
@@ -61,6 +61,8 @@ def __cleanup_create_group__(conn, user, group):
(str(user.user_id), str(grp_rsc["resource_id"])))
cursor.execute("DELETE FROM group_resources WHERE group_id=?",
(str(group.group_id),))
+ cursor.execute("DELETE FROM resources WHERE resource_id=?",
+ (grp_rsc["resource_id"],))
cursor.execute("DELETE FROM groups WHERE group_id=?",
(str(group.group_id),))
diff --git a/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py b/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py
index c34a549..a32cacb 100644
--- a/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py
+++ b/tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py
@@ -8,7 +8,7 @@ from gn_auth.migrations import get_migration, apply_migrations, rollback_migrati
from tests.unit.auth.conftest import (
apply_single_migration, rollback_single_migration, migrations_up_to)
-MIGRATION_PATH = "migrations/auth/20221108_04_CKcSL-init-data-in-resource-categories-table.py"
+MIGRATION_PATH = "gn_auth/migrations/auth/20221108_04_CKcSL-init-data-in-resource-categories-table.py"
@pytest.mark.unit_test
def test_apply_init_data(auth_testdb_path, auth_migrations_dir, backend):
diff --git a/tests/unit/auth/test_resources.py b/tests/unit/auth/test_resources.py
index 04da6df..81f967e 100644
--- a/tests/unit/auth/test_resources.py
+++ b/tests/unit/auth/test_resources.py
@@ -114,19 +114,19 @@ def test_public_resources(fxtr_resources):
"user,expected",
tuple(zip(
conftest.TEST_USERS,
- (sorted(
+ ((sorted(
{res.resource_id: res for res in
((conftest.GROUP_RESOURCES[0],) +
conftest.TEST_RESOURCES_GROUP_01 +
conftest.TEST_RESOURCES_PUBLIC)}.values(),
- key=sort_key_resources),
- sorted(
+ key=sort_key_resources), 6),
+ (sorted(
{res.resource_id: res for res in
((conftest.TEST_RESOURCES_GROUP_01[1],) +
conftest.TEST_RESOURCES_PUBLIC)}.values()
,
- key=sort_key_resources),
- PUBLIC_RESOURCES, PUBLIC_RESOURCES))))
+ key=sort_key_resources), 4),
+ (PUBLIC_RESOURCES, 3), (PUBLIC_RESOURCES, 3)))))
def test_user_resources(fxtr_resource_user_roles, user, expected):
"""
GIVEN: some resources in the database
@@ -134,6 +134,10 @@ def test_user_resources(fxtr_resource_user_roles, user, expected):
THEN: list only the resources for which the user can access
"""
conn, *_others = fxtr_resource_user_roles
+ uresources, count = user_resources(conn, user)
+ eresources, ecount = expected
+ assert count == ecount
assert sorted(
- {res.resource_id: res for res in user_resources(conn, user)
- }.values(), key=sort_key_resources) == expected
+ {res.resource_id: res for res in uresources}.values(),
+ key=sort_key_resources
+ ) == eresources
diff --git a/tests/unit/auth/test_system_admin_resources.py b/tests/unit/auth/test_system_admin_resources.py
new file mode 100644
index 0000000..76f3ca3
--- /dev/null
+++ b/tests/unit/auth/test_system_admin_resources.py
@@ -0,0 +1,94 @@
+"""Tests for system admin resource-ownership endpoints.
+
+Covers POST /auth/system/administration/resources/<id>/assign-owner
+ and POST /auth/system/administration/resources/<id>/revoke-owner.
+"""
+import pytest
+
+from tests.unit.auth import conftest
+from tests.unit.auth.fixtures.resource_fixtures import TEST_RESOURCES
+
+# Arbitrary target resource for the endpoint path — the privilege check fires
+# before any resource lookup, so the resource need not exist for 401/403 tests.
+_TARGET_RESOURCE = str(TEST_RESOURCES[0].resource_id)
+
+# Minimal body for both endpoints (the target user for ownership change).
+_BODY = {"user_id": str(conftest.TEST_USERS[3].user_id)}
+
+_ASSIGN_URL = f"/auth/system/administration/resources/{_TARGET_RESOURCE}/assign-owner"
+_REVOKE_URL = f"/auth/system/administration/resources/{_TARGET_RESOURCE}/revoke-owner"
+
+_NON_SYSADMIN = conftest.TEST_USERS[3] # unaff@iliated.user — no roles at all
+
+
+def _mock_token(mocker, user, clients):
+ """Patch require_oauth.acquire in the admin resources module."""
+ mocker.patch(
+ "gn_auth.auth.system.admin.resources.require_oauth.acquire",
+ conftest.get_tokeniser(
+ user,
+ tuple(c for c in clients if c.user == user)[0]))
+
+
+# ---------------------------------------------------------------------------
+# No-token tests (401)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit_test
+def test_assign_owner_no_token_returns_401(fxtr_app):
+ """
+ GIVEN: no Authorization header
+ WHEN: POST .../assign-owner
+ THEN: 401 is returned
+ """
+ with fxtr_app.test_client() as http:
+ res = http.post(_ASSIGN_URL, json=_BODY)
+ assert res.status_code == 401
+
+
+@pytest.mark.unit_test
+def test_revoke_owner_no_token_returns_401(fxtr_app):
+ """
+ GIVEN: no Authorization header
+ WHEN: POST .../revoke-owner
+ THEN: 401 is returned
+ """
+ with fxtr_app.test_client() as http:
+ res = http.post(_REVOKE_URL, json=_BODY)
+ assert res.status_code == 401
+
+
+# ---------------------------------------------------------------------------
+# Non-sysadmin tests (403)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.unit_test
+def test_assign_owner_non_sysadmin_returns_403(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid token for a user without system:resource:assign-owner
+ WHEN: POST .../assign-owner
+ THEN: 403 is returned
+ """
+ _conn, clients = fxtr_oauth2_clients
+ _mock_token(mocker, _NON_SYSADMIN, clients)
+ with fxtr_app.test_client() as http:
+ res = http.post(
+ _ASSIGN_URL, json=_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 403
+
+
+@pytest.mark.unit_test
+def test_revoke_owner_non_sysadmin_returns_403(fxtr_app, mocker, fxtr_oauth2_clients):
+ """
+ GIVEN: a valid token for a user without system:resource:assign-owner
+ WHEN: POST .../revoke-owner
+ THEN: 403 is returned
+ """
+ _conn, clients = fxtr_oauth2_clients
+ _mock_token(mocker, _NON_SYSADMIN, clients)
+ with fxtr_app.test_client() as http:
+ res = http.post(
+ _REVOKE_URL, json=_BODY,
+ headers={"Authorization": "Bearer some-mocked-token"})
+ assert res.status_code == 403
generated by cgit v1.2.3 (git 2.44.0) at 2026-09-24 19:01:30 +0000