diff --git a/pydatalab/src/pydatalab/routes/v0_1/items.py b/pydatalab/src/pydatalab/routes/v0_1/items.py index 0ea1589a5..e549e5fa5 100644 --- a/pydatalab/src/pydatalab/routes/v0_1/items.py +++ b/pydatalab/src/pydatalab/routes/v0_1/items.py @@ -604,6 +604,83 @@ def create_samples(): ) # 207: multi-status +@ITEMS.route("/items//permissions", methods=["PATCH"]) +def update_item_permissions(refcode: str): + """Update the permissions of an item with the given refcode.""" + + request_json = request.get_json() + creator_ids: list[ObjectId] = [] + + if not len(refcode.split(":")) == 2: + refcode = f"{CONFIG.IDENTIFIER_PREFIX}:{refcode}" + + current_item = flask_mongo.db.items.find_one( + {"refcode": refcode, **get_default_permissions(user_only=True)}, + {"_id": 1, "creator_ids": 1}, + ) # type: ignore + + if not current_item: + return ( + jsonify( + { + "status": "error", + "message": f"No valid item found with the given {refcode=}.", + } + ), + 401, + ) + + current_creator_ids = current_item["creator_ids"] + + if "creators" in request_json: + creator_ids = [ + ObjectId(creator.get("immutable_id", None)) + for creator in request_json["creators"] + if creator.get("immutable_id", None) is not None + ] + + # Validate all creator IDs are present in the database + found_ids = [d for d in flask_mongo.db.users.find({"_id": {"$in": creator_ids}}, {"_id": 1})] # type: ignore + if not len(found_ids) == len(creator_ids): + return ( + jsonify( + { + "status": "error", + "message": "One or more creator IDs not found in the database.", + } + ), + 400, + ) + + # Make sure a user cannot remove their own access to an item + current_user_id = current_user.person.immutable_id + try: + creator_ids.remove(current_user_id) + except ValueError: + pass + creator_ids.insert(0, current_user_id) + + # The first ID in the creator list takes precedence; always make sure this is included to avoid orphaned items + if current_creator_ids: + base_owner = current_creator_ids[0] + try: + creator_ids.remove(base_owner) + except ValueError: + pass + creator_ids.insert(0, base_owner) + + LOGGER.warning("Setting permissions for item %s to %s", refcode, creator_ids) + result = flask_mongo.db.items.update_one( + {"refcode": refcode, **get_default_permissions(user_only=True)}, + {"$set": {"creator_ids": creator_ids}}, + ) + + if not result.modified_count == 1: + return jsonify({"status": "error", "message": "Failed to update permissions"}), 400 + + return jsonify({"status": "success"}), 200 + + @ITEMS.route("/delete-sample/", methods=["POST"]) def delete_sample(): request_json = request.get_json() # noqa: F821 pylint: disable=undefined-variable diff --git a/pydatalab/tests/server/test_permissions.py b/pydatalab/tests/server/test_permissions.py index 91ab35594..c1c384a49 100644 --- a/pydatalab/tests/server/test_permissions.py +++ b/pydatalab/tests/server/test_permissions.py @@ -7,7 +7,7 @@ def test_unverified_user_permissions(unverified_client): response = client.get("/samples/") assert response.status_code == 200 - response = client.post("/new-sample/", json={"item_id": "test"}) + response = client.post("/new-sample/", json={"type": "samples", "item_id": "test"}) assert response.status_code == 401 response = client.get("/starting-materials/") @@ -20,7 +20,7 @@ def test_deactivated_user_permissions(deactivated_client): response = client.get("/samples/") assert response.status_code == 200 - response = client.post("/new-sample/", json={"item_id": "test"}) + response = client.post("/new-sample/", json={"type": "samples", "item_id": "test"}) assert response.status_code == 401 response = client.get("/starting-materials/") @@ -33,8 +33,52 @@ def test_unauthenticated_user_permissions(unauthenticated_client): response = client.get("/samples/") assert response.status_code == 401 - response = client.post("/new-sample/", json={"item_id": "test"}) + response = client.post("/new-sample/", json={"type": "samples", "item_id": "test"}) assert response.status_code == 401 response = client.get("/starting-materials/") assert response.status_code == 401 + + +def test_basic_permissions_update(admin_client, admin_user_id, client, user_id): + """Test that an admin can share an item with a normal user.""" + + response = admin_client.post( + "/new-sample/", json={"type": "samples", "item_id": "test-admin-sample"} + ) + assert response.status_code == 201 + + response = admin_client.get("/get-item-data/test-admin-sample") + assert response.status_code == 200 + refcode = response.json["item_data"]["refcode"] + + response = client.get(f"/items/{refcode}") + assert response.status_code == 404 + + # Add normal user to the item + response = admin_client.patch( + f"/items/{refcode}/permissions", json={"creators": [{"immutable_id": str(user_id)}]} + ) + assert response.status_code == 200 + + # Check that they can now see it + response = client.get(f"/items/{refcode}") + assert response.status_code == 200 + + # Check that they cannot remove themselves/the admin from the creators + client.patch(f"/items/{refcode}/permissions", json={"creators": []}) + assert response.status_code == 200 + + response = client.get(f"/items/{refcode}") + assert response.status_code == 200 + + # Check that the admin can remove the user from the permissions + response = admin_client.patch(f"/items/{refcode}/permissions", json={"creators": []}) + assert response.status_code == 200 + + response = client.get(f"/items/{refcode}") + assert response.status_code == 404 + + # but that the admin still remains the creator + response = admin_client.get(f"/items/{refcode}") + assert response.status_code == 200