Rev gateway tests

tests/unit/test_gateway/test_config_receiver.py

@@ -0,0 +1,356 @@
+"""
+Tests for Gateway Config Receiver
+"""
+
+import pytest
+import asyncio
+import json
+from unittest.mock import Mock, patch, AsyncMock, MagicMock
+import uuid
+
+from trustgraph.gateway.config.receiver import ConfigReceiver
+
+
+class TestConfigReceiver:
+    """Test cases for ConfigReceiver class"""
+
+    def test_config_receiver_initialization(self):
+        """Test ConfigReceiver initialization"""
+        mock_pulsar_client = Mock()
+        
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        assert config_receiver.pulsar_client == mock_pulsar_client
+        assert config_receiver.flow_handlers == []
+        assert config_receiver.flows == {}
+
+    def test_add_handler(self):
+        """Test adding flow handlers"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        handler1 = Mock()
+        handler2 = Mock()
+        
+        config_receiver.add_handler(handler1)
+        config_receiver.add_handler(handler2)
+        
+        assert len(config_receiver.flow_handlers) == 2
+        assert handler1 in config_receiver.flow_handlers
+        assert handler2 in config_receiver.flow_handlers
+
+    @pytest.mark.asyncio
+    async def test_on_config_with_new_flows(self):
+        """Test on_config method with new flows"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Mock the start_flow method
+        config_receiver.start_flow = AsyncMock()
+        
+        # Create mock message with flows
+        mock_msg = Mock()
+        mock_msg.value.return_value = Mock(
+            version="1.0",
+            config={
+                "flows": {
+                    "flow1": '{"name": "test_flow_1", "steps": []}',
+                    "flow2": '{"name": "test_flow_2", "steps": []}'
+                }
+            }
+        )
+        
+        await config_receiver.on_config(mock_msg, None, None)
+        
+        # Verify flows were added
+        assert "flow1" in config_receiver.flows
+        assert "flow2" in config_receiver.flows
+        assert config_receiver.flows["flow1"] == {"name": "test_flow_1", "steps": []}
+        assert config_receiver.flows["flow2"] == {"name": "test_flow_2", "steps": []}
+        
+        # Verify start_flow was called for each new flow
+        assert config_receiver.start_flow.call_count == 2
+        config_receiver.start_flow.assert_any_call("flow1", {"name": "test_flow_1", "steps": []})
+        config_receiver.start_flow.assert_any_call("flow2", {"name": "test_flow_2", "steps": []})
+
+    @pytest.mark.asyncio
+    async def test_on_config_with_removed_flows(self):
+        """Test on_config method with removed flows"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Pre-populate with existing flows
+        config_receiver.flows = {
+            "flow1": {"name": "test_flow_1", "steps": []},
+            "flow2": {"name": "test_flow_2", "steps": []}
+        }
+        
+        # Mock the stop_flow method
+        config_receiver.stop_flow = AsyncMock()
+        
+        # Create mock message with only flow1 (flow2 removed)
+        mock_msg = Mock()
+        mock_msg.value.return_value = Mock(
+            version="1.0",
+            config={
+                "flows": {
+                    "flow1": '{"name": "test_flow_1", "steps": []}'
+                }
+            }
+        )
+        
+        await config_receiver.on_config(mock_msg, None, None)
+        
+        # Verify flow2 was removed
+        assert "flow1" in config_receiver.flows
+        assert "flow2" not in config_receiver.flows
+        
+        # Verify stop_flow was called for removed flow
+        config_receiver.stop_flow.assert_called_once_with("flow2", {"name": "test_flow_2", "steps": []})
+
+    @pytest.mark.asyncio
+    async def test_on_config_with_no_flows(self):
+        """Test on_config method with no flows in config"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Mock the start_flow and stop_flow methods
+        config_receiver.start_flow = AsyncMock()
+        config_receiver.stop_flow = AsyncMock()
+        
+        # Create mock message without flows
+        mock_msg = Mock()
+        mock_msg.value.return_value = Mock(
+            version="1.0",
+            config={}
+        )
+        
+        await config_receiver.on_config(mock_msg, None, None)
+        
+        # Verify no flows were added
+        assert config_receiver.flows == {}
+        
+        # Verify no flow operations were called
+        config_receiver.start_flow.assert_not_called()
+        config_receiver.stop_flow.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_on_config_exception_handling(self):
+        """Test on_config method handles exceptions gracefully"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Create mock message that will cause an exception
+        mock_msg = Mock()
+        mock_msg.value.side_effect = Exception("Test exception")
+        
+        # This should not raise an exception
+        await config_receiver.on_config(mock_msg, None, None)
+        
+        # Verify flows remain empty
+        assert config_receiver.flows == {}
+
+    @pytest.mark.asyncio
+    async def test_start_flow_with_handlers(self):
+        """Test start_flow method with multiple handlers"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Add mock handlers
+        handler1 = Mock()
+        handler1.start_flow = AsyncMock()
+        handler2 = Mock()
+        handler2.start_flow = AsyncMock()
+        
+        config_receiver.add_handler(handler1)
+        config_receiver.add_handler(handler2)
+        
+        flow_data = {"name": "test_flow", "steps": []}
+        
+        await config_receiver.start_flow("flow1", flow_data)
+        
+        # Verify all handlers were called
+        handler1.start_flow.assert_called_once_with("flow1", flow_data)
+        handler2.start_flow.assert_called_once_with("flow1", flow_data)
+
+    @pytest.mark.asyncio
+    async def test_start_flow_with_handler_exception(self):
+        """Test start_flow method handles handler exceptions"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Add mock handler that raises exception
+        handler = Mock()
+        handler.start_flow = AsyncMock(side_effect=Exception("Handler error"))
+        
+        config_receiver.add_handler(handler)
+        
+        flow_data = {"name": "test_flow", "steps": []}
+        
+        # This should not raise an exception
+        await config_receiver.start_flow("flow1", flow_data)
+        
+        # Verify handler was called
+        handler.start_flow.assert_called_once_with("flow1", flow_data)
+
+    @pytest.mark.asyncio
+    async def test_stop_flow_with_handlers(self):
+        """Test stop_flow method with multiple handlers"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Add mock handlers
+        handler1 = Mock()
+        handler1.stop_flow = AsyncMock()
+        handler2 = Mock()
+        handler2.stop_flow = AsyncMock()
+        
+        config_receiver.add_handler(handler1)
+        config_receiver.add_handler(handler2)
+        
+        flow_data = {"name": "test_flow", "steps": []}
+        
+        await config_receiver.stop_flow("flow1", flow_data)
+        
+        # Verify all handlers were called
+        handler1.stop_flow.assert_called_once_with("flow1", flow_data)
+        handler2.stop_flow.assert_called_once_with("flow1", flow_data)
+
+    @pytest.mark.asyncio
+    async def test_stop_flow_with_handler_exception(self):
+        """Test stop_flow method handles handler exceptions"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Add mock handler that raises exception
+        handler = Mock()
+        handler.stop_flow = AsyncMock(side_effect=Exception("Handler error"))
+        
+        config_receiver.add_handler(handler)
+        
+        flow_data = {"name": "test_flow", "steps": []}
+        
+        # This should not raise an exception
+        await config_receiver.stop_flow("flow1", flow_data)
+        
+        # Verify handler was called
+        handler.stop_flow.assert_called_once_with("flow1", flow_data)
+
+    @pytest.mark.asyncio
+    async def test_config_loader_creates_consumer(self):
+        """Test config_loader method creates Pulsar consumer"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Mock Consumer class
+        with patch('trustgraph.gateway.config.receiver.Consumer') as mock_consumer_class, \
+             patch('uuid.uuid4') as mock_uuid:
+            
+            mock_uuid.return_value = "test-uuid"
+            mock_consumer = Mock()
+            mock_consumer.start = AsyncMock()
+            mock_consumer_class.return_value = mock_consumer
+            
+            # Create a task that will complete quickly
+            async def quick_task():
+                await config_receiver.config_loader()
+            
+            # Run the task with a timeout to prevent hanging
+            try:
+                await asyncio.wait_for(quick_task(), timeout=0.1)
+            except asyncio.TimeoutError:
+                # This is expected since the method runs indefinitely
+                pass
+            
+            # Verify Consumer was created with correct parameters
+            mock_consumer_class.assert_called_once()
+            call_args = mock_consumer_class.call_args
+            
+            assert call_args[1]['client'] == mock_pulsar_client
+            assert call_args[1]['subscriber'] == "gateway-test-uuid"
+            assert call_args[1]['handler'] == config_receiver.on_config
+            assert call_args[1]['start_of_messages'] is True
+
+    @pytest.mark.asyncio
+    async def test_start_creates_config_loader_task(self):
+        """Test start method creates config loader task"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        with patch('asyncio.create_task') as mock_create_task:
+            await config_receiver.start()
+            
+            # Verify task was created
+            mock_create_task.assert_called_once()
+            
+            # Verify the task is for config_loader
+            task_coro = mock_create_task.call_args[0][0]
+            assert hasattr(task_coro, 'cr_code')
+            assert task_coro.cr_code.co_name == 'config_loader'
+
+    @pytest.mark.asyncio
+    async def test_on_config_mixed_flow_operations(self):
+        """Test on_config with mixed add/remove operations"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Pre-populate with existing flows
+        config_receiver.flows = {
+            "flow1": {"name": "test_flow_1", "steps": []},
+            "flow2": {"name": "test_flow_2", "steps": []}
+        }
+        
+        # Mock the flow methods
+        config_receiver.start_flow = AsyncMock()
+        config_receiver.stop_flow = AsyncMock()
+        
+        # Create mock message with flow1 removed and flow3 added
+        mock_msg = Mock()
+        mock_msg.value.return_value = Mock(
+            version="1.0",
+            config={
+                "flows": {
+                    "flow2": '{"name": "test_flow_2", "steps": []}',
+                    "flow3": '{"name": "test_flow_3", "steps": []}'
+                }
+            }
+        )
+        
+        await config_receiver.on_config(mock_msg, None, None)
+        
+        # Verify final state
+        assert "flow1" not in config_receiver.flows
+        assert "flow2" in config_receiver.flows
+        assert "flow3" in config_receiver.flows
+        
+        # Verify operations
+        config_receiver.start_flow.assert_called_once_with("flow3", {"name": "test_flow_3", "steps": []})
+        config_receiver.stop_flow.assert_called_once_with("flow1", {"name": "test_flow_1", "steps": []})
+
+    @pytest.mark.asyncio
+    async def test_on_config_invalid_json_flow_data(self):
+        """Test on_config handles invalid JSON in flow data"""
+        mock_pulsar_client = Mock()
+        config_receiver = ConfigReceiver(mock_pulsar_client)
+        
+        # Mock the start_flow method
+        config_receiver.start_flow = AsyncMock()
+        
+        # Create mock message with invalid JSON
+        mock_msg = Mock()
+        mock_msg.value.return_value = Mock(
+            version="1.0",
+            config={
+                "flows": {
+                    "flow1": '{"invalid": json}',  # Invalid JSON
+                    "flow2": '{"name": "valid_flow", "steps": []}'  # Valid JSON
+                }
+            }
+        )
+        
+        # This should handle the exception gracefully
+        await config_receiver.on_config(mock_msg, None, None)
+        
+        # The entire operation should fail due to JSON parsing error
+        # So no flows should be added
+        assert config_receiver.flows == {}

