diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/unit/auth/fixtures/group_fixtures.py | 21 | ||||
| -rw-r--r-- | tests/unit/auth/fixtures/resource_fixtures.py | 48 | ||||
| -rw-r--r-- | tests/unit/auth/fixtures/role_fixtures.py | 4 | ||||
| -rw-r--r-- | tests/unit/auth/fixtures/user_fixtures.py | 21 | ||||
| -rw-r--r-- | tests/unit/auth/test_admin_user_roles.py | 262 | ||||
| -rw-r--r-- | tests/unit/auth/test_admin_users.py | 190 | ||||
| -rw-r--r-- | tests/unit/auth/test_groups.py | 8 | ||||
| -rw-r--r-- | tests/unit/auth/test_migrations_add_data_to_table.py | 4 | ||||
| -rw-r--r-- | tests/unit/auth/test_migrations_add_remove_columns.py | 4 | ||||
| -rw-r--r-- | tests/unit/auth/test_migrations_indexes.py | 4 | ||||
| -rw-r--r-- | tests/unit/auth/test_migrations_init_data_in_resource_categories_table.py | 2 | ||||
| -rw-r--r-- | tests/unit/auth/test_migrations_insert_data_into_empty_table.py | 4 | ||||
| -rw-r--r-- | tests/unit/auth/test_privileges.py | 13 | ||||
| -rw-r--r-- | tests/unit/auth/test_resources.py | 39 | ||||
| -rw-r--r-- | tests/unit/auth/test_resources_roles.py | 2 | ||||
| -rw-r--r-- | tests/unit/auth/test_roles.py | 15 | ||||
| -rw-r--r-- | tests/unit/auth/test_system_admin_resources.py | 94 |
17 files changed, 678 insertions, 57 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 1858712..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: @@ -163,7 +163,7 @@ def fxtr_system_roles(fxtr_users): @pytest.fixture(scope="function") -def fxtr_resource_user_roles(# pylint: disable=[too-many-arguments, too-many-locals] +def fxtr_resource_user_roles(# pylint: disable=[too-many-arguments, too-many-locals, too-many-positional-arguments] fxtr_resources, fxtr_users_in_group, fxtr_resource_ownership, 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 16df56e..6f1e8cd 100644 --- a/tests/unit/auth/test_groups.py +++ b/tests/unit/auth/test_groups.py @@ -27,7 +27,7 @@ PRIVILEGES = ( @pytest.mark.unit_test @pytest.mark.parametrize("user", tuple(conftest.TEST_USERS[0:3])) -def test_create_group_fails(# pylint: disable=[too-many-arguments] +def test_create_group_fails(# pylint: disable=[too-many-arguments too-many-positional-arguments] fxtr_app, auth_testdb_path, mocker, fxtr_resource_user_roles, fxtr_oauth2_clients, user):# pylint: disable=[unused-argument] """ GIVEN: an authenticated user @@ -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),)) @@ -71,7 +73,7 @@ def __cleanup_create_group__(conn, user, group): ((conftest.TEST_USERS[3], Group( UUID("d32611e3-07fc-4564-b56c-786c6db6de2b"), "a_test_group", {"group_description": "A test group"})),)) -def test_create_group_succeeds(# pylint: disable=[too-many-arguments, unused-argument] +def test_create_group_succeeds(# pylint: disable=[too-many-arguments too-many-positional-arguments, unused-argument] fxtr_app, auth_testdb_path, mocker, @@ -102,7 +104,7 @@ def test_create_group_succeeds(# pylint: disable=[too-many-arguments, unused-arg @pytest.mark.unit_test @pytest.mark.parametrize("user", conftest.TEST_USERS[1:]) -def test_create_group_raises_exception_with_non_privileged_user(# pylint: disable=[too-many-arguments] +def test_create_group_raises_exception_with_non_privileged_user(# pylint: disable=[too-many-arguments too-many-positional-arguments] fxtr_app, auth_testdb_path, mocker, fxtr_users, fxtr_oauth2_clients, user):# pylint: disable=[unused-argument] """ GIVEN: an authenticated user, without appropriate privileges diff --git a/tests/unit/auth/test_migrations_add_data_to_table.py b/tests/unit/auth/test_migrations_add_data_to_table.py index d9e2ca4..0945a20 100644 --- a/tests/unit/auth/test_migrations_add_data_to_table.py +++ b/tests/unit/auth/test_migrations_add_data_to_table.py @@ -40,7 +40,7 @@ test_params = ( @pytest.mark.unit_test @pytest.mark.parametrize("migration_file,query,query_params,data", test_params) -def test_apply_insert(# pylint: disable=[too-many-arguments] +def test_apply_insert(# pylint: disable=[too-many-arguments, too-many-positional-arguments] auth_migrations_dir, backend, auth_testdb_path, migration_file, query, query_params, data): """ @@ -65,7 +65,7 @@ def test_apply_insert(# pylint: disable=[too-many-arguments] @pytest.mark.unit_test @pytest.mark.parametrize("migration_file,query,query_params,data", test_params) -def test_rollback_insert(# pylint: disable=[too-many-arguments] +def test_rollback_insert(# pylint: disable=[too-many-arguments, too-many-positional-arguments] auth_migrations_dir, backend, auth_testdb_path, migration_file, query, query_params, data): """ diff --git a/tests/unit/auth/test_migrations_add_remove_columns.py b/tests/unit/auth/test_migrations_add_remove_columns.py index af85652..15dc3a2 100644 --- a/tests/unit/auth/test_migrations_add_remove_columns.py +++ b/tests/unit/auth/test_migrations_add_remove_columns.py @@ -51,7 +51,7 @@ def rolled_back_successfully(adding: bool, result_str: str, column: str) -> bool @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,the_table,the_column,adding", TEST_PARAMS) -def test_apply_add_remove_column(# pylint: disable=[too-many-arguments] +def test_apply_add_remove_column(# pylint: disable=[too-many-arguments too-many-positional-arguments] auth_migrations_dir, auth_testdb_path, backend, migration_file, the_table, the_column, adding): """ @@ -84,7 +84,7 @@ def test_apply_add_remove_column(# pylint: disable=[too-many-arguments] @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,the_table,the_column,adding", TEST_PARAMS) -def test_rollback_add_remove_column(# pylint: disable=[too-many-arguments] +def test_rollback_add_remove_column(# pylint: disable=[too-many-arguments too-many-positional-arguments] auth_migrations_dir, auth_testdb_path, backend, migration_file, the_table, the_column, adding): """ diff --git a/tests/unit/auth/test_migrations_indexes.py b/tests/unit/auth/test_migrations_indexes.py index 1c543c4..2d0997f 100644 --- a/tests/unit/auth/test_migrations_indexes.py +++ b/tests/unit/auth/test_migrations_indexes.py @@ -30,7 +30,7 @@ migrations_tables_and_indexes = ( @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,the_table,the_index", migrations_tables_and_indexes) -def test_index_created(# pylint: disable=[too-many-arguments] +def test_index_created(# pylint: disable=[too-many-arguments too-many-positional-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, the_table, the_index): """ @@ -61,7 +61,7 @@ def test_index_created(# pylint: disable=[too-many-arguments] @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,the_table,the_index", migrations_tables_and_indexes) -def test_index_dropped(# pylint: disable=[too-many-arguments] +def test_index_dropped(# pylint: disable=[too-many-arguments too-many-positional-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, the_table, the_index): """ 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_migrations_insert_data_into_empty_table.py b/tests/unit/auth/test_migrations_insert_data_into_empty_table.py index 0cf9a1f..c699e81 100644 --- a/tests/unit/auth/test_migrations_insert_data_into_empty_table.py +++ b/tests/unit/auth/test_migrations_insert_data_into_empty_table.py @@ -16,7 +16,7 @@ test_params = ( @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,table,row_count", test_params) -def test_apply_insert(# pylint: disable=[too-many-arguments] +def test_apply_insert(# pylint: disable=[too-many-arguments, too-many-positional-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, table, row_count): """ @@ -45,7 +45,7 @@ def test_apply_insert(# pylint: disable=[too-many-arguments] @pytest.mark.unit_test @pytest.mark.parametrize( "migration_file,table,row_count", test_params) -def test_rollback_insert(# pylint: disable=[too-many-arguments] +def test_rollback_insert(# pylint: disable=[too-many-arguments, too-many-positional-arguments] auth_testdb_path, auth_migrations_dir, backend, migration_file, table, row_count): """ diff --git a/tests/unit/auth/test_privileges.py b/tests/unit/auth/test_privileges.py index 619ccc1..41dae7f 100644 --- a/tests/unit/auth/test_privileges.py +++ b/tests/unit/auth/test_privileges.py @@ -24,7 +24,18 @@ PRIVILEGES = sorted( Privilege("group:resource:view-resource", "view a resource and use it in computations"), Privilege("group:resource:edit-resource", "edit/update a resource"), - Privilege("group:resource:delete-resource", "Delete a resource")), + Privilege("group:resource:delete-resource", "Delete a resource"), + + Privilege("group:data:link-to-group", + "Allow linking data to only one specific group."), + + # Role-management privileges + Privilege("resource:role:create-role", + "Create a new role on a specific resource"), + Privilege("resource:role:delete-role", + "Delete an existing role from a specific resource"), + Privilege("resource:role:edit-role", + "Edit an existing role on a specific resource")), key=sort_key_privileges) @pytest.mark.unit_test diff --git a/tests/unit/auth/test_resources.py b/tests/unit/auth/test_resources.py index 7f0b43d..81f967e 100644 --- a/tests/unit/auth/test_resources.py +++ b/tests/unit/auth/test_resources.py @@ -30,7 +30,7 @@ create_resource_failure = { (Resource( uuid.UUID("d32611e3-07fc-4564-b56c-786c6db6de2b"), "test_resource", resource_category, False),)))) -def test_create_resource(# pylint: disable=[too-many-arguments, unused-argument] +def test_create_resource(# pylint: disable=[too-many-arguments, too-many-positional-arguments, unused-argument] mocker, fxtr_users_in_group, fxtr_resource_user_roles, @@ -50,7 +50,7 @@ def test_create_resource(# pylint: disable=[too-many-arguments, unused-argument] with db.cursor(conn) as cursor: resource = create_resource( - cursor, "test_resource", resource_category, user, _group, False) + conn, "test_resource", resource_category, user, _group, False) assert resource == expected # Cleanup cursor.execute( @@ -82,15 +82,14 @@ def test_create_resource_raises_for_unauthorised_users( tuple(client for client in clients if client.user == user)[0])) conn, _group, _users = fxtr_users_in_group with pytest.raises(AuthorisationError): - with db.cursor(conn) as cursor: - assert create_resource( - cursor, - "test_resource", - resource_category, - user, - _group, - False - ) == expected + assert create_resource( + conn, + "test_resource", + resource_category, + user, + _group, + False + ) == expected def sort_key_resources(resource): """Sort-key for resources.""" @@ -115,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 @@ -135,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_resources_roles.py b/tests/unit/auth/test_resources_roles.py index 39a198f..e43f25c 100644 --- a/tests/unit/auth/test_resources_roles.py +++ b/tests/unit/auth/test_resources_roles.py @@ -63,7 +63,7 @@ def test_create_group_role(mocker, fxtr_users_in_group, fxtr_oauth2_clients, use "user,expected", tuple(zip(conftest.TEST_USERS[0:1], ( Role(UUID("d32611e3-07fc-4564-b56c-786c6db6de2b"), "a_test_role", True, PRIVILEGES),)))) -def test_create_role(# pylint: disable=[too-many-arguments, unused-argument] +def test_create_role(# pylint: disable=[too-many-arguments, too-many-positional-arguments, unused-argument] fxtr_app, auth_testdb_path, mocker, diff --git a/tests/unit/auth/test_roles.py b/tests/unit/auth/test_roles.py index 251defb..b7512ef 100644 --- a/tests/unit/auth/test_roles.py +++ b/tests/unit/auth/test_roles.py @@ -26,7 +26,7 @@ PRIVILEGES = ( @pytest.mark.parametrize( "user,expected", tuple(zip(conftest.TEST_USERS[1:], ( create_role_failure, create_role_failure, create_role_failure)))) -def test_create_role_raises_exception_for_unauthorised_users(# pylint: disable=[too-many-arguments, unused-argument] +def test_create_role_raises_exception_for_unauthorised_users(# pylint: disable=[too-many-arguments, unused-argument, too-many-positional-arguments] fxtr_app, auth_testdb_path, mocker, @@ -115,6 +115,10 @@ def test_create_role_raises_exception_for_unauthorised_users(# pylint: disable=[ user_editable=False, privileges=( Privilege( + "group:data:link-to-group", + "Allow linking data to only one specific group."), + + Privilege( privilege_id="group:resource:create-resource", privilege_description="Create a resource object"), Privilege( @@ -133,6 +137,15 @@ def test_create_role_raises_exception_for_unauthorised_users(# pylint: disable=[ privilege_id="group:user:remove-group-member", privilege_description="Remove a user from a group"), Privilege( + privilege_id="resource:role:create-role", + privilege_description="Create a new role on a specific resource"), + Privilege( + privilege_id="resource:role:delete-role", + privilege_description="Delete an existing role from a specific resource"), + Privilege( + privilege_id="resource:role:edit-role", + privilege_description="Edit an existing role on a specific resource"), + Privilege( privilege_id="system:group:delete-group", privilege_description="Delete a group"), Privilege( 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 |