tests: add 'distributed' marker to pytest configuration and update workflows for distributed dependencies

Author: UnicoLab

.github/workflows/UTESTS.yml

@@ -57,7 +57,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install poetry
-          poetry install --no-interaction
+          poetry install --no-interaction --with dev,distributed,visualization
       - name: Run smoke tests
         run: |
           poetry run pytest -m "micro" --maxfail=1 --tb=no -q --timeout=30
@@ -100,7 +100,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install poetry
-          poetry install --no-interaction
+          poetry install --no-interaction --with dev,distributed,visualization
 
       - name: Run tests - ${{ matrix.test-group }}
         run: |
@@ -109,7 +109,7 @@ jobs:
               poetry run pytest -c pytest-fast.ini -m "unit" --maxfail=10 --timeout=120
               ;;
             "integration")
-              poetry run pytest -m "integration" --maxfail=3 --timeout=300
+              poetry run pytest -m "integration or distributed" --maxfail=3 --timeout=300
               ;;
             "layers")
               poetry run pytest -c pytest-fast.ini -m "layers" --maxfail=5 --timeout=120
@@ -149,7 +149,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install poetry
-          poetry install --no-interaction
+          poetry install --no-interaction --with dev,distributed,visualization
 
       - name: Run tests with coverage
         run: |
@@ -186,7 +186,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           pip install poetry
-          poetry install --no-interaction
+          poetry install --no-interaction --with dev,distributed,visualization
 
       - name: Run benchmarks
         run: |

examples/TODO.md

@@ -0,0 +1,3 @@
+- documentation needs to be updated -> adding decorators for nodes registers them already in the nodes list, not need to do it manually of call automatic adding neither
+
+- rewiev all the examples in the docs to make sure they run correctly -> using notebooks maybe

pytest-fast.ini

@@ -23,3 +23,4 @@ markers =
     processor: marks tests for processor functionality
     time_series: marks tests for time series functionality
     performance: marks tests as performance benchmarks
+    distributed: marks tests that require distributed dependencies (ray, dask)

pytest.ini

@@ -24,5 +24,6 @@ markers =
     layers: marks tests for layer functionality
     processor: marks tests for processor functionality
     time_series: marks tests for time series functionality
+    distributed: marks tests that require distributed dependencies (ray, dask)
 asyncio_mode = auto
 

tests/test_distributed_executors.py

@@ -16,11 +16,18 @@
 from graphflow.core.executors import ExecutionResult
 
 
+@pytest.mark.distributed
 class TestRayExecutor:
     """Test Ray executor functionality."""
 
     def test_ray_executor_import(self):
         """Test that Ray executor can be imported."""
+        # Check if Ray is available before importing
+        try:
+            import ray
+        except ImportError:
+            pytest.skip("Ray not available")
+        
         try:
             from graphflow.executors.ray_executor import RayExecutor, create_ray_executor
             assert RayExecutor is not None
@@ -30,6 +37,12 @@ def test_ray_executor_import(self):
 
     def test_ray_executor_initialization(self):
         """Test Ray executor initialization."""
+        # Check if Ray is available before importing
+        try:
+            import ray
+        except ImportError:
+            pytest.skip("Ray not available")
+        
         try:
             from graphflow.executors.ray_executor import RayExecutor
 
@@ -72,6 +85,7 @@ def test_ray_cluster_executor_not_initialized(self):
             pytest.skip("Ray not available")
 
 
+@pytest.mark.distributed
 class TestDaskExecutor:
     """Test Dask executor functionality."""
 
@@ -123,6 +137,7 @@ def test_dask_distributed_executor(self):
             pytest.skip("Dask not available")
 
 
+@pytest.mark.distributed
 class TestExecutorRegistry:
     """Test executor registry with distributed executors."""
 
@@ -180,6 +195,7 @@ def test_auto_select_executor_with_distributed(self):
                 assert executor == "processes"
 
 
+@pytest.mark.distributed
 class TestDistributedExecution:
     """Test distributed execution scenarios."""
 
@@ -202,6 +218,12 @@ def create_test_context(self):
 
     def test_ray_execution_simulation(self):
         """Test Ray execution with mocked Ray."""
+        # Check if Ray is available before importing
+        try:
+            import ray
+        except ImportError:
+            pytest.skip("Ray not available")
+        
         try:
             from graphflow.executors.ray_executor import RayExecutor
 
@@ -210,29 +232,35 @@ def test_ray_execution_simulation(self):
             inputs = {"df": pd.DataFrame({"value": [1, 2, 3]})}
 
             # Mock Ray components comprehensively
