async def test_get_valid_credentials_with_refresh_failure(
      self, mock_request_class, manager, mock_tool_context
  ):
    """Test OAuth flow trigger when credential refresh fails.

    When refresh tokens expire or become invalid, the system should
    gracefully fall back to requesting a new OAuth flow.
    """
    # Create expired credentials that fail to refresh
    mock_creds = Mock(spec=Credentials)
    mock_creds.valid = False
    mock_creds.expired = True
    mock_creds.refresh_token = "expired_refresh_token"
    mock_creds.refresh = Mock(side_effect=RefreshError("Refresh failed"))
    manager.credentials_config.credentials = mock_creds

    result = await manager.get_valid_credentials(mock_tool_context)

    # Should trigger OAuth flow and return None (flow in progress)
    assert result is None
    mock_tool_context.request_credential.assert_called_once()
Test OAuth flow trigger when credential refresh fails.

    When refresh tokens expire or become invalid, the system should
    gracefully fall back to requesting a new OAuth flow.
    
tests/unittests/tools/bigquery/test_bigquery_credentials_manager.py