From 8b7c598407a5fea9a3d78473e72df87606998cd4 Mon Sep 17 00:00:00 2001 From: Frederick Muriuki Muriithi Date: Fri, 4 Aug 2023 10:10:28 +0300 Subject: Copy over files from GN3 repository. --- tests/unit/auth/fixtures/__init__.py | 8 ++ tests/unit/auth/fixtures/group_fixtures.py | 147 +++++++++++++++++++++ tests/unit/auth/fixtures/migration_fixtures.py | 51 +++++++ tests/unit/auth/fixtures/oauth2_client_fixtures.py | 51 +++++++ tests/unit/auth/fixtures/resource_fixtures.py | 25 ++++ tests/unit/auth/fixtures/role_fixtures.py | 45 +++++++ tests/unit/auth/fixtures/user_fixtures.py | 66 +++++++++ 7 files changed, 393 insertions(+) create mode 100644 tests/unit/auth/fixtures/__init__.py create mode 100644 tests/unit/auth/fixtures/group_fixtures.py create mode 100644 tests/unit/auth/fixtures/migration_fixtures.py create mode 100644 tests/unit/auth/fixtures/oauth2_client_fixtures.py create mode 100644 tests/unit/auth/fixtures/resource_fixtures.py create mode 100644 tests/unit/auth/fixtures/role_fixtures.py create mode 100644 tests/unit/auth/fixtures/user_fixtures.py (limited to 'tests/unit/auth/fixtures') diff --git a/tests/unit/auth/fixtures/__init__.py b/tests/unit/auth/fixtures/__init__.py new file mode 100644 index 0000000..a675fc7 --- /dev/null +++ b/tests/unit/auth/fixtures/__init__.py @@ -0,0 +1,8 @@ +"""pytest's conftest as a module.""" +from .role_fixtures import * +from .user_fixtures import * +from .group_fixtures import * +from .resource_fixtures import * +# from .privilege_fixtures import * +from .migration_fixtures import * +from .oauth2_client_fixtures import * diff --git a/tests/unit/auth/fixtures/group_fixtures.py b/tests/unit/auth/fixtures/group_fixtures.py new file mode 100644 index 0000000..d7bbc56 --- /dev/null +++ b/tests/unit/auth/fixtures/group_fixtures.py @@ -0,0 +1,147 @@ +"""Fixtures and utilities for group-related tests""" +import uuid + +import pytest + +from gn3.auth import db +from gn3.auth.authorisation.groups import Group, GroupRole +from gn3.auth.authorisation.resources import Resource, ResourceCategory + +from .role_fixtures import RESOURCE_EDITOR_ROLE, RESOURCE_READER_ROLE + +TEST_GROUP_01 = Group(uuid.UUID("9988c21d-f02f-4d45-8966-22c968ac2fbf"), + "TheTestGroup", {}) +TEST_GROUP_02 = Group(uuid.UUID("e37d59d7-c05e-4d67-b479-81e627d8d634"), + "AnotherTestGroup", {}) +TEST_GROUPS = (TEST_GROUP_01, TEST_GROUP_02) + +TEST_RESOURCES_GROUP_01 = ( + Resource(TEST_GROUPS[0], uuid.UUID("26ad1668-29f5-439d-b905-84d551f85955"), + "ResourceG01R01", + ResourceCategory(uuid.UUID("48056f84-a2a6-41ac-8319-0e1e212cba2a"), + "genotype", "Genotype Dataset"), + True), + Resource(TEST_GROUPS[0], uuid.UUID("2130aec0-fefd-434d-92fd-9ca342348b2d"), + "ResourceG01R02", + ResourceCategory(uuid.UUID("548d684b-d4d1-46fb-a6d3-51a56b7da1b3"), + "phenotype", "Phenotype (Publish) Dataset"), + False), + Resource(TEST_GROUPS[0], uuid.UUID("e9a1184a-e8b4-49fb-b713-8d9cbeea5b83"), + "ResourceG01R03", + ResourceCategory(uuid.UUID("fad071a3-2fc8-40b8-992b-cdefe7dcac79"), + "mrna", "mRNA Dataset"), + False)) + +TEST_RESOURCES_GROUP_02 = ( + Resource(TEST_GROUPS[1], uuid.UUID("14496a1c-c234-49a2-978c-8859ea274054"), + "ResourceG02R01", + ResourceCategory(uuid.UUID("48056f84-a2a6-41ac-8319-0e1e212cba2a"), + "genotype", "Genotype Dataset"), + False), + Resource(TEST_GROUPS[1], uuid.UUID("04ad9e09-94ea-4390-8a02-11f92999806b"), + "ResourceG02R02", + ResourceCategory(uuid.UUID("fad071a3-2fc8-40b8-992b-cdefe7dcac79"), + "mrna", "mRNA Dataset"), + True)) + +TEST_RESOURCES = TEST_RESOURCES_GROUP_01 + TEST_RESOURCES_GROUP_02 +TEST_RESOURCES_PUBLIC = (TEST_RESOURCES_GROUP_01[0], TEST_RESOURCES_GROUP_02[1]) + +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] + """Fixture: setup a test group.""" + query = "INSERT INTO groups(group_id, group_name) VALUES (?, ?)" + with db.cursor(conn_after_auth_migrations) as cursor: + cursor.executemany( + query, tuple( + (str(group.group_id), group.group_name) + for group in TEST_GROUPS)) + + yield (conn_after_auth_migrations, TEST_GROUPS[0]) + + with db.cursor(conn_after_auth_migrations) as cursor: + cursor.executemany( + "DELETE FROM groups WHERE group_id=?", + ((str(group.group_id),) for group in TEST_GROUPS)) + +@pytest.fixture(scope="function") +def fxtr_users_in_group(fxtr_group, fxtr_users):# pylint: disable=[redefined-outer-name, unused-argument] + """Link the users to the groups.""" + conn, all_users = fxtr_users + users = tuple( + user for user in all_users if user.email not in ("unaff@iliated.user",)) + query_params = tuple( + (str(TEST_GROUP_01.group_id), str(user.user_id)) for user in users) + with db.cursor(conn) as cursor: + cursor.executemany( + "INSERT INTO group_users(group_id, user_id) VALUES (?, ?)", + query_params) + + yield (conn, TEST_GROUP_01, users) + + with db.cursor(conn) as cursor: + cursor.executemany( + "DELETE FROM group_users WHERE group_id=? AND user_id=?", + query_params) + +@pytest.fixture(scope="function") +def fxtr_group_roles(fxtr_group, fxtr_roles):# pylint: disable=[redefined-outer-name,unused-argument] + """Link roles to group""" + group_roles = ( + GroupRole(uuid.UUID("9c25efb2-b477-4918-a95c-9914770cbf4d"), + TEST_GROUP_01, RESOURCE_EDITOR_ROLE), + GroupRole(uuid.UUID("82aed039-fe2f-408c-ab1e-81cd1ba96630"), + TEST_GROUP_02, RESOURCE_READER_ROLE)) + conn, groups = fxtr_group + with db.cursor(conn) as cursor: + cursor.executemany( + "INSERT INTO group_roles VALUES (?, ?, ?)", + ((str(role.group_role_id), str(role.group.group_id), + str(role.role.role_id)) + for role in group_roles)) + + yield conn, groups, group_roles + + with db.cursor(conn) as cursor: + cursor.execute("SELECT * FROM group_user_roles_on_resources") + cursor.executemany( + ("DELETE FROM group_roles " + "WHERE group_role_id=? AND group_id=? AND role_id=?"), + ((str(role.group_role_id), str(role.group.group_id), + str(role.role.role_id)) + for role in group_roles)) + +@pytest.fixture(scope="function") +def fxtr_group_user_roles(fxtr_resources, fxtr_group_roles, fxtr_users_in_group):#pylint: disable=[redefined-outer-name,unused-argument] + """Assign roles to users.""" + conn, _groups, group_roles = fxtr_group_roles + _conn, group_resources = fxtr_resources + _conn, _group, group_users = fxtr_users_in_group + users = tuple(user for user in group_users if user.email + not in ("unaff@iliated.user", "group@lead.er")) + users_roles_resources = ( + (user, RESOURCE_EDITOR_ROLE, TEST_RESOURCES_GROUP_01[1]) + for user in users if user.email == "group@mem.ber01") + with db.cursor(conn) as cursor: + params = tuple({ + "group_id": str(resource.group.group_id), + "user_id": str(user.user_id), + "role_id": str(role.role_id), + "resource_id": str(resource.resource_id) + } for user, role, resource in users_roles_resources) + cursor.executemany( + ("INSERT INTO group_user_roles_on_resources " + "VALUES (:group_id, :user_id, :role_id, :resource_id)"), + params) + + yield conn, group_users, group_roles, group_resources + + with db.cursor(conn) as cursor: + cursor.executemany( + ("DELETE FROM group_user_roles_on_resources WHERE " + "group_id=:group_id AND user_id=:user_id AND role_id=:role_id AND " + "resource_id=:resource_id"), + params) diff --git a/tests/unit/auth/fixtures/migration_fixtures.py b/tests/unit/auth/fixtures/migration_fixtures.py new file mode 100644 index 0000000..eb42c2b --- /dev/null +++ b/tests/unit/auth/fixtures/migration_fixtures.py @@ -0,0 +1,51 @@ +"""Fixtures and utilities for migration-related tests""" +import pytest +from yoyo.backends import DatabaseBackend +from yoyo import get_backend, read_migrations +from yoyo.migrations import Migration, MigrationList + +from gn3.auth import db +from gn3.migrations import apply_migrations, rollback_migrations + +@pytest.fixture(scope="session") +def auth_testdb_path(fxtr_app_config): # pylint: disable=redefined-outer-name + """Get the test application's auth database file""" + return fxtr_app_config["AUTH_DB"] + +@pytest.fixture(scope="session") +def auth_migrations_dir(fxtr_app_config): # pylint: disable=redefined-outer-name + """Get the test application's auth database file""" + return fxtr_app_config["AUTH_MIGRATIONS"] + +def apply_single_migration(backend: DatabaseBackend, migration: Migration):# pylint: disable=[redefined-outer-name] + """Utility to apply a single migration""" + apply_migrations(backend, MigrationList([migration])) + +def rollback_single_migration(backend: DatabaseBackend, migration: Migration):# pylint: disable=[redefined-outer-name] + """Utility to rollback a single migration""" + rollback_migrations(backend, MigrationList([migration])) + +@pytest.fixture(scope="session") +def backend(auth_testdb_path):# pylint: disable=redefined-outer-name + """Fixture: retrieve yoyo backend for auth database""" + return get_backend(f"sqlite:///{auth_testdb_path}") + +@pytest.fixture(scope="session") +def all_migrations(auth_migrations_dir): # pylint: disable=redefined-outer-name + """Retrieve all the migrations""" + return read_migrations(auth_migrations_dir) + +@pytest.fixture(scope="function") +def conn_after_auth_migrations(backend, auth_testdb_path, all_migrations): # pylint: disable=redefined-outer-name + """Run all migrations and return a connection to the database after""" + apply_migrations(backend, all_migrations) + with db.connection(auth_testdb_path) as conn: + yield conn + + rollback_migrations(backend, all_migrations) + +def migrations_up_to(migration, migrations_dir): + """Run all the migration before `migration`.""" + migrations = read_migrations(migrations_dir) + index = [mig.path for mig in migrations].index(migration) + return MigrationList(migrations[0:index]) diff --git a/tests/unit/auth/fixtures/oauth2_client_fixtures.py b/tests/unit/auth/fixtures/oauth2_client_fixtures.py new file mode 100644 index 0000000..654d048 --- /dev/null +++ b/tests/unit/auth/fixtures/oauth2_client_fixtures.py @@ -0,0 +1,51 @@ +"""Fixtures for OAuth2 clients""" +import uuid +import json +import datetime + +import pytest + +from gn3.auth import db +from gn3.auth.authentication.users import hash_password +from gn3.auth.authentication.oauth2.models.oauth2client import OAuth2Client + +@pytest.fixture(autouse=True) +def fxtr_patch_envvars(monkeypatch): + """Fixture: patch environment variable""" + monkeypatch.setenv("AUTHLIB_INSECURE_TRANSPORT", "true") + +@pytest.fixture +def fxtr_oauth2_clients(fxtr_users_with_passwords): + """Fixture: Create the OAuth2 clients for use with tests.""" + conn, users = fxtr_users_with_passwords + now = datetime.datetime.now() + + clients = tuple( + OAuth2Client(str(uuid.uuid4()), f"yabadabadoo_{idx:03}", now, + now + datetime.timedelta(hours = 2), + { + "client_name": f"test_client_{idx:03}", + "scope": ["profile", "group", "role", "resource", "register-client"], + "redirect_uri": "/test_oauth2", + "token_endpoint_auth_method": [ + "client_secret_post", "client_secret_basic"], + "grant_types": ["password", "authorisation_code", "refresh_token"], + "response_type": "token" + }, user) + for idx, user in enumerate(users, start=1)) + + with db.cursor(conn) as cursor: + cursor.executemany( + "INSERT INTO oauth2_clients VALUES (?, ?, ?, ?, ?, ?)", + ((str(client.client_id), hash_password(client.client_secret), + int(client.client_id_issued_at.timestamp()), + int(client.client_secret_expires_at.timestamp()), + json.dumps(client.client_metadata), str(client.user.user_id)) + for client in clients)) + + yield conn, clients + + with db.cursor(conn) as cursor: + cursor.executemany( + "DELETE FROM oauth2_clients WHERE client_id=?", + ((str(client.client_id),) for client in clients)) diff --git a/tests/unit/auth/fixtures/resource_fixtures.py b/tests/unit/auth/fixtures/resource_fixtures.py new file mode 100644 index 0000000..117b4f4 --- /dev/null +++ b/tests/unit/auth/fixtures/resource_fixtures.py @@ -0,0 +1,25 @@ +"""Fixtures and utilities for resource-related tests""" +import pytest + +from gn3.auth import db + +from .group_fixtures import TEST_RESOURCES + +@pytest.fixture(scope="function") +def fxtr_resources(fxtr_group):# pylint: disable=[redefined-outer-name] + """fixture: setup test resources in the database""" + conn, _group = fxtr_group + with db.cursor(conn) as cursor: + cursor.executemany( + "INSERT INTO resources VALUES (?,?,?,?,?)", + ((str(res.group.group_id), 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)) + + yield (conn, TEST_RESOURCES) + + with db.cursor(conn) as cursor: + cursor.executemany( + "DELETE FROM resources WHERE group_id=? AND resource_id=?", + ((str(res.group.group_id), str(res.resource_id),) + for res in TEST_RESOURCES)) diff --git a/tests/unit/auth/fixtures/role_fixtures.py b/tests/unit/auth/fixtures/role_fixtures.py new file mode 100644 index 0000000..ee86aa2 --- /dev/null +++ b/tests/unit/auth/fixtures/role_fixtures.py @@ -0,0 +1,45 @@ +"""Fixtures and utilities for role-related tests""" +import uuid + +import pytest + +from gn3.auth import db +from gn3.auth.authorisation.roles import Role +from gn3.auth.authorisation.privileges import Privilege + +RESOURCE_READER_ROLE = Role( + uuid.UUID("c3ca2507-ee24-4835-9b31-8c21e1c072d3"), "resource_reader", True, + (Privilege("group:resource:view-resource", + "view a resource and use it in computations"),)) + +RESOURCE_EDITOR_ROLE = Role( + uuid.UUID("89819f84-6346-488b-8955-86062e9eedb7"), "resource_editor", True, + ( + Privilege("group:resource:view-resource", + "view a resource and use it in computations"), + Privilege("group:resource:edit-resource", "edit/update a resource"))) + +TEST_ROLES = (RESOURCE_READER_ROLE, RESOURCE_EDITOR_ROLE) + +@pytest.fixture(scope="function") +def fxtr_roles(conn_after_auth_migrations): + """Setup some example roles.""" + with db.cursor(conn_after_auth_migrations) as cursor: + cursor.executemany( + ("INSERT INTO roles VALUES (?, ?, ?)"), + ((str(role.role_id), role.role_name, 1) for role in TEST_ROLES)) + cursor.executemany( + ("INSERT INTO role_privileges VALUES (?, ?)"), + ((str(role.role_id), str(privilege.privilege_id)) + for role in TEST_ROLES for privilege in role.privileges)) + + yield conn_after_auth_migrations, TEST_ROLES + + with db.cursor(conn_after_auth_migrations) as cursor: + cursor.executemany( + ("DELETE FROM role_privileges WHERE role_id=? AND privilege_id=?"), + ((str(role.role_id), str(privilege.privilege_id)) + for role in TEST_ROLES for privilege in role.privileges)) + cursor.executemany( + ("DELETE FROM roles WHERE role_id=?"), + ((str(role.role_id),) for role in TEST_ROLES)) diff --git a/tests/unit/auth/fixtures/user_fixtures.py b/tests/unit/auth/fixtures/user_fixtures.py new file mode 100644 index 0000000..d248f54 --- /dev/null +++ b/tests/unit/auth/fixtures/user_fixtures.py @@ -0,0 +1,66 @@ +"""Fixtures and utilities for user-related tests""" +import uuid + +import pytest + +from gn3.auth import db +from gn3.auth.authentication.users import User, hash_password + +TEST_USERS = ( + User(uuid.UUID("ecb52977-3004-469e-9428-2a1856725c7f"), "group@lead.er", + "Group Leader"), + User(uuid.UUID("21351b66-8aad-475b-84ac-53ce528451e3"), + "group@mem.ber01", "Group Member 01"), + User(uuid.UUID("ae9c6245-0966-41a5-9a5e-20885a96bea7"), + "group@mem.ber02", "Group Member 02"), + User(uuid.UUID("9a0c7ce5-2f40-4e78-979e-bf3527a59579"), + "unaff@iliated.user", "Unaffiliated User")) + +@pytest.fixture(scope="function") +def fxtr_users(conn_after_auth_migrations):# pylint: disable=[redefined-outer-name] + """Fixture: setup test users.""" + query = "INSERT INTO users(user_id, email, name) VALUES (?, ?, ?)" + query_user_roles = "INSERT INTO user_roles(user_id, role_id) VALUES (?, ?)" + test_user_roles = ( + ("ecb52977-3004-469e-9428-2a1856725c7f", + "a0e67630-d502-4b9f-b23f-6805d0f30e30"), + ("ecb52977-3004-469e-9428-2a1856725c7f", + "ade7e6b0-ba9c-4b51-87d0-2af7fe39a347")) + 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)) + cursor.executemany(query_user_roles, test_user_roles) + + yield (conn_after_auth_migrations, TEST_USERS) + + with db.cursor(conn_after_auth_migrations) as cursor: + cursor.executemany( + "DELETE FROM user_roles WHERE user_id=?", + (("ecb52977-3004-469e-9428-2a1856725c7f",),)) + cursor.executemany( + "DELETE FROM users WHERE user_id=?", + (("ecb52977-3004-469e-9428-2a1856725c7f",), + ("21351b66-8aad-475b-84ac-53ce528451e3",), + ("ae9c6245-0966-41a5-9a5e-20885a96bea7",), + ("9a0c7ce5-2f40-4e78-979e-bf3527a59579",))) + +@pytest.fixture(scope="function") +def fxtr_users_with_passwords(fxtr_users): # pylint: disable=[redefined-outer-name] + """Fixture: add passwords to the users""" + conn, users = fxtr_users + user_passwords_params = tuple( + (str(user.user_id), hash_password( + f"password_for_user_{idx:03}".encode("utf8"))) + for idx, user in enumerate(users, start=1)) + + with db.cursor(conn) as cursor: + cursor.executemany( + "INSERT INTO user_credentials VALUES (?, ?)", + user_passwords_params) + + yield conn, users + + with db.cursor(conn) as cursor: + cursor.executemany( + "DELETE FROM user_credentials WHERE user_id=?", + ((item[0],) for item in user_passwords_params)) -- cgit v1.2.3