From b14415a8fb4b288c735b63d523104ea136098c12 Mon Sep 17 00:00:00 2001 From: Gianluca Ficarelli Date: Mon, 20 Oct 2025 14:12:42 +0200 Subject: [PATCH] Add tests for authorization triggers --- app/db/triggers.py | 14 +- tests/conftest.py | 50 +++++ tests/test_cell_morphology.py | 68 ++++-- tests/test_electrical_recording_stimulus.py | 28 +++ tests/test_em_cell_mesh.py | 55 ++++- tests/test_emodel.py | 133 ++++------- tests/test_measurement_annotation.py | 4 +- tests/test_memodel.py | 235 ++++++-------------- tests/test_memodel_calibration_result.py | 55 ++++- tests/{utils.py => utils/__init__.py} | 14 +- tests/utils/api_create.py | 82 +++++++ tests/utils/check.py | 118 ++++++++++ tests/utils/db_create.py | 18 ++ 13 files changed, 570 insertions(+), 304 deletions(-) rename tests/{utils.py => utils/__init__.py} (99%) create mode 100644 tests/utils/api_create.py create mode 100644 tests/utils/check.py create mode 100644 tests/utils/db_create.py diff --git a/app/db/triggers.py b/app/db/triggers.py index ebf84ec62..6c8678bd8 100644 --- a/app/db/triggers.py +++ b/app/db/triggers.py @@ -140,18 +140,18 @@ def unauthorized_private_reference_trigger(model: type[Entity], field_name: str) # list of protected relationships between entities as (model, field_name) protected_entity_relationships = [ (BrainAtlasRegion, "brain_atlas_id"), - (CellMorphology, "cell_morphology_protocol_id"), + (CellMorphology, "cell_morphology_protocol_id"), # tested (Circuit, "atlas_id"), (Circuit, "root_circuit_id"), - (ElectricalRecordingStimulus, "recording_id"), - (EMCellMesh, "em_dense_reconstruction_dataset_id"), - (EModel, "exemplar_morphology_id"), + (ElectricalRecordingStimulus, "recording_id"), # tested + (EMCellMesh, "em_dense_reconstruction_dataset_id"), # tested + (EModel, "exemplar_morphology_id"), # tested (ExperimentalBoutonDensity, "subject_id"), (ExperimentalNeuronDensity, "subject_id"), (ExperimentalSynapsesPerConnection, "subject_id"), - (MEModel, "emodel_id"), - (MEModel, "morphology_id"), - (MEModelCalibrationResult, "calibrated_entity_id"), + (MEModel, "emodel_id"), # tested + (MEModel, "morphology_id"), # tested + (MEModelCalibrationResult, "calibrated_entity_id"), # tested (ScientificArtifact, "subject_id"), (Simulation, "entity_id"), (Simulation, "simulation_campaign_id"), diff --git a/tests/conftest.py b/tests/conftest.py index 8d5ad876e..ae3a6d0e9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1470,6 +1470,24 @@ def em_dense_reconstruction_dataset_json_data(subject_id, brain_region_id): @pytest.fixture def em_dense_reconstruction_dataset(db, em_dense_reconstruction_dataset_json_data, person_id): + return add_db( + db, + EMDenseReconstructionDataset( + **em_dense_reconstruction_dataset_json_data + | { + "created_by_id": person_id, + "updated_by_id": person_id, + "authorized_public": False, + "authorized_project_id": PROJECT_ID, + } + ), + ) + + +@pytest.fixture +def public_em_dense_reconstruction_dataset( + db, em_dense_reconstruction_dataset_json_data, person_id +): return add_db( db, EMDenseReconstructionDataset( @@ -1498,6 +1516,22 @@ def em_cell_mesh_json_data(em_dense_reconstruction_dataset, subject_id, brain_re } +@pytest.fixture +def public_em_cell_mesh_json_data( + public_em_dense_reconstruction_dataset, subject_id, brain_region_id +): + return { + "subject_id": str(subject_id), + "brain_region_id": str(brain_region_id), + "release_version": 1, + "dense_reconstruction_cell_id": 2**63 - 1, # max signed bigint + "generation_method": "marching_cubes", + "level_of_detail": 10, + "mesh_type": "static", + "em_dense_reconstruction_dataset_id": str(public_em_dense_reconstruction_dataset.id), + } + + @pytest.fixture def em_cell_mesh(db, em_cell_mesh_json_data, person_id): return add_db( @@ -1523,6 +1557,22 @@ def cell_morphology_protocol_json_data(): @pytest.fixture def cell_morphology_protocol(db, cell_morphology_protocol_json_data, person_id): + return add_db( + db, + PlaceholderCellMorphologyProtocol( + **cell_morphology_protocol_json_data + | { + "created_by_id": person_id, + "updated_by_id": person_id, + "authorized_project_id": PROJECT_ID, + "authorized_public": False, + } + ), + ) + + +@pytest.fixture +def public_cell_morphology_protocol(db, cell_morphology_protocol_json_data, person_id): return add_db( db, PlaceholderCellMorphologyProtocol( diff --git a/tests/test_cell_morphology.py b/tests/test_cell_morphology.py index 6093a0c09..7fb030ff9 100644 --- a/tests/test_cell_morphology.py +++ b/tests/test_cell_morphology.py @@ -28,13 +28,15 @@ check_entity_update_one, create_cell_morphology_id, ) +from tests.utils.api_create import create_cell_morphology_protocol_id +from tests.utils.check import check_auth_triggers ROUTE = "/cell-morphology" ADMIN_ROUTE = "/admin/cell-morphology" @pytest.fixture -def json_data(subject_id, license_id, brain_region_id, cell_morphology_protocol): +def common_json_data(subject_id, license_id, brain_region_id): return { "brain_region_id": str(brain_region_id), "subject_id": str(subject_id), @@ -43,12 +45,25 @@ def json_data(subject_id, license_id, brain_region_id, cell_morphology_protocol) "location": {"x": 10, "y": 20, "z": 30}, "legacy_id": ["Test Legacy ID"], "license_id": str(license_id), - "cell_morphology_protocol_id": str(cell_morphology_protocol.id), "contact_email": "test@example.com", "experiment_date": "2025-01-01T00:00:00", } +@pytest.fixture +def json_data(common_json_data, cell_morphology_protocol): + return common_json_data | { + "cell_morphology_protocol_id": str(cell_morphology_protocol.id), + } + + +@pytest.fixture +def public_json_data(common_json_data, public_cell_morphology_protocol): + return common_json_data | { + "cell_morphology_protocol_id": str(public_cell_morphology_protocol.id), + } + + def test_create_one( client, brain_region_id, @@ -92,13 +107,13 @@ def test_create_one( assert data[0]["cell_morphology_protocol"] == expected_cell_morphology_protocol_json_data -def test_delete_one(db, clients, json_data): +def test_delete_one(db, clients, public_json_data): check_entity_delete_one( db=db, clients=clients, route=ROUTE, admin_route=ADMIN_ROUTE, - json_data=json_data, + json_data=public_json_data, expected_counts_before={CellMorphology: 1, CellMorphologyProtocol: 1}, expected_counts_after={ CellMorphology: 0, @@ -107,12 +122,12 @@ def test_delete_one(db, clients, json_data): ) -def test_update_one(clients, json_data): +def test_update_one(clients, public_json_data): check_entity_update_one( route=ROUTE, admin_route=ADMIN_ROUTE, clients=clients, - json_data=json_data, + json_data=public_json_data, patch_payload={ "name": "name", "description": "description", @@ -478,24 +493,35 @@ def test_query_cell_morphology_species_join(db, client, brain_region_id, subject } -def test_authorization( +def test_authorization(client_user_1, client_user_2, client_no_project, public_json_data): + check_authorization(ROUTE, client_user_1, client_user_2, client_no_project, public_json_data) + + +def test_auth_triggers( client_user_1, client_user_2, - client_no_project, - subject_id, - license_id, - brain_region_id, + public_json_data, + cell_morphology_protocol, + public_cell_morphology_protocol, ): - json_data = { - "location": {"x": 10, "y": 20, "z": 30}, - "brain_region_id": str(brain_region_id), - "description": "morph description", - "legacy_id": ["Test Legacy ID"], - "license_id": license_id, - "name": "Test Morphology Name", - "subject_id": str(subject_id), - } - check_authorization(ROUTE, client_user_1, client_user_2, client_no_project, json_data) + linked_private_u2 = create_cell_morphology_protocol_id( + client_user_2, + authorized_public=False, + ) + linked_public_u2 = create_cell_morphology_protocol_id( + client_user_2, + authorized_public=True, + ) + check_auth_triggers( + ROUTE, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="cell_morphology_protocol_id", + linked_private_u1_id=str(cell_morphology_protocol.id), + linked_public_u1_id=str(public_cell_morphology_protocol.id), + linked_private_u2_id=linked_private_u2, + linked_public_u2_id=linked_public_u2, + ) def test_pagination(db, client, brain_region_id, person_id): diff --git a/tests/test_electrical_recording_stimulus.py b/tests/test_electrical_recording_stimulus.py index 18e064380..2e4148a26 100644 --- a/tests/test_electrical_recording_stimulus.py +++ b/tests/test_electrical_recording_stimulus.py @@ -12,7 +12,9 @@ check_entity_update_one, check_missing, check_pagination, + create_electrical_cell_recording_id, ) +from tests.utils.check import check_auth_triggers ROUTE = "/electrical-recording-stimulus" ADMIN_ROUTE = "/admin/electrical-recording-stimulus" @@ -122,6 +124,32 @@ def test_authorization( check_authorization(ROUTE, client_user_1, client_user_2, client_no_project, public_json_data) +def test_auth_triggers( + client_user_1, + client_user_2, + public_json_data, + trace_id_minimal, + public_trace_id_minimal, + electrical_cell_recording_json_data, +): + linked_private_u2 = create_electrical_cell_recording_id( + client_user_2, electrical_cell_recording_json_data | {"authorized_public": False} + ) + linked_public_u2 = create_electrical_cell_recording_id( + client_user_2, electrical_cell_recording_json_data | {"authorized_public": True} + ) + check_auth_triggers( + ROUTE, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="recording_id", + linked_private_u1_id=trace_id_minimal, + linked_public_u1_id=public_trace_id_minimal, + linked_private_u2_id=linked_private_u2, + linked_public_u2_id=linked_public_u2, + ) + + def test_pagination(client, create_id): check_pagination(ROUTE, client, create_id) diff --git a/tests/test_em_cell_mesh.py b/tests/test_em_cell_mesh.py index 205eaa695..5c60f634a 100644 --- a/tests/test_em_cell_mesh.py +++ b/tests/test_em_cell_mesh.py @@ -1,10 +1,11 @@ import pytest -from app.db.model import EMCellMesh +from app.db.model import EMCellMesh, EMDenseReconstructionDataset from app.db.types import EntityType from .utils import ( PROJECT_ID, + UNRELATED_PROJECT_ID, add_all_db, assert_request, check_authorization, @@ -13,6 +14,8 @@ check_missing, check_pagination, ) +from tests.utils.check import check_auth_triggers +from tests.utils.db_create import create_entity ROUTE = "/em-cell-mesh" ADMIN_ROUTE = "/admin/em-cell-mesh" @@ -24,6 +27,11 @@ def json_data(em_cell_mesh_json_data): return em_cell_mesh_json_data +@pytest.fixture +def public_json_data(public_em_cell_mesh_json_data): + return public_em_cell_mesh_json_data + + @pytest.fixture def model(em_cell_mesh): return em_cell_mesh @@ -68,8 +76,45 @@ def test_missing(client): check_missing(ROUTE, client) -def test_authorization(client_user_1, client_user_2, client_no_project, json_data): - check_authorization(ROUTE, client_user_1, client_user_2, client_no_project, json_data) +def test_authorization(client_user_1, client_user_2, client_no_project, public_json_data): + check_authorization(ROUTE, client_user_1, client_user_2, client_no_project, public_json_data) + + +def test_auth_triggers( + client_user_1, + db, + person_id, + public_json_data, + em_dense_reconstruction_dataset, + public_em_dense_reconstruction_dataset, + em_dense_reconstruction_dataset_json_data, +): + linked_private_u2 = create_entity( + db, + EMDenseReconstructionDataset, + person_id=person_id, + authorized_public=False, + authorized_project_id=UNRELATED_PROJECT_ID, + json_data=em_dense_reconstruction_dataset_json_data, + ) + linked_public_u2 = create_entity( + db, + EMDenseReconstructionDataset, + person_id=person_id, + authorized_public=True, + authorized_project_id=UNRELATED_PROJECT_ID, + json_data=em_dense_reconstruction_dataset_json_data, + ) + check_auth_triggers( + ROUTE, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="em_dense_reconstruction_dataset_id", + linked_private_u1_id=em_dense_reconstruction_dataset.id, + linked_public_u1_id=public_em_dense_reconstruction_dataset.id, + linked_private_u2_id=linked_private_u2.id, + linked_public_u2_id=linked_public_u2.id, + ) def test_pagination(client, create_id): @@ -151,13 +196,13 @@ def test_filtering(client, models, brain_region_id, species_id, strain_id): } -def test_delete_one(db, clients, json_data): +def test_delete_one(db, clients, public_json_data): check_entity_delete_one( db=db, route=ROUTE, admin_route=ADMIN_ROUTE, clients=clients, - json_data=json_data, + json_data=public_json_data, expected_counts_before={ EMCellMesh: 1, }, diff --git a/tests/test_emodel.py b/tests/test_emodel.py index 1f9f485b3..df47f9700 100644 --- a/tests/test_emodel.py +++ b/tests/test_emodel.py @@ -11,11 +11,13 @@ from .utils import ( TEST_DATA_DIR, assert_request, + check_authorization, check_entity_delete_one, check_entity_update_one, create_cell_morphology_id, upload_entity_asset, ) +from tests.utils.check import check_auth_triggers FILE_EXAMPLE_PATH = TEST_DATA_DIR / "example.json" ROUTE = "/emodel" @@ -23,7 +25,7 @@ @pytest.fixture -def json_data(species_id, strain_id, brain_region_id, morphology_id): +def common_json_data(species_id, strain_id, brain_region_id): return { "brain_region_id": str(brain_region_id), "species_id": species_id, @@ -33,7 +35,22 @@ def json_data(species_id, strain_id, brain_region_id, morphology_id): "iteration": "test iteration", "score": -1, "seed": -1, + } + + +@pytest.fixture +def json_data(common_json_data, morphology_id): + return common_json_data | { "exemplar_morphology_id": morphology_id, + "authorized_public": False, + } + + +@pytest.fixture +def public_json_data(common_json_data, public_morphology_id): + return common_json_data | { + "exemplar_morphology_id": public_morphology_id, + "authorized_public": True, } @@ -58,11 +75,6 @@ def test_create_emodel(client: TestClient, species_id, strain_id, brain_region_i assert data[0]["created_by"]["id"] == data[0]["updated_by"]["id"] -@pytest.fixture -def public_json_data(json_data, public_morphology_id): - return json_data | {"exemplar_morphology_id": str(public_morphology_id)} - - def test_update_one(clients, public_json_data): check_entity_update_one( route=ROUTE, @@ -317,108 +329,43 @@ def test_facets(client: TestClient, faceted_emodel_ids: EModelIds, ion_channel_m ] -def test_authorization( +def test_authorization(client_user_1, client_user_2, client_no_project, public_json_data): + check_authorization( + ROUTE, client_user_1, client_user_2, client_no_project, json_data=public_json_data + ) + + +def test_auth_triggers( client_user_1, client_user_2, - client_no_project, - species_id, - strain_id, + public_json_data, subject_id, brain_region_id, morphology_id, + public_morphology_id, ): - public_morphology_id = create_cell_morphology_id( - client_user_1, + linked_private_u2 = create_cell_morphology_id( + client_user_2, subject_id=subject_id, brain_region_id=brain_region_id, - authorized_public=True, - ) - - emodel_json = { - "brain_region_id": str(brain_region_id), - "description": "morph description", - "legacy_id": "Test Legacy ID", - "name": "Test Morphology Name", - "species_id": species_id, - "strain_id": strain_id, - "exemplar_morphology_id": morphology_id, - "score": 0, - "iteration": "0", - "seed": 0, - } - - public_emodel = client_user_1.post( - ROUTE, - json=emodel_json - | {"exemplar_morphology_id": public_morphology_id} - | { - "name": "public emodel", - "authorized_public": True, - }, - ) - assert public_emodel.status_code == 200 - public_emodel = public_emodel.json() - - unauthorized_exemplar_morphology = client_user_2.post(ROUTE, json=emodel_json) - - assert unauthorized_exemplar_morphology.status_code == 403 - - unauthorized_public_with_private_exemplar_morphology = client_user_1.post( - ROUTE, json=emodel_json | {"authorized_public": True} + authorized_public=False, ) - - assert unauthorized_public_with_private_exemplar_morphology.status_code == 403 - - exemplar_morphology_id = create_cell_morphology_id( + linked_public_u2 = create_cell_morphology_id( client_user_2, subject_id=subject_id, brain_region_id=brain_region_id, - authorized_public=False, - ) - - inaccessible_obj = client_user_2.post( - ROUTE, - json=emodel_json - | {"name": "inaccessible emodel", "exemplar_morphology_id": exemplar_morphology_id}, + authorized_public=True, ) - - assert inaccessible_obj.status_code == 200 - - inaccessible_obj = inaccessible_obj.json() - - # Public Morphology reference authorized from private emodel - private_emodel0 = client_user_1.post( + check_auth_triggers( ROUTE, - json=emodel_json - | {"name": "private emodel 0", "exemplar_morphology_id": public_morphology_id}, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="exemplar_morphology_id", + linked_private_u1_id=morphology_id, + linked_public_u1_id=public_morphology_id, + linked_private_u2_id=linked_private_u2, + linked_public_u2_id=linked_public_u2, ) - assert private_emodel0.status_code == 200 - private_emodel0 = private_emodel0.json() - - private_emodel1 = client_user_1.post(ROUTE, json=emodel_json | {"name": "private emodel 1"}) - assert private_emodel1.status_code == 200 - private_emodel1 = private_emodel1.json() - - # only return results that matches the desired project, and public ones - response = client_user_1.get(ROUTE) - data = response.json()["data"] - assert len(data) == 3 - - ids = {row["id"] for row in data} - assert ids == { - public_emodel["id"], - private_emodel0["id"], - private_emodel1["id"], - } - - response = client_user_1.get(f"{ROUTE}/{inaccessible_obj['id']}") - assert response.status_code == 404 - - # only return public results - response = client_no_project.get(ROUTE) - data = response.json()["data"] - assert len(data) == 1 - assert data[0]["id"] == public_emodel["id"] def test_pagination(client, create_emodel_ids): diff --git a/tests/test_measurement_annotation.py b/tests/test_measurement_annotation.py index e02a6d449..1cd750cd8 100644 --- a/tests/test_measurement_annotation.py +++ b/tests/test_measurement_annotation.py @@ -52,8 +52,8 @@ def test_update_one(clients, json_data, subject_id, brain_region_id): new_morph_id = create_cell_morphology_id( clients.user_1, - subject_id, - brain_region_id, + subject_id=subject_id, + brain_region_id=brain_region_id, authorized_public=True, ) diff --git a/tests/test_memodel.py b/tests/test_memodel.py index 446234f06..0f431e795 100644 --- a/tests/test_memodel.py +++ b/tests/test_memodel.py @@ -17,24 +17,33 @@ from .utils import ( PROJECT_ID, assert_request, + check_authorization, check_brain_region_filter, check_entity_delete_one, check_entity_update_one, create_cell_morphology_id, ) +from tests.utils.api_create import create_emodel_id +from tests.utils.check import check_auth_triggers ROUTE = "/memodel" ADMIN_ROUTE = "/admin/memodel" @pytest.fixture -def json_data(brain_region_id, species_id, strain_id, morphology_id, emodel_id): +def common_json_data(brain_region_id, species_id, strain_id): return { "brain_region_id": str(brain_region_id), "species_id": species_id, "strain_id": strain_id, "description": "Test MEModel Description", "name": "Test MEModel Name", + } + + +@pytest.fixture +def json_data(common_json_data, morphology_id, emodel_id): + return common_json_data | { "morphology_id": morphology_id, "emodel_id": emodel_id, "authorized_public": False, @@ -42,18 +51,11 @@ def json_data(brain_region_id, species_id, strain_id, morphology_id, emodel_id): @pytest.fixture -def public_json_data( - brain_region_id, species_id, strain_id, public_morphology_id, public_emodel_id -): - return { - "brain_region_id": str(brain_region_id), - "species_id": species_id, - "strain_id": strain_id, - "description": "Test MEModel Description", - "name": "Test MEModel Name", +def public_json_data(common_json_data, public_morphology_id, public_emodel_id): + return common_json_data | { "morphology_id": str(public_morphology_id), "emodel_id": str(public_emodel_id), - "authorized_public": False, + "authorized_public": True, } @@ -514,187 +516,76 @@ def test_memodel_search(client: TestClient, faceted_memodels: MEModels): # noqa assert all(d["description"] == "foo" for d in data) -def test_authorization( - client_user_1: TestClient, - client_user_2: TestClient, +def test_authorization(client_user_1, client_user_2, client_no_project, public_json_data): + check_authorization( + ROUTE, client_user_1, client_user_2, client_no_project, json_data=public_json_data + ) + + +def test_auth_triggers( + client_user_1, + client_user_2, + public_json_data, + subject_id, species_id, strain_id, brain_region_id, morphology_id, - subject_id, + public_morphology_id, emodel_id, + public_emodel_id, ): - public_morphology_id = create_cell_morphology_id( - client_user_1, + # check memodel.morphology_id + linked_private_u2 = create_cell_morphology_id( + client_user_2, subject_id=subject_id, brain_region_id=brain_region_id, - authorized_public=True, - ) - - # Different user but public accessible - public_emodel_id = client_user_2.post( - "/emodel", - json={ - "brain_region_id": str(brain_region_id), - "description": "morph description", - "legacy_id": "Test Legacy ID", - "name": "Test Morphology Name", - "species_id": species_id, - "strain_id": strain_id, - "exemplar_morphology_id": public_morphology_id, - "score": 0, - "iteration": "0", - "seed": 0, - "authorized_public": True, - }, - ).json()["id"] - - json = { - "brain_region_id": str(brain_region_id), - "description": "description", - "legacy_id": "Test Legacy ID", - "name": "Test name", - "species_id": species_id, - "strain_id": strain_id, - "emodel_id": emodel_id, - "morphology_id": morphology_id, - } - - public_obj = client_user_1.post( - ROUTE, - json=json - | {"emodel_id": public_emodel_id, "morphology_id": public_morphology_id} - | { - "name": "public obj", - "authorized_public": True, - }, - ) - assert public_obj.status_code == 200 - public_obj = public_obj.json() - - unauthorized_relations = client_user_2.post( - ROUTE, - json=json, - ) - - assert unauthorized_relations.status_code == 403 - - unauthorized_public_with_private_relations = client_user_1.post( - ROUTE, - json=json | {"authorized_public": True}, + authorized_public=False, ) - - assert unauthorized_public_with_private_relations.status_code == 403 - - morphology_id = create_cell_morphology_id( + linked_public_u2 = create_cell_morphology_id( client_user_2, subject_id=subject_id, brain_region_id=brain_region_id, - authorized_public=False, - ) - - unauthorized_emodel = client_user_2.post( - ROUTE, - json=json | {"morphology_id": morphology_id}, + authorized_public=True, ) - - assert unauthorized_emodel.status_code == 403 - - morphology_id_2 = ( - client_user_2.post( - "/cell-morphology", - json={ - "name": "test", - "description": "test", - "subject_id": str(subject_id), - "brain_region_id": str(brain_region_id), - "location": None, - "legacy_id": None, - "authorized_public": True, - }, - ) - ).json()["id"] - - emodel_id = ( - client_user_2.post( - "/emodel", - json={ - "brain_region_id": str(brain_region_id), - "species_id": species_id, - "exemplar_morphology_id": morphology_id_2, - "description": "test", - "name": "test", - "iteration": "test", - "seed": 0, - "score": 0, - "authorized_public": True, - }, - ) - ).json()["id"] - - inaccessible_obj = client_user_2.post( + check_auth_triggers( ROUTE, - json=json | {"morphology_id": morphology_id_2, "emodel_id": emodel_id}, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="morphology_id", + linked_private_u1_id=morphology_id, + linked_public_u1_id=public_morphology_id, + linked_private_u2_id=linked_private_u2, + linked_public_u2_id=linked_public_u2, ) - - assert inaccessible_obj.status_code == 200 - - inaccessible_obj = inaccessible_obj.json() - - # Public reference from private entity authorized - private_obj0 = client_user_1.post( - ROUTE, - json=json - | { - "name": "private obj 0", - "morphology_id": public_morphology_id, - "emodel_id": public_emodel_id, - }, + # check memodel.emodel_id + linked_private_u2 = create_emodel_id( + client_user_2, + species_id=species_id, + strain_id=strain_id, + brain_region_id=brain_region_id, + morphology_id=public_morphology_id, + authorized_public=False, ) - assert private_obj0.status_code == 200 - private_obj0 = private_obj0.json() - - private_obj1 = client_user_1.post( - ROUTE, - json=json - | { - "name": "private obj 1", - }, + linked_public_u2 = create_emodel_id( + client_user_2, + species_id=species_id, + strain_id=strain_id, + brain_region_id=brain_region_id, + morphology_id=public_morphology_id, + authorized_public=True, ) - assert private_obj1.status_code == 200 - private_obj1 = private_obj1.json() - - public_obj_diff_project = client_user_1.post( + check_auth_triggers( ROUTE, - json=json - | { - "morphology_id": morphology_id_2, - "emodel_id": emodel_id, - "authorized_public": True, - }, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="emodel_id", + linked_private_u1_id=emodel_id, + linked_public_u1_id=public_emodel_id, + linked_private_u2_id=linked_private_u2, + linked_public_u2_id=linked_public_u2, ) - assert public_obj_diff_project.status_code == 200 - - public_obj_diff_project = public_obj_diff_project.json() - - # only return results that matches the desired project, and public ones - response = client_user_1.get(ROUTE) - data = response.json()["data"] - assert len(data) == 4 - - ids = {row["id"] for row in data} - assert ids == { - public_obj["id"], - private_obj0["id"], - private_obj1["id"], - public_obj_diff_project["id"], - } - - response = client_user_1.get(f"{ROUTE}/{inaccessible_obj['id']}") - - assert response.status_code == 404 - def test_brain_region_filter( db, client, brain_region_hierarchy_id, species_id, morphology_id, emodel_id, person_id diff --git a/tests/test_memodel_calibration_result.py b/tests/test_memodel_calibration_result.py index 3bc61afe7..08ace1585 100644 --- a/tests/test_memodel_calibration_result.py +++ b/tests/test_memodel_calibration_result.py @@ -5,7 +5,14 @@ from app.db.model import MEModelCalibrationResult -from .utils import assert_request, check_entity_delete_one, check_entity_update_one +from .utils import ( + assert_request, + check_authorization, + check_entity_delete_one, + check_entity_update_one, +) +from tests.utils.api_create import create_memodel_id +from tests.utils.check import check_auth_triggers MODEL = MEModelCalibrationResult ROUTE = "/memodel-calibration-result" @@ -113,6 +120,52 @@ def test_missing(client): assert response.status_code == 422 +def test_authorization(client_user_1, client_user_2, client_no_project, public_json_data): + check_authorization(ROUTE, client_user_1, client_user_2, client_no_project, public_json_data) + + +def test_auth_triggers( + client_user_1, + client_user_2, + species_id, + strain_id, + brain_region_id, + public_json_data, + memodel_id, + public_memodel_id, + public_morphology_id, + public_emodel_id, +): + linked_private_u2_id = create_memodel_id( + client_user_2, + species_id=species_id, + strain_id=strain_id, + brain_region_id=brain_region_id, + morphology_id=public_morphology_id, + emodel_id=public_emodel_id, + authorized_public=False, + ) + linked_public_u2_id = create_memodel_id( + client_user_2, + species_id=species_id, + strain_id=strain_id, + brain_region_id=brain_region_id, + morphology_id=public_morphology_id, + emodel_id=public_emodel_id, + authorized_public=True, + ) + check_auth_triggers( + ROUTE, + client_user_1=client_user_1, + json_data=public_json_data, + link_key="calibrated_entity_id", + linked_private_u1_id=memodel_id, + linked_public_u1_id=public_memodel_id, + linked_private_u2_id=linked_private_u2_id, + linked_public_u2_id=linked_public_u2_id, + ) + + def test_filtering__one_entry(client, memodel_calibration_result_id, memodel_id, morphology_id): # no results expected for unrelated id data = assert_request( diff --git a/tests/utils.py b/tests/utils/__init__.py similarity index 99% rename from tests/utils.py rename to tests/utils/__init__.py index 69a2a6eb6..723da9753 100644 --- a/tests/utils.py +++ b/tests/utils/__init__.py @@ -17,13 +17,17 @@ BrainRegion, BrainRegionHierarchy, CellMorphology, + CellMorphologyProtocol, Contribution, ElectricalCellRecording, ElectricalRecordingStimulus, + EMDenseReconstructionDataset, + EModel, Entity, ETypeClass, ETypeClassification, IonChannelRecording, + MEModel, MTypeClass, MTypeClassification, Person, @@ -31,13 +35,13 @@ Strain, Subject, ) -from app.db.types import EntityType +from app.db.types import CellMorphologyGenerationType as CellMorphologyGenerationType, EntityType from app.routers.asset import EntityRoute from app.utils.uuid import create_uuid DateTimeAdapter = TypeAdapter(datetime) -TEST_DATA_DIR = Path(__file__).parent / "data" +TEST_DATA_DIR = Path(__file__).parent.parent / "data" ADMIN_SUB_ID = "00000000-0000-0000-0000-000000000000" USER_SUB_ID_1 = "00000000-0000-0000-0000-000000000001" @@ -76,8 +80,12 @@ ROUTES = { CellMorphology: "/cell-morphology", + CellMorphologyProtocol: "/cell-morphology-protocol", ElectricalCellRecording: "/electrical-cell-recording", + EMDenseReconstructionDataset: "/em-dense-reconstruction-dataset", + EModel: "/emodel", IonChannelRecording: "/ion-channel-recording", + MEModel: "/memodel", } @@ -118,11 +126,11 @@ class ClientProxies(NamedTuple): def create_cell_morphology_id( client, + *, subject_id, brain_region_id, name="Test Morphology Name", description="Test Morphology Description", - *, authorized_public: bool = False, ): response = client.post( diff --git a/tests/utils/api_create.py b/tests/utils/api_create.py new file mode 100644 index 000000000..af637e936 --- /dev/null +++ b/tests/utils/api_create.py @@ -0,0 +1,82 @@ +from app.db.model import CellMorphologyProtocol, EModel, MEModel +from app.db.types import CellMorphologyGenerationType + +from tests.utils import ROUTES + + +def create_cell_morphology_protocol_id( + client, + *, + authorized_public: bool = False, +): + response = client.post( + ROUTES[CellMorphologyProtocol], + json={ + "generation_type": CellMorphologyGenerationType.placeholder, + "authorized_public": authorized_public, + }, + ) + + assert response.status_code == 200 + return response.json()["id"] + + +def create_emodel_id( + client, + *, + species_id, + strain_id, + brain_region_id, + morphology_id, + name="Test EModel Name", + description="Test EModel Description", + authorized_public: bool = False, +): + response = client.post( + ROUTES[EModel], + json={ + "name": name, + "description": description, + "brain_region_id": str(brain_region_id), + "species_id": str(species_id), + "strain_id": str(strain_id), + "exemplar_morphology_id": morphology_id, + "iteration": "test iteration", + "score": 10, + "seed": -1, + "authorized_public": authorized_public, + }, + ) + + assert response.status_code == 200 + return response.json()["id"] + + +def create_memodel_id( + client, + *, + species_id, + strain_id, + brain_region_id, + morphology_id, + emodel_id, + name="Test MEModel Name", + description="Test MEModel Description", + authorized_public: bool = False, +): + response = client.post( + ROUTES[MEModel], + json={ + "name": name, + "description": description, + "brain_region_id": str(brain_region_id), + "species_id": str(species_id), + "strain_id": str(strain_id), + "morphology_id": str(morphology_id), + "emodel_id": str(emodel_id), + "authorized_public": authorized_public, + }, + ) + + assert response.status_code == 200 + return response.json()["id"] diff --git a/tests/utils/check.py b/tests/utils/check.py new file mode 100644 index 000000000..e5450026a --- /dev/null +++ b/tests/utils/check.py @@ -0,0 +1,118 @@ +import re +import uuid + +from tests.utils import PROJECT_ID, ClientProxy, assert_request + + +def check_auth_triggers( + route, + *, + client_user_1: ClientProxy, + json_data: dict, + link_key: str, + linked_private_u1_id: uuid.UUID | str, + linked_public_u1_id: uuid.UUID | str, + linked_private_u2_id: uuid.UUID | str, + linked_public_u2_id: uuid.UUID | str, +): + """Check the authorization triggers when creating an entity having a linked entity. + + The following table summarizes what are the expected permitted links + between the main entity and the linked entity. + + main/linked | private_u1 | private_u2 | public_u1 | public_u2 + ------------|------------|------------|-----------|----------- + private_u1 | y | n | y | y + public_u1 | n | n | y | y + + Args: + client_user_1: user_1 client. + json_data: json data that will be sent to the create endpoint. + link_key: name of the key that contains the reference to the linked entity. + linked_private_u1_id: id of an existing private entity belonging to user_1. + linked_public_u1_id: id of an existing public entity belonging to user_1. + linked_private_u2_id: id of an existing private entity belonging to user_2. + linked_public_u2_id: id of an existing public entity belonging to user_2. + """ + + err_template = ( + r"One of the entities referenced by .* is not public or not owned by the user .*" + r"unauthorized private reference: .*\.{}" + ) + + # main_private_u1, linked_private_u1 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": False, link_key: str(linked_private_u1_id)}, + expected_status_code=200, + ).json() + assert result["authorized_public"] is False + assert result["authorized_project_id"] == PROJECT_ID + + # main_private_u1, linked_private_u2 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": False, link_key: str(linked_private_u2_id)}, + expected_status_code=403, + ).json() + assert re.match(err_template.format(link_key), result["message"]) + + # main_private_u1, linked_public_u1 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": False, link_key: str(linked_public_u1_id)}, + expected_status_code=200, + ).json() + assert result["authorized_public"] is False + assert result["authorized_project_id"] == PROJECT_ID + + # main_private_u1, linked_public_u2 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": False, link_key: str(linked_public_u2_id)}, + expected_status_code=200, + ).json() + assert result["authorized_public"] is False + assert result["authorized_project_id"] == PROJECT_ID + + # main_public_u1, linked_private_u1 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": True, link_key: str(linked_private_u1_id)}, + expected_status_code=403, + ).json() + assert re.match(err_template.format(link_key), result["message"]) + + # main_public_u1, linked_private_u2 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": True, link_key: str(linked_private_u2_id)}, + expected_status_code=403, + ).json() + assert re.match(err_template.format(link_key), result["message"]) + + # main_public_u1, linked_public_u1 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": True, link_key: str(linked_public_u1_id)}, + expected_status_code=200, + ).json() + assert result["authorized_public"] is True + assert result["authorized_project_id"] == PROJECT_ID + + # main_public_u1, linked_public_u2 + result = assert_request( + client_user_1.post, + url=route, + json=json_data | {"authorized_public": True, link_key: str(linked_public_u2_id)}, + expected_status_code=200, + ).json() + assert result["authorized_public"] is True + assert result["authorized_project_id"] == PROJECT_ID diff --git a/tests/utils/db_create.py b/tests/utils/db_create.py new file mode 100644 index 000000000..ce89fcfa8 --- /dev/null +++ b/tests/utils/db_create.py @@ -0,0 +1,18 @@ +from tests.utils import add_db + + +def create_entity( + db, entity_class, *, person_id, authorized_public, authorized_project_id, json_data +): + return add_db( + db, + entity_class( + **json_data + | { + "created_by_id": person_id, + "updated_by_id": person_id, + "authorized_public": authorized_public, + "authorized_project_id": authorized_project_id, + } + ), + )