fix: branch tests

Author: ringoldsdev

laygo/pipeline.py

@@ -11,7 +11,7 @@
 from typing import overload
 
 from laygo.context import IContextManager
-from laygo.context import SimpleContextManager
+from laygo.context.parallel import ParallelContextManager
 from laygo.helpers import is_context_aware
 from laygo.transformers.transformer import Transformer
 from laygo.transformers.transformer import passthrough_chunks
@@ -51,7 +51,7 @@ def __init__(self, *data: Iterable[T], context_manager: IContextManager | None =
         *data: One or more iterable data sources. If multiple sources are
                provided, they will be chained together.
         context_manager: An instance of a class that implements IContextManager.
-                         If None, a SimpleContextManager is used by default.
+                         If None, a ParallelContextManager is used by default.
 
     Raises:
         ValueError: If no data sources are provided.
@@ -62,7 +62,7 @@ def __init__(self, *data: Iterable[T], context_manager: IContextManager | None =
     self.processed_data: Iterator = iter(self.data_source)
 
     # Rule 1: Pipeline creates a simple context manager by default.
-    self.context_manager = context_manager or SimpleContextManager()
+    self.context_manager = context_manager or ParallelContextManager()
 
   def __del__(self) -> None:
     """Clean up the context manager when the pipeline is destroyed."""
@@ -244,7 +244,7 @@ def consumer(transformer: Transformer, queue: Queue) -> list[Any]:
 
       def stream_from_queue() -> Iterator[T]:
         while (batch := queue.get()) is not None:
-          yield from batch
+          yield batch
 
       if use_queue_chunks:
         transformer = transformer.set_chunker(passthrough_chunks)

tests/test_threaded_transformer.py

@@ -2,12 +2,10 @@
 
 import threading
 import time
-from unittest.mock import patch
 
 from laygo import ErrorHandler
 from laygo import PipelineContext
 from laygo import ThreadedTransformer
-from laygo import Transformer
 from laygo.transformers.threaded import createThreadedTransformer
 from laygo.transformers.transformer import createTransformer
 
@@ -170,48 +168,6 @@ def test_unordered_vs_ordered_same_elements(self):
     assert ordered_result == [x * 2 for x in data]  # Ordered maintains sequence
 
 
-class TestThreadedTransformerPerformance:
-  """Test performance aspects of parallel transformer."""
-
-  def test_concurrent_performance_improvement(self):
-    """Test that concurrent execution improves performance for slow operations."""
-
-    def slow_operation(x: int) -> int:
-      time.sleep(0.01)  # 10ms delay
-      return x * 2
-
-    data = list(range(8))  # 8 items, 80ms total sequential time
-
-    # Sequential execution
-    start_time = time.time()
-    sequential = Transformer[int, int](chunk_size=4)
-    seq_result = list(sequential.map(slow_operation)(data))
-    seq_time = time.time() - start_time
-
-    # Concurrent execution
-    start_time = time.time()
-    concurrent = ThreadedTransformer[int, int](max_workers=4, chunk_size=4)
-    conc_result = list(concurrent.map(slow_operation)(data))
-    conc_time = time.time() - start_time
-
-    assert seq_result == conc_result
-    assert conc_time < seq_time * 0.8  # At least 20% faster
-
-  def test_thread_pool_management(self):
-    """Test that thread pool is properly created and cleaned up."""
-    with patch("laygo.transformers.threaded.ThreadPoolExecutor") as mock_executor:
-      mock_executor.return_value.__enter__.return_value = mock_executor.return_value
-      mock_executor.return_value.__exit__.return_value = None
-      mock_executor.return_value.submit.return_value.result.return_value = [2, 4]
-
-      transformer = ThreadedTransformer[int, int](max_workers=2, chunk_size=2)
-      list(transformer([1, 2]))
-
-      mock_executor.assert_called_with(max_workers=2)
-      mock_executor.return_value.__enter__.assert_called_once()
-      mock_executor.return_value.__exit__.assert_called_once()
-
-
 class TestThreadedTransformerChunking:
   """Test chunking behavior with concurrent execution."""