8 tests pass, 1 deselected

Author: forsyth2

tests/mock/test_globus_refresh.py

@@ -1,11 +1,3 @@
-import json
-from unittest.mock import Mock, mock_open, patch
-
-import pytest
-
-from zstash.globus import globus_block_wait, globus_transfer
-from zstash.globus_utils import load_tokens
-
 """
 # Run all tests
 pytest test_globus_refresh.py -v
@@ -20,14 +12,25 @@
 pytest test_globus_refresh.py -v -s
 """
 
-# Mock Token Expiration #######################################################
+import json
+from unittest.mock import Mock, mock_open, patch
 
+import pytest
 
+from zstash.globus import globus_block_wait, globus_transfer
+from zstash.globus_utils import load_tokens
+
+
+# Verifies that globus_transfer() calls endpoint_autoactivate for both endpoints
 def test_globus_transfer_refreshes_tokens():
     """Test that globus_transfer calls endpoint_autoactivate"""
     with patch("zstash.globus.transfer_client") as mock_client, patch(
         "zstash.globus.local_endpoint", "local-uuid"
-    ), patch("zstash.globus.remote_endpoint", "remote-uuid"):
+    ), patch("zstash.globus.remote_endpoint", "remote-uuid"), patch(
+        "zstash.globus.task_id", None
+    ), patch(
+        "zstash.globus.transfer_data", None
+    ):
 
         mock_client.endpoint_autoactivate = Mock()
         mock_client.operation_ls = Mock(return_value=[])
@@ -46,6 +49,7 @@ def test_globus_transfer_refreshes_tokens():
         assert any("if_expires_in=86400" in str(call) for call in calls)
 
 
+# Confirms periodic refresh during long waits
 def test_globus_block_wait_refreshes_periodically():
     """Test that globus_block_wait refreshes tokens on each retry"""
     with patch("zstash.globus.transfer_client") as mock_client, patch(
@@ -63,11 +67,11 @@ def test_globus_block_wait_refreshes_periodically():
         assert mock_client.endpoint_autoactivate.call_count >= 1
 
 
-# Mock Time to Simulate Expiration ############################################
-
-
+# Validates expiration detection logic
 def test_load_tokens_detects_expiration(caplog):
     """Test that load_tokens detects soon-to-expire tokens"""
+    import time as time_module
+
     # Create a token file with expiration in 30 minutes
     current_time = 1000000
     expires_at = current_time + 1800  # 30 minutes from now
@@ -80,21 +84,20 @@ def test_load_tokens_detects_expiration(caplog):
         }
     }
 
