Add tests for filter_endpoint_use

- Add test data for endpoint_manager_task_list
- Convert existing tests to use that data (minimal changes)
- Add a new test which uses the relevant test data
- Add a new test which exercises the usage error on bad
  `filter_endpoint_use` usage

src/globus_sdk/_testing/data/transfer/endpoint_manager_task_list.py

@@ -0,0 +1,105 @@
+import uuid
+
+from globus_sdk._testing.models import RegisteredResponse, ResponseSet
+
+from ._common import ENDPOINT_ID as SRC_ENDPOINT_ID
+from ._common import TASK_ID
+
+DEST_ENDPOINT_ID = str(uuid.uuid4())
+OWNER_ID = str(uuid.uuid4())
+
+
+RESPONSES = ResponseSet(
+    default=RegisteredResponse(
+        service="transfer",
+        method="GET",
+        path="/endpoint_manager/task_list",
+        metadata={
+            "task_id": TASK_ID,
+            "source": SRC_ENDPOINT_ID,
+            "destination": DEST_ENDPOINT_ID,
+            "owner_id": OWNER_ID,
+        },
+        json={
+            "DATA": [
+                {
+                    "DATA_TYPE": "task",
+                    "bytes_checksummed": 0,
+                    "bytes_transferred": 14,
+                    "canceled_by_admin": None,
+                    "canceled_by_admin_message": None,
+                    "command": "API 0.10",
+                    "completion_time": "2024-02-06T06:59:02+00:00",
+                    "deadline": "2024-02-07T06:58:59+00:00",
+                    "delete_destination_extra": False,
+                    "destination_base_path": None,
+                    "destination_endpoint": f"pliny_the_elder#{DEST_ENDPOINT_ID}",
+                    "destination_endpoint_display_name": "Ercolano",
+                    "destination_endpoint_id": DEST_ENDPOINT_ID,
+                    "destination_host_endpoint": None,
+                    "destination_host_endpoint_display_name": None,
+                    "destination_host_endpoint_id": None,
+                    "destination_host_path": None,
+                    "destination_local_user": "pliny",
+                    "destination_local_user_status": "OK",
+                    "destination_mapped_collection_display_name": None,
+                    "destination_mapped_collection_id": None,
+                    "directories": 2,
+                    "effective_bytes_per_second": 5,
+                    "encrypt_data": False,
+                    "fail_on_quota_errors": False,
+                    "fatal_error": None,
+                    "faults": 0,
+                    "files": 3,
+                    "files_skipped": 0,
+                    "files_transferred": 3,
+                    "filter_rules": None,
+                    "history_deleted": False,
+                    "is_ok": None,
+                    "is_paused": False,
+                    "label": None,
+                    "nice_status": None,
+                    "nice_status_details": None,
+                    "nice_status_expires_in": None,
+                    "nice_status_short_description": None,
+                    "owner_id": OWNER_ID,
+                    "owner_string": "pliny-the-elder@globus.org",
+                    "preserve_timestamp": False,
+                    "recursive_symlinks": "ignore",
+                    "request_time": "2024-02-06T06:58:59+00:00",
+                    "skip_source_errors": False,
+                    "source_base_path": None,
+                    "source_endpoint": f"pliny#{SRC_ENDPOINT_ID}",
+                    "source_endpoint_display_name": "Pompeii",
+                    "source_endpoint_id": SRC_ENDPOINT_ID,
+                    "source_host_endpoint": None,
+                    "source_host_endpoint_display_name": None,
+                    "source_host_endpoint_id": None,
+                    "source_host_path": None,
+                    "source_local_user": None,
+                    "source_local_user_status": "NO_PERMISSION",
+                    "source_mapped_collection_display_name": None,
+                    "source_mapped_collection_id": None,
+                    "status": "SUCCEEDED",
+                    "subtasks_canceled": 0,
+                    "subtasks_expired": 0,
+                    "subtasks_failed": 0,
+                    "subtasks_pending": 0,
+                    "subtasks_retrying": 0,
+                    "subtasks_skipped_errors": 0,
+                    "subtasks_succeeded": 6,
+                    "subtasks_total": 6,
+                    "symlinks": 0,
+                    "sync_level": None,
+                    "task_id": TASK_ID,
+                    "type": "TRANSFER",
+                    "username": "pliny",
+                    "verify_checksum": True,
+                }
+            ],
+            "DATA_TYPE": "task_list",
+            "has_next_page": False,
+            "last_key": "complete,2024-02-06T06:59:02.291996",
+        },
+    ),
+)

