update

Author: Tom Softreck

src/dialogchain/cli.py

@@ -55,13 +55,28 @@ def run(config, env_file, route, dry_run, verbose):
     # Run routes
     try:
         if route:
-            asyncio.run(engine.run_route(route))
+            # Find the specific route by name
+            route_config = next((r for r in config_data.get('routes', []) if r.get('name') == route), None)
+            if not route_config:
+                click.echo(f"❌ Route '{route}' not found in configuration", err=True)
+                return
+                
+            # Create source and destination
+            source = engine.create_source(route_config['from'])
+            destination = engine.create_destination(route_config['to'])
+            
+            # Run the specific route
+            asyncio.run(engine.run_route(route_config, source, destination))
         else:
+            # Run all routes
             asyncio.run(engine.run_all_routes())
     except KeyboardInterrupt:
         click.echo("\n🛑 Shutting down...")
     except Exception as e:
         click.echo(f"❌ Error: {e}", err=True)
+        if verbose:
+            import traceback
+            traceback.print_exc()
 
 
 def update_env_file(env_path, required_vars):

src/dialogchain/engine.py

@@ -16,6 +16,9 @@
     Processor, create_processor, FilterProcessor, TransformProcessor,
     AggregateProcessor, DebugProcessor, ExternalProcessor
 )
+
+# For backward compatibility
+ProcessorType = type(Processor)
 from .connectors import (
     Source, Destination, RTSPSource, FileSource, 
     HTTPDestination, FileDestination, IMAPSource, TimerSource, GRPCSource,
@@ -159,9 +162,35 @@ async def run_all_routes(self):
         """Run all routes concurrently"""
         tasks = []
         for route_config in self.routes:
-            task = asyncio.create_task(self.run_route_config(route_config))
-            tasks.append(task)
+            try:
+                # Create source and destination for this route
+                from_uri = self.resolve_variables(route_config.get('from', ''))
+                to_uri = self.resolve_variables(route_config.get('to', ''))
+                
+                if not from_uri or not to_uri:
+                    self.log(f"Skipping route '{route_config.get('name', 'unnamed')}': missing 'from' or 'to' URI")
+                    continue
+                    
+                source = self.create_source(from_uri)
+                destination = self.create_destination(to_uri)
+                
+                # Create and start task for this route
+                task = asyncio.create_task(
+                    self.run_route(route_config, source, destination)
+                )
+                tasks.append(task)
+                
+            except Exception as e:
+                self.log(f"Error setting up route '{route_config.get('name', 'unnamed')}': {e}")
+                if self.verbose:
+                    import traceback
+                    traceback.print_exc()
+                continue
 
+        if not tasks:
+            self.log("No valid routes to run")
+            return
+            
         self.log(f"Starting {len(tasks)} routes...")
         await asyncio.gather(*tasks, return_exceptions=True)
 

tests/unit/test_http_destination.py

@@ -4,6 +4,16 @@
 from unittest.mock import AsyncMock, patch
 from dialogchain.connectors import HTTPDestination
 
+class AsyncMockContext:
+    def __init__(self, return_value):
+        self.return_value = return_value
+    
+    async def __aenter__(self):
+        return self.return_value
+    
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
 class TestHTTPDestination:
     """Test HTTP destination connector."""
 
@@ -15,23 +25,14 @@ def http_dest(self):
     @pytest.mark.asyncio
     async def test_send_http_request(self, http_dest, capsys):
         """Test sending an HTTP request."""
-        # Create a mock response for successful request
+        # Create a mock response
         mock_response = AsyncMock()
         mock_response.status = 200
         mock_response.text = AsyncMock(return_value="OK")
 
-        # Create a mock post method
-        async def mock_post(*args, **kwargs):
-            return mock_response
-            
-        # Create a mock session
-        mock_session = AsyncMock()
-        mock_session.post.side_effect = mock_post
-        mock_session.__aenter__.return_value = mock_session
-        mock_session.__aexit__.return_value = None
-        
-        # Patch the ClientSession
-        with patch('aiohttp.ClientSession', return_value=mock_session):
+        # Patch the ClientSession.post method directly
+        with patch('aiohttp.ClientSession.post', 
+                  return_value=AsyncMockContext(mock_response)) as mock_post:
             # Test with dict message
             await http_dest.send({"key": "value"})
 
@@ -40,7 +41,7 @@ async def mock_post(*args, **kwargs):
             assert "🌐 HTTP sent to http://example.com/webhook" in captured.out
 
             # Verify the post was called correctly
-            mock_session.post.assert_called_once_with(
+            mock_post.assert_called_once_with(
                 'http://example.com/webhook',
                 json={"key": "value"}
             )
@@ -53,18 +54,9 @@ async def test_send_string_message(self, http_dest, capsys):
         mock_response.status = 200
         mock_response.text = AsyncMock(return_value="OK")
 
-        # Create a mock post method
-        async def mock_post(*args, **kwargs):
-            return mock_response
-            
-        # Create a mock session
-        mock_session = AsyncMock()
-        mock_session.post.side_effect = mock_post
-        mock_session.__aenter__.return_value = mock_session
-        mock_session.__aexit__.return_value = None
-        
-        # Patch the ClientSession
-        with patch('aiohttp.ClientSession', return_value=mock_session):
+        # Patch the ClientSession.post method directly
+        with patch('aiohttp.ClientSession.post',
+                  return_value=AsyncMockContext(mock_response)) as mock_post:
             # Test with string message
             await http_dest.send("test message")
 
@@ -73,7 +65,7 @@ async def mock_post(*args, **kwargs):
             assert "🌐 HTTP sent to http://example.com/webhook" in captured.out
 
             # Verify the post was called correctly
-            mock_session.post.assert_called_once_with(
+            mock_post.assert_called_once_with(
                 'http://example.com/webhook',
                 json={"data": "test message"}
             )
@@ -86,18 +78,9 @@ async def test_http_error(self, http_dest, capsys):
         mock_response.status = 400
         mock_response.text = AsyncMock(return_value="Bad Request")
 
-        # Create a mock post method
-        async def mock_post(*args, **kwargs):
-            return mock_response
-            
-        # Create a mock session
-        mock_session = AsyncMock()
-        mock_session.post.side_effect = mock_post
-        mock_session.__aenter__.return_value = mock_session
-        mock_session.__aexit__.return_value = None
-        
-        # Patch the ClientSession
-        with patch('aiohttp.ClientSession', return_value=mock_session):
+        # Patch the ClientSession.post method directly
+        with patch('aiohttp.ClientSession.post',
+                  return_value=AsyncMockContext(mock_response)) as mock_post:
             # Test with dict message
             await http_dest.send({"key": "value"})