-    with patch("time.time", return_value=current_time), patch(
+    with patch.object(time_module, "time", return_value=current_time), patch(
         "builtins.open", mock_open(read_data=json.dumps(tokens))
     ), patch("os.path.exists", return_value=True):
 
-        result = load_tokens()
+        with caplog.at_level("INFO"):
+            result = load_tokens()
 
         # Check that warning was logged
         assert "expiring soon" in caplog.text
         assert result == tokens
 
 
-# Integration Test with Short Timeout #########################################
-
-
-@pytest.mark.integration  # Mark as integration test
+@pytest.mark.integration
+@pytest.mark.skip(reason="Requires real Globus credentials")
 def test_refresh_mechanism_with_short_token():
     """
     Integration test: Authenticate, manually expire token, verify refresh works.
@@ -105,8 +108,9 @@ def test_refresh_mechanism_with_short_token():
     # Set up with real credentials (skip if no credentials available)
     pytest.importorskip("globus_sdk")
 
-    endpoint1 = "your-test-endpoint-1"
-    endpoint2 = "your-test-endpoint-2"
+    # Use actual endpoint UUIDs if running this test
+    endpoint1 = "your-actual-endpoint-uuid-1"
+    endpoint2 = "your-actual-endpoint-uuid-2"
 
     transfer_client = get_transfer_client_with_auth([endpoint1, endpoint2])
 
@@ -120,9 +124,7 @@ def test_refresh_mechanism_with_short_token():
     assert result is not None
 
 
-#  Stress Test with Rapid Calls ###############################################
-
-
+# Ensures no issues with many rapid refresh calls
 def test_multiple_rapid_refreshes():
     """Test that calling refresh many times doesn't break"""
     with patch("zstash.globus.transfer_client") as mock_client:
@@ -136,15 +138,19 @@ def test_multiple_rapid_refreshes():
         assert mock_client.endpoint_autoactivate.call_count == 100
 
 
-# End-to-End Test with Short Transfer #########################################
-
-
+# End-to-end test with mocked transfer
 def test_small_transfer_with_refresh_enabled():
     """
     Functional test: Transfer a small file and verify refresh calls were made.
-    Uses real Globus but completes in seconds.
     """
-    with patch("zstash.globus.transfer_client") as mock_client:
+    with patch("zstash.globus.transfer_client") as mock_client, patch(
+        "zstash.globus.local_endpoint", "local-uuid"
+    ), patch("zstash.globus.remote_endpoint", "remote-uuid"), patch(
+        "zstash.globus.task_id", None
+    ), patch(
+        "zstash.globus.transfer_data", None
+    ):
+
         # Set up mock to track calls
         mock_client.endpoint_autoactivate = Mock()
         mock_client.submit_transfer = Mock(return_value={"task_id": "test-123"})
@@ -158,56 +164,89 @@ def test_small_transfer_with_refresh_enabled():
         assert mock_client.endpoint_autoactivate.called
 
 
-# Parametrized Test for Different Scenarios ###################################
-
-
+# Tests blocking PUT mode
+# Tests non-blocking PUT mode
 @pytest.mark.parametrize(
     "transfer_type,non_blocking",
     [
         ("put", False),
         ("put", True),
-        ("get", False),
     ],
 )
 def test_globus_transfer_refreshes_in_all_modes(transfer_type, non_blocking):
     """Test that token refresh works for all transfer types"""
     with patch("zstash.globus.transfer_client") as mock_client, patch(
         "zstash.globus.local_endpoint", "local-uuid"
     ), patch("zstash.globus.remote_endpoint", "remote-uuid"), patch(
+        "zstash.globus.task_id", None
+    ), patch(
+        "zstash.globus.transfer_data", None
+    ), patch(
         "zstash.globus.archive_directory_listing", [{"name": "file.tar"}]
     ):
 
         mock_client.endpoint_autoactivate = Mock()
         mock_client.operation_ls = Mock(return_value=[{"name": "file.tar"}])
-        mock_client.submit_transfer = Mock(return_value={"task_id": "test-123"})
+        # Need to return a complete task dict to avoid KeyError
+        mock_client.submit_transfer = Mock(
+            return_value={
+                "task_id": "test-123",
+                "source_endpoint_id": "src-uuid",
+                "destination_endpoint_id": "dst-uuid",
+                "label": "test transfer",
+            }
+        )
         mock_client.task_wait = Mock(return_value=True)
-        mock_client.get_task = Mock(return_value={"status": "SUCCEEDED"})
+        mock_client.get_task = Mock(
+            return_value={
+                "status": "SUCCEEDED",
+                "source_endpoint_id": "src-uuid",
+                "destination_endpoint_id": "dst-uuid",
+                "label": "test transfer",
+            }
+        )
 
         globus_transfer("remote-ep", "/path", "file.tar", transfer_type, non_blocking)
 
         # Verify refresh was called
         assert mock_client.endpoint_autoactivate.called
 
 
-# Fixture for Common Setup ####################################################
-
-
 @pytest.fixture
 def mock_globus_client():
     """Fixture to set up a mock Globus client"""
     with patch("zstash.globus.transfer_client") as mock_client, patch(
         "zstash.globus.local_endpoint", "local-uuid"
-    ), patch("zstash.globus.remote_endpoint", "remote-uuid"):
+    ), patch("zstash.globus.remote_endpoint", "remote-uuid"), patch(
+        "zstash.globus.task_id", None
+    ), patch(
+        "zstash.globus.transfer_data", None
+    ):
 
         mock_client.endpoint_autoactivate = Mock()
         mock_client.operation_ls = Mock(return_value=[])
-        mock_client.submit_transfer = Mock(return_value={"task_id": "test-123"})
+        mock_client.submit_transfer = Mock(
+            return_value={
+                "task_id": "test-123",
+                "source_endpoint_id": "src-uuid",
+                "destination_endpoint_id": "dst-uuid",
+                "label": "test transfer",
+            }
+        )
         mock_client.task_wait = Mock(return_value=True)
-        mock_client.get_task = Mock(return_value={"status": "SUCCEEDED"})
+        mock_client.get_task = Mock(
+            return_value={
+                "status": "SUCCEEDED",
+                "source_endpoint_id": "src-uuid",
+                "destination_endpoint_id": "dst-uuid",
+                "label": "test transfer",
+            }
+        )
 
         yield mock_client
 
 
+# Demonstrates reusable fixture pattern
 def test_with_fixture(mock_globus_client):
     """Test using the fixture"""
     globus_transfer("remote-ep", "/path", "file.tar", "put", False)