tests/unit/test_gateway/test_dispatch_manager.py

@@ -0,0 +1,550 @@
+"""
+Tests for Gateway Dispatcher Manager
+"""
+
+import pytest
+import asyncio
+from unittest.mock import Mock, patch, AsyncMock, MagicMock
+import uuid
+
+from trustgraph.gateway.dispatch.manager import DispatcherManager, DispatcherWrapper
+
+
+class TestDispatcherWrapper:
+    """Test cases for DispatcherWrapper class"""
+
+    def test_dispatcher_wrapper_initialization(self):
+        """Test DispatcherWrapper initialization"""
+        mock_handler = Mock()
+        wrapper = DispatcherWrapper(mock_handler)
+        
+        assert wrapper.handler == mock_handler
+
+    @pytest.mark.asyncio
+    async def test_dispatcher_wrapper_process(self):
+        """Test DispatcherWrapper process method"""
+        mock_handler = AsyncMock()
+        wrapper = DispatcherWrapper(mock_handler)
+        
+        result = await wrapper.process("arg1", "arg2", kwarg1="value1")
+        
+        mock_handler.assert_called_once_with("arg1", "arg2", kwarg1="value1")
+        assert result == mock_handler.return_value
+
+
+class TestDispatcherManager:
+    """Test cases for DispatcherManager class"""
+
+    def test_dispatcher_manager_initialization(self):
+        """Test DispatcherManager initialization"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        assert manager.pulsar_client == mock_pulsar_client
+        assert manager.config_receiver == mock_config_receiver
+        assert manager.prefix == "api-gateway"  # default prefix
+        assert manager.flows == {}
+        assert manager.dispatchers == {}
+        
+        # Verify manager was added as handler to config receiver
+        mock_config_receiver.add_handler.assert_called_once_with(manager)
+
+    def test_dispatcher_manager_initialization_with_custom_prefix(self):
+        """Test DispatcherManager initialization with custom prefix"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver, prefix="custom-prefix")
+        
+        assert manager.prefix == "custom-prefix"
+
+    @pytest.mark.asyncio
+    async def test_start_flow(self):
+        """Test start_flow method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        flow_data = {"name": "test_flow", "steps": []}
+        
+        await manager.start_flow("flow1", flow_data)
+        
+        assert "flow1" in manager.flows
+        assert manager.flows["flow1"] == flow_data
+
+    @pytest.mark.asyncio
+    async def test_stop_flow(self):
+        """Test stop_flow method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Pre-populate with a flow
+        flow_data = {"name": "test_flow", "steps": []}
+        manager.flows["flow1"] = flow_data
+        
+        await manager.stop_flow("flow1", flow_data)
+        
+        assert "flow1" not in manager.flows
+
+    def test_dispatch_global_service_returns_wrapper(self):
+        """Test dispatch_global_service returns DispatcherWrapper"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        wrapper = manager.dispatch_global_service()
+        
+        assert isinstance(wrapper, DispatcherWrapper)
+        assert wrapper.handler == manager.process_global_service
+
+    def test_dispatch_core_export_returns_wrapper(self):
+        """Test dispatch_core_export returns DispatcherWrapper"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        wrapper = manager.dispatch_core_export()
+        
+        assert isinstance(wrapper, DispatcherWrapper)
+        assert wrapper.handler == manager.process_core_export
+
+    def test_dispatch_core_import_returns_wrapper(self):
+        """Test dispatch_core_import returns DispatcherWrapper"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        wrapper = manager.dispatch_core_import()
+        
+        assert isinstance(wrapper, DispatcherWrapper)
+        assert wrapper.handler == manager.process_core_import
+
+    @pytest.mark.asyncio
+    async def test_process_core_import(self):
+        """Test process_core_import method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        with patch('trustgraph.gateway.dispatch.manager.CoreImport') as mock_core_import:
+            mock_importer = Mock()
+            mock_importer.process = AsyncMock(return_value="import_result")
+            mock_core_import.return_value = mock_importer
+            
+            result = await manager.process_core_import("data", "error", "ok", "request")
+            
+            mock_core_import.assert_called_once_with(mock_pulsar_client)
+            mock_importer.process.assert_called_once_with("data", "error", "ok", "request")
+            assert result == "import_result"
+
+    @pytest.mark.asyncio
+    async def test_process_core_export(self):
+        """Test process_core_export method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        with patch('trustgraph.gateway.dispatch.manager.CoreExport') as mock_core_export:
+            mock_exporter = Mock()
+            mock_exporter.process = AsyncMock(return_value="export_result")
+            mock_core_export.return_value = mock_exporter
+            
+            result = await manager.process_core_export("data", "error", "ok", "request")
+            
+            mock_core_export.assert_called_once_with(mock_pulsar_client)
+            mock_exporter.process.assert_called_once_with("data", "error", "ok", "request")
+            assert result == "export_result"
+
+    @pytest.mark.asyncio
+    async def test_process_global_service(self):
+        """Test process_global_service method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        manager.invoke_global_service = AsyncMock(return_value="global_result")
+        
+        params = {"kind": "test_kind"}
+        result = await manager.process_global_service("data", "responder", params)
+        
+        manager.invoke_global_service.assert_called_once_with("data", "responder", "test_kind")
+        assert result == "global_result"
+
+    @pytest.mark.asyncio
+    async def test_invoke_global_service_with_existing_dispatcher(self):
+        """Test invoke_global_service with existing dispatcher"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Pre-populate with existing dispatcher
+        mock_dispatcher = Mock()
+        mock_dispatcher.process = AsyncMock(return_value="cached_result")
+        manager.dispatchers[(None, "config")] = mock_dispatcher
+        
+        result = await manager.invoke_global_service("data", "responder", "config")
+        
+        mock_dispatcher.process.assert_called_once_with("data", "responder")
+        assert result == "cached_result"
+
+    @pytest.mark.asyncio
+    async def test_invoke_global_service_creates_new_dispatcher(self):
+        """Test invoke_global_service creates new dispatcher"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        with patch('trustgraph.gateway.dispatch.manager.global_dispatchers') as mock_dispatchers:
+            mock_dispatcher_class = Mock()
+            mock_dispatcher = Mock()
+            mock_dispatcher.start = AsyncMock()
+            mock_dispatcher.process = AsyncMock(return_value="new_result")
+            mock_dispatcher_class.return_value = mock_dispatcher
+            mock_dispatchers.__getitem__.return_value = mock_dispatcher_class
+            
+            result = await manager.invoke_global_service("data", "responder", "config")
+            
+            # Verify dispatcher was created with correct parameters
+            mock_dispatcher_class.assert_called_once_with(
+                pulsar_client=mock_pulsar_client,
+                timeout=120,
+                consumer="api-gateway-config-request",
+                subscriber="api-gateway-config-request"
+            )
+            mock_dispatcher.start.assert_called_once()
+            mock_dispatcher.process.assert_called_once_with("data", "responder")
+            
+            # Verify dispatcher was cached
+            assert manager.dispatchers[(None, "config")] == mock_dispatcher
+            assert result == "new_result"
+
+    def test_dispatch_flow_import_returns_method(self):
+        """Test dispatch_flow_import returns correct method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        result = manager.dispatch_flow_import()
+        
+        assert result == manager.process_flow_import
+
+    def test_dispatch_flow_export_returns_method(self):
+        """Test dispatch_flow_export returns correct method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        result = manager.dispatch_flow_export()
+        
+        assert result == manager.process_flow_export
+
+    def test_dispatch_socket_returns_method(self):
+        """Test dispatch_socket returns correct method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        result = manager.dispatch_socket()
+        
+        assert result == manager.process_socket
+
+    def test_dispatch_flow_service_returns_wrapper(self):
+        """Test dispatch_flow_service returns DispatcherWrapper"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        wrapper = manager.dispatch_flow_service()
+        
+        assert isinstance(wrapper, DispatcherWrapper)
+        assert wrapper.handler == manager.process_flow_service
+
+    @pytest.mark.asyncio
+    async def test_process_flow_import_with_valid_flow_and_kind(self):
+        """Test process_flow_import with valid flow and kind"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "triples-store": {"queue": "test_queue"}
+            }
+        }
+        
+        with patch('trustgraph.gateway.dispatch.manager.import_dispatchers') as mock_dispatchers, \
+             patch('uuid.uuid4') as mock_uuid:
+            mock_uuid.return_value = "test-uuid"
+            mock_dispatcher_class = Mock()
+            mock_dispatcher = Mock()
+            mock_dispatcher.start = AsyncMock()
+            mock_dispatcher_class.return_value = mock_dispatcher
+            mock_dispatchers.__getitem__.return_value = mock_dispatcher_class
+            mock_dispatchers.__contains__.return_value = True
+            
+            params = {"flow": "test_flow", "kind": "triples"}
+            result = await manager.process_flow_import("ws", "running", params)
+            
+            mock_dispatcher_class.assert_called_once_with(
+                pulsar_client=mock_pulsar_client,
+                ws="ws",
+                running="running",
+                queue={"queue": "test_queue"}
+            )
+            mock_dispatcher.start.assert_called_once()
+            assert result == mock_dispatcher
+
+    @pytest.mark.asyncio
+    async def test_process_flow_import_with_invalid_flow(self):
+        """Test process_flow_import with invalid flow"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        params = {"flow": "invalid_flow", "kind": "triples"}
+        
+        with pytest.raises(RuntimeError, match="Invalid flow"):
+            await manager.process_flow_import("ws", "running", params)
+
+    @pytest.mark.asyncio
+    async def test_process_flow_import_with_invalid_kind(self):
+        """Test process_flow_import with invalid kind"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "triples-store": {"queue": "test_queue"}
+            }
+        }
+        
+        with patch('trustgraph.gateway.dispatch.manager.import_dispatchers') as mock_dispatchers:
+            mock_dispatchers.__contains__.return_value = False
+            
+            params = {"flow": "test_flow", "kind": "invalid_kind"}
+            
+            with pytest.raises(RuntimeError, match="Invalid kind"):
+                await manager.process_flow_import("ws", "running", params)
+
+    @pytest.mark.asyncio
+    async def test_process_flow_export_with_valid_flow_and_kind(self):
+        """Test process_flow_export with valid flow and kind"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "triples-store": {"queue": "test_queue"}
+            }
+        }
+        
+        with patch('trustgraph.gateway.dispatch.manager.export_dispatchers') as mock_dispatchers, \
+             patch('uuid.uuid4') as mock_uuid:
+            mock_uuid.return_value = "test-uuid"
+            mock_dispatcher_class = Mock()
+            mock_dispatcher = Mock()
+            mock_dispatcher_class.return_value = mock_dispatcher
+            mock_dispatchers.__getitem__.return_value = mock_dispatcher_class
+            mock_dispatchers.__contains__.return_value = True
+            
+            params = {"flow": "test_flow", "kind": "triples"}
+            result = await manager.process_flow_export("ws", "running", params)
+            
+            mock_dispatcher_class.assert_called_once_with(
+                pulsar_client=mock_pulsar_client,
+                ws="ws",
+                running="running",
+                queue={"queue": "test_queue"},
+                consumer="api-gateway-test-uuid",
+                subscriber="api-gateway-test-uuid"
+            )
+            assert result == mock_dispatcher
+
+    @pytest.mark.asyncio
+    async def test_process_socket(self):
+        """Test process_socket method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        with patch('trustgraph.gateway.dispatch.manager.Mux') as mock_mux:
+            mock_mux_instance = Mock()
+            mock_mux.return_value = mock_mux_instance
+            
+            result = await manager.process_socket("ws", "running", {})
+            
+            mock_mux.assert_called_once_with(manager, "ws", "running")
+            assert result == mock_mux_instance
+
+    @pytest.mark.asyncio
+    async def test_process_flow_service(self):
+        """Test process_flow_service method"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        manager.invoke_flow_service = AsyncMock(return_value="flow_result")
+        
+        params = {"flow": "test_flow", "kind": "agent"}
+        result = await manager.process_flow_service("data", "responder", params)
+        
+        manager.invoke_flow_service.assert_called_once_with("data", "responder", "test_flow", "agent")
+        assert result == "flow_result"
+
+    @pytest.mark.asyncio
+    async def test_invoke_flow_service_with_existing_dispatcher(self):
+        """Test invoke_flow_service with existing dispatcher"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Pre-populate with existing dispatcher
+        mock_dispatcher = Mock()
+        mock_dispatcher.process = AsyncMock(return_value="cached_result")
+        manager.dispatchers[("test_flow", "agent")] = mock_dispatcher
+        
+        result = await manager.invoke_flow_service("data", "responder", "test_flow", "agent")
+        
+        mock_dispatcher.process.assert_called_once_with("data", "responder")
+        assert result == "cached_result"
+
+    @pytest.mark.asyncio
+    async def test_invoke_flow_service_creates_request_response_dispatcher(self):
+        """Test invoke_flow_service creates request-response dispatcher"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "agent": {
+                    "request": "agent_request_queue",
+                    "response": "agent_response_queue"
+                }
+            }
+        }
+        
+        with patch('trustgraph.gateway.dispatch.manager.request_response_dispatchers') as mock_dispatchers:
+            mock_dispatcher_class = Mock()
+            mock_dispatcher = Mock()
+            mock_dispatcher.start = AsyncMock()
+            mock_dispatcher.process = AsyncMock(return_value="new_result")
+            mock_dispatcher_class.return_value = mock_dispatcher
+            mock_dispatchers.__getitem__.return_value = mock_dispatcher_class
+            mock_dispatchers.__contains__.return_value = True
+            
+            result = await manager.invoke_flow_service("data", "responder", "test_flow", "agent")
+            
+            # Verify dispatcher was created with correct parameters
+            mock_dispatcher_class.assert_called_once_with(
+                pulsar_client=mock_pulsar_client,
+                request_queue="agent_request_queue",
+                response_queue="agent_response_queue",
+                timeout=120,
+                consumer="api-gateway-test_flow-agent-request",
+                subscriber="api-gateway-test_flow-agent-request"
+            )
+            mock_dispatcher.start.assert_called_once()
+            mock_dispatcher.process.assert_called_once_with("data", "responder")
+            
+            # Verify dispatcher was cached
+            assert manager.dispatchers[("test_flow", "agent")] == mock_dispatcher
+            assert result == "new_result"
+
+    @pytest.mark.asyncio
+    async def test_invoke_flow_service_creates_sender_dispatcher(self):
+        """Test invoke_flow_service creates sender dispatcher"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "text-load": {"queue": "text_load_queue"}
+            }
+        }
+        
+        with patch('trustgraph.gateway.dispatch.manager.request_response_dispatchers') as mock_rr_dispatchers, \
+             patch('trustgraph.gateway.dispatch.manager.sender_dispatchers') as mock_sender_dispatchers:
+            mock_rr_dispatchers.__contains__.return_value = False
+            mock_sender_dispatchers.__contains__.return_value = True
+            
+            mock_dispatcher_class = Mock()
+            mock_dispatcher = Mock()
+            mock_dispatcher.start = AsyncMock()
+            mock_dispatcher.process = AsyncMock(return_value="sender_result")
+            mock_dispatcher_class.return_value = mock_dispatcher
+            mock_sender_dispatchers.__getitem__.return_value = mock_dispatcher_class
+            
+            result = await manager.invoke_flow_service("data", "responder", "test_flow", "text-load")
+            
+            # Verify dispatcher was created with correct parameters
+            mock_dispatcher_class.assert_called_once_with(
+                pulsar_client=mock_pulsar_client,
+                queue={"queue": "text_load_queue"}
+            )
+            mock_dispatcher.start.assert_called_once()
+            mock_dispatcher.process.assert_called_once_with("data", "responder")
+            
+            # Verify dispatcher was cached
+            assert manager.dispatchers[("test_flow", "text-load")] == mock_dispatcher
+            assert result == "sender_result"
+
+    @pytest.mark.asyncio
+    async def test_invoke_flow_service_invalid_flow(self):
+        """Test invoke_flow_service with invalid flow"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        with pytest.raises(RuntimeError, match="Invalid flow"):
+            await manager.invoke_flow_service("data", "responder", "invalid_flow", "agent")
+
+    @pytest.mark.asyncio
+    async def test_invoke_flow_service_unsupported_kind_by_flow(self):
+        """Test invoke_flow_service with kind not supported by flow"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow without agent interface
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "text-completion": {"request": "req", "response": "resp"}
+            }
+        }
+        
+        with pytest.raises(RuntimeError, match="This kind not supported by flow"):
+            await manager.invoke_flow_service("data", "responder", "test_flow", "agent")
+
+    @pytest.mark.asyncio
+    async def test_invoke_flow_service_invalid_kind(self):
+        """Test invoke_flow_service with invalid kind"""
+        mock_pulsar_client = Mock()
+        mock_config_receiver = Mock()
+        manager = DispatcherManager(mock_pulsar_client, mock_config_receiver)
+        
+        # Setup test flow with interface but unsupported kind
+        manager.flows["test_flow"] = {
+            "interfaces": {
+                "invalid-kind": {"request": "req", "response": "resp"}
+            }
+        }
+        
+        with patch('trustgraph.gateway.dispatch.manager.request_response_dispatchers') as mock_rr_dispatchers, \
+             patch('trustgraph.gateway.dispatch.manager.sender_dispatchers') as mock_sender_dispatchers:
+            mock_rr_dispatchers.__contains__.return_value = False
+            mock_sender_dispatchers.__contains__.return_value = False
+            
+            with pytest.raises(RuntimeError, match="Invalid kind"):
+                await manager.invoke_flow_service("data", "responder", "test_flow", "invalid-kind")

tests/unit/test_gateway/test_running.py

@@ -0,0 +1,90 @@
+"""
+Tests for Gateway Running utility class
+"""
+
+import pytest
+
+from trustgraph.gateway.running import Running
+
+
+class TestRunning:
+    """Test cases for Running class"""
+
+    def test_running_initialization(self):
+        """Test Running class initialization"""
+        running = Running()
+        
+        # Should start with running = True
+        assert running.running is True
+
+    def test_running_get_method(self):
+        """Test Running.get() method returns current state"""
+        running = Running()
+        
+        # Should return True initially
+        assert running.get() is True
+        
+        # Should return False after stopping
+        running.stop()
+        assert running.get() is False
+
+    def test_running_stop_method(self):
+        """Test Running.stop() method sets running to False"""
+        running = Running()
+        
+        # Initially should be True
+        assert running.running is True
+        
+        # After calling stop(), should be False
+        running.stop()
+        assert running.running is False
+
+    def test_running_stop_is_idempotent(self):
+        """Test that calling stop() multiple times is safe"""
+        running = Running()
+        
+        # Stop multiple times
+        running.stop()
+        assert running.running is False
+        
+        running.stop()
+        assert running.running is False
+        
+        # get() should still return False
+        assert running.get() is False
+
+    def test_running_state_transitions(self):
+        """Test the complete state transition from running to stopped"""
+        running = Running()
+        
+        # Initial state: running
+        assert running.get() is True
+        assert running.running is True
+        
+        # Transition to stopped
+        running.stop()
+        assert running.get() is False
+        assert running.running is False
+
+    def test_running_multiple_instances_independent(self):
+        """Test that multiple Running instances are independent"""
+        running1 = Running()
+        running2 = Running()
+        
+        # Both should start as running
+        assert running1.get() is True
+        assert running2.get() is True
+        
+        # Stop only one
+        running1.stop()
+        
+        # States should be independent
+        assert running1.get() is False
+        assert running2.get() is True
+        
+        # Stop the other
+        running2.stop()
+        
+        # Both should now be stopped
+        assert running1.get() is False
+        assert running2.get() is False

tests/unit/test_gateway/test_service.py

@@ -0,0 +1,338 @@
+"""
+Tests for Gateway Service API
+"""
+
+import pytest
+import asyncio
+from unittest.mock import Mock, patch, MagicMock, AsyncMock
+from aiohttp import web
+import pulsar
+
+from trustgraph.gateway.service import Api, run, default_pulsar_host, default_prometheus_url, default_timeout, default_port, default_api_token
+
+
+class TestApi:
+    """Test cases for Api class"""
+
+    def test_api_initialization_with_defaults(self):
+        """Test Api initialization with default values"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            api = Api()
+            
+            assert api.port == default_port
+            assert api.timeout == default_timeout
+            assert api.pulsar_host == default_pulsar_host
+            assert api.pulsar_api_key is None
+            assert api.prometheus_url == default_prometheus_url + "/"
+            assert api.auth.allow_all is True
+            
+            # Verify Pulsar client was created without API key
+            mock_client.assert_called_once_with(
+                default_pulsar_host,
+                listener_name=None
+            )
+
+    def test_api_initialization_with_custom_config(self):
+        """Test Api initialization with custom configuration"""
+        config = {
+            "port": 9000,
+            "timeout": 300,
+            "pulsar_host": "pulsar://custom-host:6650",
+            "pulsar_api_key": "test-api-key",
+            "pulsar_listener": "custom-listener",
+            "prometheus_url": "http://custom-prometheus:9090",
+            "api_token": "secret-token"
+        }
+        
+        with patch('pulsar.Client') as mock_client, \
+             patch('pulsar.AuthenticationToken') as mock_auth:
+            mock_client.return_value = Mock()
+            mock_auth.return_value = Mock()
+            
+            api = Api(**config)
+            
+            assert api.port == 9000
+            assert api.timeout == 300
+            assert api.pulsar_host == "pulsar://custom-host:6650"
+            assert api.pulsar_api_key == "test-api-key"
+            assert api.prometheus_url == "http://custom-prometheus:9090/"
+            assert api.auth.token == "secret-token"
+            assert api.auth.allow_all is False
+            
+            # Verify Pulsar client was created with API key
+            mock_auth.assert_called_once_with("test-api-key")
+            mock_client.assert_called_once_with(
+                "pulsar://custom-host:6650",
+                listener_name="custom-listener",
+                authentication=mock_auth.return_value
+            )
+
+    def test_api_initialization_with_pulsar_api_key(self):
+        """Test Api initialization with Pulsar API key authentication"""
+        with patch('pulsar.Client') as mock_client, \
+             patch('pulsar.AuthenticationToken') as mock_auth:
+            mock_client.return_value = Mock()
+            mock_auth.return_value = Mock()
+            
+            api = Api(pulsar_api_key="test-key")
+            
+            mock_auth.assert_called_once_with("test-key")
+            mock_client.assert_called_once_with(
+                default_pulsar_host,
+                listener_name=None,
+                authentication=mock_auth.return_value
+            )
+
+    def test_api_initialization_prometheus_url_normalization(self):
+        """Test that prometheus_url gets normalized with trailing slash"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            # Test URL without trailing slash
+            api = Api(prometheus_url="http://prometheus:9090")
+            assert api.prometheus_url == "http://prometheus:9090/"
+            
+            # Test URL with trailing slash
+            api = Api(prometheus_url="http://prometheus:9090/")
+            assert api.prometheus_url == "http://prometheus:9090/"
+
+    def test_api_initialization_empty_api_token_means_no_auth(self):
+        """Test that empty API token results in allow_all authentication"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            api = Api(api_token="")
+            assert api.auth.allow_all is True
+
+    def test_api_initialization_none_api_token_means_no_auth(self):
+        """Test that None API token results in allow_all authentication"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            api = Api(api_token=None)
+            assert api.auth.allow_all is True
+
+    @pytest.mark.asyncio
+    async def test_app_factory_creates_application(self):
+        """Test that app_factory creates aiohttp application"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            api = Api()
+            
+            # Mock the dependencies
+            api.config_receiver = Mock()
+            api.config_receiver.start = AsyncMock()
+            api.endpoint_manager = Mock()
+            api.endpoint_manager.add_routes = Mock()
+            api.endpoint_manager.start = AsyncMock()
+            
+            app = await api.app_factory()
+            
+            assert isinstance(app, web.Application)
+            assert app.client_max_size == 256 * 1024 * 1024
+            
+            # Verify that config receiver was started
+            api.config_receiver.start.assert_called_once()
+            
+            # Verify that endpoint manager was configured
+            api.endpoint_manager.add_routes.assert_called_once_with(app)
+            api.endpoint_manager.start.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_app_factory_with_custom_endpoints(self):
+        """Test app_factory with custom endpoints"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            api = Api()
+            
+            # Mock custom endpoints
+            mock_endpoint1 = Mock()
+            mock_endpoint1.add_routes = Mock()
+            mock_endpoint1.start = AsyncMock()
+            
+            mock_endpoint2 = Mock()
+            mock_endpoint2.add_routes = Mock()
+            mock_endpoint2.start = AsyncMock()
+            
+            api.endpoints = [mock_endpoint1, mock_endpoint2]
+            
+            # Mock the dependencies
+            api.config_receiver = Mock()
+            api.config_receiver.start = AsyncMock()
+            api.endpoint_manager = Mock()
+            api.endpoint_manager.add_routes = Mock()
+            api.endpoint_manager.start = AsyncMock()
+            
+            app = await api.app_factory()
+            
+            # Verify custom endpoints were configured
+            mock_endpoint1.add_routes.assert_called_once_with(app)
+            mock_endpoint1.start.assert_called_once()
+            mock_endpoint2.add_routes.assert_called_once_with(app)
+            mock_endpoint2.start.assert_called_once()
+
+    def test_run_method_calls_web_run_app(self):
+        """Test that run method calls web.run_app"""
+        with patch('pulsar.Client') as mock_client, \
+             patch('aiohttp.web.run_app') as mock_run_app:
+            mock_client.return_value = Mock()
+            
+            api = Api(port=8080)
+            api.run()
+            
+            mock_run_app.assert_called_once_with(api.app_factory(), port=8080)
+
+    def test_api_components_initialization(self):
+        """Test that all API components are properly initialized"""
+        with patch('pulsar.Client') as mock_client:
+            mock_client.return_value = Mock()
+            
+            api = Api()
+            
+            # Verify all components are initialized
+            assert api.config_receiver is not None
+            assert api.dispatcher_manager is not None
+            assert api.endpoint_manager is not None
+            assert api.endpoints == []
+            
+            # Verify component relationships
+            assert api.dispatcher_manager.pulsar_client == api.pulsar_client
+            assert api.dispatcher_manager.config_receiver == api.config_receiver
+            assert api.endpoint_manager.dispatcher_manager == api.dispatcher_manager
+            assert api.endpoint_manager.auth == api.auth
+
+
+class TestRunFunction:
+    """Test cases for the run() function"""
+
+    @patch('trustgraph.gateway.service.Api')
+    @patch('trustgraph.gateway.service.start_http_server')
+    @patch('argparse.ArgumentParser.parse_args')
+    def test_run_function_with_metrics_enabled(self, mock_parse_args, mock_start_http_server, mock_api):
+        """Test run function with metrics enabled"""
+        # Mock command line arguments
+        mock_args = Mock()
+        mock_args.metrics = True
+        mock_args.metrics_port = 8000
+        mock_parse_args.return_value = mock_args
+        
+        # Mock the Api instance
+        mock_api_instance = Mock()
+        mock_api.return_value = mock_api_instance
+        
+        # Mock vars() to return a dict
+        with patch('builtins.vars') as mock_vars:
+            mock_vars.return_value = {
+                'metrics': True,
+                'metrics_port': 8000,
+                'pulsar_host': default_pulsar_host,
+                'timeout': default_timeout
+            }
+            
+            run()
+            
+            # Verify metrics server was started
+            mock_start_http_server.assert_called_once_with(8000)
+            
+            # Verify Api was created and run was called
+            mock_api.assert_called_once()
+            mock_api_instance.run.assert_called_once()
+
+    @patch('trustgraph.gateway.service.Api')
+    @patch('trustgraph.gateway.service.start_http_server')
+    @patch('argparse.ArgumentParser.parse_args')
+    def test_run_function_with_metrics_disabled(self, mock_parse_args, mock_start_http_server, mock_api):
+        """Test run function with metrics disabled"""
+        # Mock command line arguments
+        mock_args = Mock()
+        mock_args.metrics = False
+        mock_parse_args.return_value = mock_args
+        
+        # Mock the Api instance
+        mock_api_instance = Mock()
+        mock_api.return_value = mock_api_instance
+        
+        # Mock vars() to return a dict
+        with patch('builtins.vars') as mock_vars:
+            mock_vars.return_value = {
+                'metrics': False,
+                'metrics_port': 8000,
+                'pulsar_host': default_pulsar_host,
+                'timeout': default_timeout
+            }
+            
+            run()
+            
+            # Verify metrics server was NOT started
+            mock_start_http_server.assert_not_called()
+            
+            # Verify Api was created and run was called
+            mock_api.assert_called_once()
+            mock_api_instance.run.assert_called_once()
+
+    @patch('trustgraph.gateway.service.Api')
+    @patch('argparse.ArgumentParser.parse_args')
+    def test_run_function_argument_parsing(self, mock_parse_args, mock_api):
+        """Test that run function properly parses command line arguments"""
+        # Mock command line arguments
+        mock_args = Mock()
+        mock_args.metrics = False
+        mock_parse_args.return_value = mock_args
+        
+        # Mock the Api instance
+        mock_api_instance = Mock()
+        mock_api.return_value = mock_api_instance
+        
+        # Mock vars() to return a dict with all expected arguments
+        expected_args = {
+            'pulsar_host': 'pulsar://test:6650',
+            'pulsar_api_key': 'test-key',
+            'pulsar_listener': 'test-listener',
+            'prometheus_url': 'http://test-prometheus:9090',
+            'port': 9000,
+            'timeout': 300,
+            'api_token': 'secret',
+            'log_level': 'INFO',
+            'metrics': False,
+            'metrics_port': 8001
+        }
+        
+        with patch('builtins.vars') as mock_vars:
+            mock_vars.return_value = expected_args
+            
+            run()
+            
+            # Verify Api was created with the parsed arguments
+            mock_api.assert_called_once_with(**expected_args)
+            mock_api_instance.run.assert_called_once()
+
+    def test_run_function_creates_argument_parser(self):
+        """Test that run function creates argument parser with correct arguments"""
+        with patch('argparse.ArgumentParser') as mock_parser_class:
+            mock_parser = Mock()
+            mock_parser_class.return_value = mock_parser
+            mock_parser.parse_args.return_value = Mock(metrics=False)
+            
+            with patch('trustgraph.gateway.service.Api') as mock_api, \
+                 patch('builtins.vars') as mock_vars:
+                mock_vars.return_value = {'metrics': False}
+                mock_api.return_value = Mock()
+                
+                run()
+                
+                # Verify ArgumentParser was created
+                mock_parser_class.assert_called_once()
+                
+                # Verify add_argument was called for each expected argument
+                expected_arguments = [
+                    'pulsar-host', 'pulsar-api-key', 'pulsar-listener',
+                    'prometheus-url', 'port', 'timeout', 'api-token',
+                    'log-level', 'metrics', 'metrics-port'
+                ]
+                
+                # Check that add_argument was called multiple times (once for each arg)
+                assert mock_parser.add_argument.call_count >= len(expected_arguments)