-            with patch('ray.is_initialized', return_value=False):
-                with patch('ray.init') as mock_init:
-                    with patch('ray.get') as mock_get:
-                        with patch('ray.remote') as mock_remote:
-                            # Mock the remote function to return a callable that returns our expected result
-                            mock_remote_func = MagicMock()
-                            mock_remote_func.remote.return_value = "mock_ray_object"
-                            mock_remote.return_value = mock_remote_func
-                            
-                            # Mock ray.get to return our expected result
-                            mock_get.return_value = pd.DataFrame({"value": [12, 14, 16]})
+            with patch('ray.init') as mock_init:
+                with patch('ray.get') as mock_get:
+                    with patch('ray.remote') as mock_remote:
+                        # Mock the remote function to return a callable that returns our expected result
+                        mock_remote_func = MagicMock()
+                        mock_remote_func.remote.return_value = "mock_ray_object"
+                        mock_remote.return_value = mock_remote_func
+                        
+                        # Mock ray.get to return our expected result
+                        mock_get.return_value = pd.DataFrame({"value": [12, 14, 16]})
+                        
+                        # Mock ray.is_initialized to return False initially, then True after init
+                        init_called = False
+                        def mock_is_initialized():
+                            return init_called
+                        
+                        def mock_init_side_effect(*args, **kwargs):
+                            nonlocal init_called
+                            init_called = True
+                        
+                        mock_init.side_effect = mock_init_side_effect
+                        
+                        with patch('ray.is_initialized', side_effect=mock_is_initialized):
+                            executor = RayExecutor()
+                            result = executor.execute_node(node_spec, context, inputs)
 
-                            # Mock ray.is_initialized to return True after init is called
-                            def mock_is_initialized():
-                                return mock_init.called
-                            
-                            with patch('ray.is_initialized', side_effect=mock_is_initialized):
-                                executor = RayExecutor()
-                                result = executor.execute_node(node_spec, context, inputs)
-                                
-                                assert result.success
-                                assert result.node_name == "test_func"
-                                assert result.cache_hit is False
+                            assert result.success
+                            assert result.node_name == "test_func"
+                            assert result.cache_hit is False
         except ImportError:
             pytest.skip("Ray not available")
 
@@ -258,6 +286,7 @@ def test_dask_execution_simulation(self):
             pytest.skip("Dask not available")
 
 
+@pytest.mark.distributed
 class TestExecutorFactory:
     """Test executor factory functions."""
 
@@ -291,11 +320,18 @@ def test_create_dask_executor(self):
             pytest.skip("Dask not available")
 
 
+@pytest.mark.distributed
 class TestExecutorErrorHandling:
     """Test error handling in distributed executors."""
 
     def test_ray_execution_error(self):
         """Test Ray executor error handling."""
+        # Check if Ray is available before importing
+        try:
+            import ray
+        except ImportError:
+            pytest.skip("Ray not available")
+        
         try:
             from graphflow.executors.ray_executor import RayExecutor
 
@@ -311,25 +347,31 @@ def failing_func(df, param):
             context = Context(data={"param": 1})
             inputs = {"df": pd.DataFrame({"value": [1, 2, 3]})}
 
-            with patch('ray.is_initialized', return_value=False):
-                with patch('ray.init') as mock_init:
-                    with patch('ray.get', side_effect=ValueError("Test error")):
-                        with patch('ray.remote') as mock_remote:
-                            # Mock the remote function
-                            mock_remote_func = MagicMock()
-                            mock_remote_func.remote.return_value = "mock_ray_object"
-                            mock_remote.return_value = mock_remote_func
-                            
-                            # Mock ray.is_initialized to return True after init is called
-                            def mock_is_initialized():
-                                return mock_init.called
+            with patch('ray.init') as mock_init:
+                with patch('ray.get', side_effect=ValueError("Test error")):
+                    with patch('ray.remote') as mock_remote:
+                        # Mock the remote function
+                        mock_remote_func = MagicMock()
+                        mock_remote_func.remote.return_value = "mock_ray_object"
+                        mock_remote.return_value = mock_remote_func
+                        
+                        # Mock ray.is_initialized to return False initially, then True after init
+                        init_called = False
+                        def mock_is_initialized():
+                            return init_called
+                        
+                        def mock_init_side_effect(*args, **kwargs):
+                            nonlocal init_called
+                            init_called = True
+                        
+                        mock_init.side_effect = mock_init_side_effect
+                        
+                        with patch('ray.is_initialized', side_effect=mock_is_initialized):
+                            executor = RayExecutor()
+                            result = executor.execute_node(node_spec, context, inputs)
 
-                            with patch('ray.is_initialized', side_effect=mock_is_initialized):
-                                executor = RayExecutor()
-                                result = executor.execute_node(node_spec, context, inputs)
-                                
-                                assert not result.success
-                                assert "Test error" in result.error
+                            assert not result.success
+                            assert "Test error" in result.error
         except ImportError:
             pytest.skip("Ray not available")