tests/functional/services/transfer/endpoint_manager/test_task_list.py

@@ -3,8 +3,8 @@
 
 import pytest
 
-from globus_sdk._testing import get_last_request
-from tests.common import register_api_route
+import globus_sdk
+from globus_sdk._testing import get_last_request, load_response
 
 ZERO_ID = uuid.UUID(int=0)
 
@@ -13,12 +13,6 @@ def get_last_params():
     return get_last_request().params
 
 
-# stub in empty data, this can be explicitly replaced if a test wants specific data
-@pytest.fixture(autouse=True)
-def empty_response():
-    register_api_route("transfer", "/endpoint_manager/task_list", json={"DATA": []})
-
-
 @pytest.mark.parametrize(
     "paramname, paramvalue",
     [
@@ -38,6 +32,7 @@ def empty_response():
     ],
 )
 def test_strsafe_params(client, paramname, paramvalue):
+    load_response(client.endpoint_manager_task_list)
     paramstr = str(paramvalue)
     client.endpoint_manager_task_list(**{paramname: paramvalue})
     params = get_last_params()
@@ -46,13 +41,15 @@ def test_strsafe_params(client, paramname, paramvalue):
 
 
 def test_filter_status_list(client):
+    load_response(client.endpoint_manager_task_list)
     client.endpoint_manager_task_list(filter_status=["ACTIVE", "INACTIVE"])
     params = get_last_params()
     assert "filter_status" in params
     assert params["filter_status"] == "ACTIVE,INACTIVE"
 
 
 def test_filter_task_id_list(client):
+    load_response(client.endpoint_manager_task_list)
     # mixed list of str and UUID
     client.endpoint_manager_task_list(filter_task_id=["foo", ZERO_ID, "bar"])
     params = get_last_params()
@@ -65,6 +62,8 @@ def _fromisoformat(datestr):  # for py3.6, datetime.fromisoformat was added in p
 
 
 def test_filter_completion_time_datetime_tuple(client):
+    load_response(client.endpoint_manager_task_list)
+
     dt1 = _fromisoformat("2020-08-25T00:00:00")
     dt2 = _fromisoformat("2021-08-25T16:05:28")
 
@@ -82,3 +81,32 @@ def test_filter_completion_time_datetime_tuple(client):
     params = get_last_params()
     assert "filter_completion_time" in params
     assert params["filter_completion_time"] == ",2020-08-25T00:00:00"
+
+
+@pytest.mark.parametrize("ep_use", ("source", "destination"))
+def test_filter_by_endpoint_use(client, ep_use):
+    meta = load_response(client.endpoint_manager_task_list).metadata
+    if ep_use == "source":
+        ep_id = meta["source"]
+    else:
+        ep_id = meta["destination"]
+
+    client.endpoint_manager_task_list(filter_endpoint=ep_id, filter_endpoint_use=ep_use)
+    params = get_last_params()
+
+    assert "filter_endpoint" in params
+    assert params["filter_endpoint"] == str(ep_id)
+    assert "filter_endpoint_use" in params
+    assert params["filter_endpoint_use"] == ep_use
+
+
+@pytest.mark.parametrize("ep_use", ("source", "destination"))
+def test_usage_error_on_filter_endpoint_use_without_endpoint(client, ep_use):
+    with pytest.raises(
+        globus_sdk.GlobusSDKUsageError,
+        match=(
+            "`filter_endpoint_use` is only valid when `filter_endpoint` is "
+            r"also supplied\."
+        ),
+    ):
+        client.endpoint_manager_task_list(filter_endpoint_use=ep_use)