Merge pull request #1022 from sirosen/check-uuids

Add UUID checking for add_app_data_access_scope

changelog.d/20240802_143010_sirosen_check_uuids.rst

@@ -0,0 +1,5 @@
+Changed
+~~~~~~~
+
+- ``TransferClient.add_app_data_access_scope`` now raises an error if it is
+  given an invalid collection ID. (:pr:`NUMBER`)

src/globus_sdk/services/transfer/client.py

@@ -143,6 +143,7 @@ def add_app_data_access_scope(self, collection_id: UUIDLike) -> TransferClient:
 
                     res = client.operation_ls(COLLECTION_ID)
         """
+        utils.check_uuid(collection_id, name="collection_id")
         base_scope = Scope(TransferScopes.all)
         data_access_scope = Scope(
             GCSCollectionScopeBuilder(str(collection_id)).data_access,

src/globus_sdk/utils.py

@@ -70,6 +70,38 @@ def b64str(s: str) -> str:
     return b64encode(s.encode("utf-8")).decode("utf-8")
 
 
+def check_uuid(s: UUIDLike, *, name: str) -> t.Literal[True]:
+    """
+    Raise an error if the input is not a UUID
+
+    :param s: the UUID|str value
+    :param name: the name for this value to use in error messages
+
+    :raises TypeError: if the input is not a UUID|str
+    :raises ValueError: if the input is a non-UUID str
+
+    Example usage:
+
+    .. code-block:: python
+
+        def frob_it(collection_id: UUIDLike) -> Frob:
+            utils.check_uuid(collection_id, name="collection_id")
+            return Frob(collection_id)
+
+    """
+    if isinstance(s, uuid.UUID):
+        return True
+    elif not isinstance(s, str):
+        raise TypeError(f"'{name}' must be a UUID or str (value='{s}')")
+
+    try:
+        uuid.UUID(s)
+    except ValueError as e:
+        raise ValueError(f"'{name}' must be a valid UUID (value='{s}')") from e
+
+    return True
+
+
 def slash_join(a: str, b: str | None) -> str:
     """
     Join a and b with a single slash, regardless of whether they already

tests/unit/experimental/test_client_integration.py

@@ -1,3 +1,5 @@
+import uuid
+
 import pytest
 
 import globus_sdk
@@ -44,22 +46,25 @@ def test_transfer_client_default_scopes(app):
 def test_transfer_client_add_app_data_access_scope(app):
     client = globus_sdk.TransferClient(app=app)
 
-    client.add_app_data_access_scope("collection_id")
+    collection_id = str(uuid.UUID(int=0))
+    client.add_app_data_access_scope(collection_id)
     str_list = [str(s) for s in app.get_scope_requirements("transfer.api.globus.org")]
-    expected = "urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/collection_id/data_access]"  # noqa
+    expected = f"urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id}/data_access]"  # noqa: E501
     assert expected in str_list
 
 
 def test_transfer_client_add_app_data_access_scope_chaining(app):
+    collection_id_1 = str(uuid.UUID(int=1))
+    collection_id_2 = str(uuid.UUID(int=2))
     (
         globus_sdk.TransferClient(app=app)
-        .add_app_data_access_scope("collection_id_1")
-        .add_app_data_access_scope("collection_id_2")
+        .add_app_data_access_scope(collection_id_1)
+        .add_app_data_access_scope(collection_id_2)
     )
 
     str_list = [str(s) for s in app.get_scope_requirements("transfer.api.globus.org")]
-    expected_1 = "urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/collection_id_1/data_access]"  # noqa
-    expected_2 = "urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/collection_id_2/data_access]"  # noqa
+    expected_1 = f"urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id_1}/data_access]"  # noqa: E501
+    expected_2 = f"urn:globus:auth:scope:transfer.api.globus.org:all[*https://auth.globus.org/scopes/{collection_id_2}/data_access]"  # noqa: E501
     assert expected_1 in str_list
     assert expected_2 in str_list
 

tests/unit/test_utils.py

@@ -98,3 +98,27 @@ def y(self_or_cls):
 )
 def test_safe_strseq_iter(value, expected_result):
     assert list(utils.safe_strseq_iter(value)) == expected_result
+
+
+@pytest.mark.parametrize("value", (uuid.UUID(int=0), str(uuid.UUID(int=1))))
+def test_check_uuid_ok(value):
+    # no error and returns True
+    assert utils.check_uuid(value, name="foo")
+
+
+@pytest.mark.parametrize("value", (str(uuid.UUID(int=0))[:-1], ""))
+def test_check_uuid_fails_value(value):
+    with pytest.raises(ValueError, match="'foo' must be a valid UUID") as excinfo:
+        utils.check_uuid(value, name="foo")
+
+    err = excinfo.value
+    assert f"value='{value}'" in str(err)
+
+
+@pytest.mark.parametrize("value", (object(), None, ["bar"]))
+def test_check_uuid_fails_type(value):
+    with pytest.raises(TypeError, match="'foo' must be a UUID or str") as excinfo:
+        utils.check_uuid(value, name="foo")
+
+    err = excinfo.value
+    assert f"value='{value}'" in str(err)