From 4462d609460bb7a204367e96893346f31787784a Mon Sep 17 00:00:00 2001 From: Jeremy Stone <74574922+jstone-uw@users.noreply.github.com> Date: Tue, 17 Mar 2026 18:09:41 -0700 Subject: [PATCH 1/2] GET request for multiple score sets --- src/mavedb/routers/score_sets.py | 32 ++++++++++++++++++++ tests/routers/test_score_set.py | 50 ++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index 7376ca4b..bab41df9 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -672,6 +672,38 @@ def search_my_score_sets( return {"score_sets": enriched_score_sets, "num_score_sets": num_score_sets} +@router.get( + "/score-sets/", + status_code=200, + response_model=list[score_set.ScoreSet], + responses={**ACCESS_CONTROL_ERROR_RESPONSES}, + response_model_exclude_none=True, + summary="Fetch score sets by URN list", +) +async def show_score_sets( + *, + urns: str = Query(..., description="Comma-separated list of score set URNs"), + db: Session = Depends(deps.get_db), + user_data: UserData = Depends(get_current_user), +) -> Any: + """ + Fetch score sets identified by a list of URNs. + """ + urn_list = [urn.strip() for urn in urns.split(",") if urn.strip()] + if not urn_list: + raise HTTPException(status_code=422, detail="At least one URN is required") + + save_to_logging_context({"requested_resource": urn_list}) + response_items: list[score_set.ScoreSet] = [] + for urn in urn_list: + item = await fetch_score_set_by_urn(db, urn, user_data, None, False) + enriched_experiment = enrich_experiment_with_num_score_sets(item.experiment, user_data) + response_item = score_set.ScoreSet.model_validate(item).copy(update={"experiment": enriched_experiment}) + response_items.append(response_item) + + return response_items + + @router.get( "/score-sets/{urn}", status_code=200, diff --git a/tests/routers/test_score_set.py b/tests/routers/test_score_set.py index f412b16a..85e17d64 100644 --- a/tests/routers/test_score_set.py +++ b/tests/routers/test_score_set.py @@ -713,6 +713,56 @@ def test_get_own_private_score_set(client, setup_router_db): assert (key, expected_response[key]) == (key, response_data[key]) +def test_get_score_sets_by_comma_separated_urns(client, setup_router_db): + experiment = create_experiment(client) + first_score_set = create_seq_score_set(client, experiment["urn"]) + second_score_set = create_seq_score_set(client, experiment["urn"]) + + response = client.get( + "/api/v1/score-sets/", + params={"urns": f"{first_score_set['urn']}, {second_score_set['urn']}"}, + ) + assert response.status_code == 200 + + response_data = response.json() + assert [item["urn"] for item in response_data] == [first_score_set["urn"], second_score_set["urn"]] + + for item in response_data: + jsonschema.validate(instance=item, schema=ScoreSet.model_json_schema()) + + +def test_get_score_sets_requires_at_least_one_urn(client, setup_router_db): + response = client.get("/api/v1/score-sets/", params={"urns": " , "}) + assert response.status_code == 422 + assert response.json()["detail"] == "At least one URN is required" + + +def test_get_score_sets_with_mixed_valid_and_invalid_urns_returns_404(client, setup_router_db): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + missing_urn = "urn:mavedb:99999999-z-9" + + response = client.get( + "/api/v1/score-sets/", + params={"urns": f"{score_set['urn']},{missing_urn}"}, + ) + assert response.status_code == 404 + assert response.json()["detail"] == f"score set with URN '{missing_urn}' not found" + + +def test_get_score_sets_with_whitespace_around_urns_in_mixed_list_returns_404(client, setup_router_db): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + missing_urn = "urn:mavedb:99999999-z-9" + + response = client.get( + "/api/v1/score-sets/", + params={"urns": f" {score_set['urn']} , {missing_urn} "}, + ) + assert response.status_code == 404 + assert response.json()["detail"] == f"score set with URN '{missing_urn}' not found" + + def test_cannot_get_other_user_private_score_set(session, client, setup_router_db): experiment = create_experiment(client) score_set = create_seq_score_set(client, experiment["urn"]) From 3aaecfa5eac6a5c660c7300a5472037d9c077855 Mon Sep 17 00:00:00 2001 From: Jeremy Stone <74574922+jstone-uw@users.noreply.github.com> Date: Thu, 19 Mar 2026 10:31:23 -0700 Subject: [PATCH 2/2] Unit tests of API-level authorization when getting multiple score sets --- tests/routers/test_score_set.py | 237 ++++++++++++++++++++++---------- 1 file changed, 166 insertions(+), 71 deletions(-) diff --git a/tests/routers/test_score_set.py b/tests/routers/test_score_set.py index 85e17d64..60aeeff3 100644 --- a/tests/routers/test_score_set.py +++ b/tests/routers/test_score_set.py @@ -713,6 +713,97 @@ def test_get_own_private_score_set(client, setup_router_db): assert (key, expected_response[key]) == (key, response_data[key]) +def test_cannot_get_other_user_private_score_set(session, client, setup_router_db): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + change_ownership(session, score_set["urn"], ScoreSetDbModel) + response = client.get(f"/api/v1/score-sets/{score_set['urn']}") + assert response.status_code == 404 + response_data = response.json() + assert f"score set with URN '{score_set['urn']}' not found" in response_data["detail"] + + +def test_anonymous_user_cannot_get_user_private_score_set(session, client, setup_router_db, anonymous_app_overrides): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + change_ownership(session, score_set["urn"], ScoreSetDbModel) + with DependencyOverrider(anonymous_app_overrides): + response = client.get(f"/api/v1/score-sets/{score_set['urn']}") + + assert response.status_code == 404 + response_data = response.json() + assert f"score set with URN '{score_set['urn']}' not found" in response_data["detail"] + + +def test_contributor_can_get_other_users_private_score_set(session, client, setup_router_db): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + change_ownership(session, score_set["urn"], ScoreSetDbModel) + add_contributor( + session, + score_set["urn"], + ScoreSetDbModel, + TEST_USER["username"], + TEST_USER["first_name"], + TEST_USER["last_name"], + ) + + expected_response = update_expected_response_for_created_resources( + deepcopy(TEST_MINIMAL_SEQ_SCORESET_RESPONSE), experiment, score_set + ) + expected_response["contributors"] = [ + { + "recordType": "Contributor", + "orcidId": TEST_USER["username"], + "givenName": TEST_USER["first_name"], + "familyName": TEST_USER["last_name"], + } + ] + expected_response["createdBy"] = { + "recordType": "User", + "orcidId": EXTRA_USER["username"], + "firstName": EXTRA_USER["first_name"], + "lastName": EXTRA_USER["last_name"], + } + expected_response["modifiedBy"] = { + "recordType": "User", + "orcidId": EXTRA_USER["username"], + "firstName": EXTRA_USER["first_name"], + "lastName": EXTRA_USER["last_name"], + } + expected_response["experiment"].update({"numScoreSets": 1}) + + response = client.get(f"/api/v1/score-sets/{score_set['urn']}") + assert response.status_code == 200 + response_data = response.json() + + assert sorted(expected_response.keys()) == sorted(response_data.keys()) + for key in expected_response: + assert (key, expected_response[key]) == (key, response_data[key]) + + +def test_admin_can_get_other_user_private_score_set(session, client, admin_app_overrides, setup_router_db): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + expected_response = update_expected_response_for_created_resources( + deepcopy(TEST_MINIMAL_SEQ_SCORESET_RESPONSE), experiment, score_set + ) + expected_response["experiment"].update({"numScoreSets": 1}) + with DependencyOverrider(admin_app_overrides): + response = client.get(f"/api/v1/score-sets/{score_set['urn']}") + + assert response.status_code == 200 + response_data = response.json() + assert sorted(expected_response.keys()) == sorted(response_data.keys()) + for key in expected_response: + assert (key, expected_response[key]) == (key, response_data[key]) + + +######################################################################################################################## +# Multiple score set fetching +######################################################################################################################## + + def test_get_score_sets_by_comma_separated_urns(client, setup_router_db): experiment = create_experiment(client) first_score_set = create_seq_score_set(client, experiment["urn"]) @@ -763,26 +854,94 @@ def test_get_score_sets_with_whitespace_around_urns_in_mixed_list_returns_404(cl assert response.json()["detail"] == f"score set with URN '{missing_urn}' not found" -def test_cannot_get_other_user_private_score_set(session, client, setup_router_db): +def test_show_score_sets_anonymous_can_fetch_public_score_sets( + session, client, setup_router_db, anonymous_app_overrides, data_provider, data_files +): experiment = create_experiment(client) score_set = create_seq_score_set(client, experiment["urn"]) + score_set = mock_worker_variant_insertion(client, session, data_provider, score_set, data_files / "scores.csv") + with patch.object(arq.ArqRedis, "enqueue_job", return_value=None): + published_score_set = publish_score_set(client, score_set["urn"]) + + with DependencyOverrider(anonymous_app_overrides): + response = client.get( + "/api/v1/score-sets/", + params={"urns": published_score_set["urn"]}, + ) + + assert response.status_code == 200 + response_data = response.json() + assert len(response_data) == 1 + assert response_data[0]["urn"] == published_score_set["urn"] + + +def test_show_score_sets_anonymous_cannot_fetch_private_score_sets(session, client, setup_router_db, anonymous_app_overrides): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + # Score set is private (not published); change ownership so it belongs to another user change_ownership(session, score_set["urn"], ScoreSetDbModel) - response = client.get(f"/api/v1/score-sets/{score_set['urn']}") + + with DependencyOverrider(anonymous_app_overrides): + response = client.get( + "/api/v1/score-sets/", + params={"urns": score_set["urn"]}, + ) + assert response.status_code == 404 + assert f"score set with URN '{score_set['urn']}' not found" in response.json()["detail"] + + +def test_show_score_sets_authenticated_user_can_fetch_own_private_score_sets(client, setup_router_db): + experiment = create_experiment(client) + score_set = create_seq_score_set(client, experiment["urn"]) + + response = client.get( + "/api/v1/score-sets/", + params={"urns": score_set["urn"]}, + ) + + assert response.status_code == 200 response_data = response.json() - assert f"score set with URN '{score_set['urn']}' not found" in response_data["detail"] + assert len(response_data) == 1 + assert response_data[0]["urn"] == score_set["urn"] -def test_anonymous_user_cannot_get_user_private_score_set(session, client, setup_router_db, anonymous_app_overrides): +def test_show_score_sets_authenticated_user_cannot_fetch_other_users_private_score_sets( + session, client, setup_router_db +): experiment = create_experiment(client) score_set = create_seq_score_set(client, experiment["urn"]) change_ownership(session, score_set["urn"], ScoreSetDbModel) + + response = client.get( + "/api/v1/score-sets/", + params={"urns": score_set["urn"]}, + ) + + assert response.status_code == 404 + assert f"score set with URN '{score_set['urn']}' not found" in response.json()["detail"] + + +def test_show_score_sets_mixed_public_and_private_returns_404( + session, client, setup_router_db, anonymous_app_overrides, data_provider, data_files +): + experiment = create_experiment(client) + public_score_set = create_seq_score_set(client, experiment["urn"]) + public_score_set = mock_worker_variant_insertion(client, session, data_provider, public_score_set, data_files / "scores.csv") + private_score_set = create_seq_score_set(client, experiment["urn"]) + with patch.object(arq.ArqRedis, "enqueue_job", return_value=None): + published_score_set = publish_score_set(client, public_score_set["urn"]) + # Make private_score_set belong to a different user to make it inaccessible anonymously + change_ownership(session, private_score_set["urn"], ScoreSetDbModel) + with DependencyOverrider(anonymous_app_overrides): - response = client.get(f"/api/v1/score-sets/{score_set['urn']}") + response = client.get( + "/api/v1/score-sets/", + params={"urns": f"{published_score_set['urn']},{private_score_set['urn']}"}, + ) assert response.status_code == 404 - response_data = response.json() - assert f"score set with URN '{score_set['urn']}' not found" in response_data["detail"] + assert f"score set with URN '{private_score_set['urn']}' not found" in response.json()["detail"] def test_can_add_contributor_in_both_experiment_and_score_set(session, client, setup_router_db): @@ -818,70 +977,6 @@ def test_can_add_contributor_in_both_experiment_and_score_set(session, client, s assert any(c["orcidId"] == TEST_USER["username"] for c in exp_response_data["contributors"]) -def test_contributor_can_get_other_users_private_score_set(session, client, setup_router_db): - experiment = create_experiment(client) - score_set = create_seq_score_set(client, experiment["urn"]) - change_ownership(session, score_set["urn"], ScoreSetDbModel) - add_contributor( - session, - score_set["urn"], - ScoreSetDbModel, - TEST_USER["username"], - TEST_USER["first_name"], - TEST_USER["last_name"], - ) - - expected_response = update_expected_response_for_created_resources( - deepcopy(TEST_MINIMAL_SEQ_SCORESET_RESPONSE), experiment, score_set - ) - expected_response["contributors"] = [ - { - "recordType": "Contributor", - "orcidId": TEST_USER["username"], - "givenName": TEST_USER["first_name"], - "familyName": TEST_USER["last_name"], - } - ] - expected_response["createdBy"] = { - "recordType": "User", - "orcidId": EXTRA_USER["username"], - "firstName": EXTRA_USER["first_name"], - "lastName": EXTRA_USER["last_name"], - } - expected_response["modifiedBy"] = { - "recordType": "User", - "orcidId": EXTRA_USER["username"], - "firstName": EXTRA_USER["first_name"], - "lastName": EXTRA_USER["last_name"], - } - expected_response["experiment"].update({"numScoreSets": 1}) - - response = client.get(f"/api/v1/score-sets/{score_set['urn']}") - assert response.status_code == 200 - response_data = response.json() - - assert sorted(expected_response.keys()) == sorted(response_data.keys()) - for key in expected_response: - assert (key, expected_response[key]) == (key, response_data[key]) - - -def test_admin_can_get_other_user_private_score_set(session, client, admin_app_overrides, setup_router_db): - experiment = create_experiment(client) - score_set = create_seq_score_set(client, experiment["urn"]) - expected_response = update_expected_response_for_created_resources( - deepcopy(TEST_MINIMAL_SEQ_SCORESET_RESPONSE), experiment, score_set - ) - expected_response["experiment"].update({"numScoreSets": 1}) - with DependencyOverrider(admin_app_overrides): - response = client.get(f"/api/v1/score-sets/{score_set['urn']}") - - assert response.status_code == 200 - response_data = response.json() - assert sorted(expected_response.keys()) == sorted(response_data.keys()) - for key in expected_response: - assert (key, expected_response[key]) == (key, response_data[key]) - - @pytest.mark.parametrize( "mock_publication_fetch", [