diff --git a/src/globus_sdk/_testing/data/transfer/endpoint_manager_task_list.py b/src/globus_sdk/_testing/data/transfer/endpoint_manager_task_list.py new file mode 100644 index 000000000..1bad1ccc4 --- /dev/null +++ b/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", + }, + ), +) diff --git a/tests/functional/services/transfer/endpoint_manager/test_task_list.py b/tests/functional/services/transfer/endpoint_manager/test_task_list.py index eb45c68a4..abce649a3 100644 --- a/tests/functional/services/transfer/endpoint_manager/test_task_list.py +++ b/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,6 +41,7 @@ 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 @@ -53,6 +49,7 @@ def test_filter_status_list(client): 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)