diff --git a/tests/unit/test_gateway/test_config_receiver.py b/tests/unit/test_gateway/test_config_receiver.py new file mode 100644 index 00000000..81019f0c --- /dev/null +++ b/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 == {} \ No newline at end of file diff --git a/tests/unit/test_gateway/test_dispatch_manager.py b/tests/unit/test_gateway/test_dispatch_manager.py new file mode 100644 index 00000000..51c52bab --- /dev/null +++ b/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") \ No newline at end of file diff --git a/tests/unit/test_gateway/test_running.py b/tests/unit/test_gateway/test_running.py new file mode 100644 index 00000000..be02dfe7 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/tests/unit/test_gateway/test_service.py b/tests/unit/test_gateway/test_service.py new file mode 100644 index 00000000..8cfc1517 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/tests/unit/test_rev_gateway/test_rev_gateway_service.py b/tests/unit/test_rev_gateway/test_rev_gateway_service.py new file mode 100644 index 00000000..6d87a2e0 --- /dev/null +++ b/tests/unit/test_rev_gateway/test_rev_gateway_service.py @@ -0,0 +1,544 @@ +""" +Tests for Reverse Gateway Service +""" + +import pytest +import asyncio +from unittest.mock import MagicMock, AsyncMock, patch, Mock +from aiohttp import WSMsgType, ClientWebSocketResponse +import json + +from trustgraph.rev_gateway.service import ReverseGateway, parse_args, run + + +class TestReverseGateway: + """Test cases for ReverseGateway class""" + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_initialization_defaults(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway initialization with default parameters""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + assert gateway.websocket_uri == "ws://localhost:7650/out" + assert gateway.host == "localhost" + assert gateway.port == 7650 + assert gateway.scheme == "ws" + assert gateway.path == "/out" + assert gateway.url == "ws://localhost:7650/out" + assert gateway.max_workers == 10 + assert gateway.running is False + assert gateway.reconnect_delay == 3.0 + assert gateway.pulsar_host == "pulsar://pulsar:6650" + assert gateway.pulsar_api_key is None + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_initialization_custom_params(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway initialization with custom parameters""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway( + websocket_uri="wss://example.com:8080/websocket", + max_workers=20, + pulsar_host="pulsar://custom:6650", + pulsar_api_key="test-key", + pulsar_listener="test-listener" + ) + + assert gateway.websocket_uri == "wss://example.com:8080/websocket" + assert gateway.host == "example.com" + assert gateway.port == 8080 + assert gateway.scheme == "wss" + assert gateway.path == "/websocket" + assert gateway.url == "wss://example.com:8080/websocket" + assert gateway.max_workers == 20 + assert gateway.pulsar_host == "pulsar://custom:6650" + assert gateway.pulsar_api_key == "test-key" + assert gateway.pulsar_listener == "test-listener" + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_initialization_with_missing_path(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway initialization with WebSocket URI missing path""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway(websocket_uri="ws://example.com") + + assert gateway.path == "/ws" + assert gateway.url == "ws://example.com/ws" + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_initialization_invalid_scheme(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway initialization with invalid WebSocket scheme""" + with pytest.raises(ValueError, match="WebSocket URI must use ws:// or wss:// scheme"): + ReverseGateway(websocket_uri="http://example.com") + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_initialization_missing_hostname(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway initialization with missing hostname""" + with pytest.raises(ValueError, match="WebSocket URI must include hostname"): + ReverseGateway(websocket_uri="ws://") + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_pulsar_client_with_auth(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway creates Pulsar client with authentication""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + with patch('pulsar.AuthenticationToken') as mock_auth: + mock_auth_instance = MagicMock() + mock_auth.return_value = mock_auth_instance + + gateway = ReverseGateway( + pulsar_api_key="test-key", + pulsar_listener="test-listener" + ) + + mock_auth.assert_called_once_with("test-key") + mock_pulsar_client.assert_called_once_with( + "pulsar://pulsar:6650", + listener_name="test-listener", + authentication=mock_auth_instance + ) + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_connect_success(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway successful connection""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + with patch('aiohttp.ClientSession') as mock_session_class: + mock_session = AsyncMock() + mock_ws = AsyncMock() + mock_session.ws_connect.return_value = mock_ws + mock_session_class.return_value = mock_session + + result = await gateway.connect() + + assert result is True + assert gateway.session == mock_session + assert gateway.ws == mock_ws + mock_session.ws_connect.assert_called_once_with(gateway.url) + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_connect_failure(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway connection failure""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + with patch('aiohttp.ClientSession') as mock_session_class: + mock_session = AsyncMock() + mock_session.ws_connect.side_effect = Exception("Connection failed") + mock_session_class.return_value = mock_session + + result = await gateway.connect() + + assert result is False + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_disconnect(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway disconnect""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + # Mock websocket and session + mock_ws = AsyncMock() + mock_ws.closed = False + mock_session = AsyncMock() + mock_session.closed = False + + gateway.ws = mock_ws + gateway.session = mock_session + + await gateway.disconnect() + + mock_ws.close.assert_called_once() + mock_session.close.assert_called_once() + assert gateway.ws is None + assert gateway.session is None + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_send_message(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway send message""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + # Mock websocket + mock_ws = AsyncMock() + mock_ws.closed = False + gateway.ws = mock_ws + + test_message = {"id": "test", "data": "hello"} + + await gateway.send_message(test_message) + + mock_ws.send_str.assert_called_once_with(json.dumps(test_message)) + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_send_message_closed_connection(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway send message with closed connection""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + # Mock closed websocket + mock_ws = AsyncMock() + mock_ws.closed = True + gateway.ws = mock_ws + + test_message = {"id": "test", "data": "hello"} + + await gateway.send_message(test_message) + + # Should not call send_str on closed connection + mock_ws.send_str.assert_not_called() + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_handle_message(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway handle message""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + mock_dispatcher_instance = AsyncMock() + mock_dispatcher_instance.handle_message.return_value = {"response": "success"} + mock_dispatcher.return_value = mock_dispatcher_instance + + gateway = ReverseGateway() + + # Mock send_message + gateway.send_message = AsyncMock() + + test_message = '{"id": "test", "service": "test-service", "request": {"data": "test"}}' + + await gateway.handle_message(test_message) + + mock_dispatcher_instance.handle_message.assert_called_once_with({ + "id": "test", + "service": "test-service", + "request": {"data": "test"} + }) + gateway.send_message.assert_called_once_with({"response": "success"}) + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_handle_message_invalid_json(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway handle message with invalid JSON""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + + # Mock send_message + gateway.send_message = AsyncMock() + + test_message = 'invalid json' + + # Should not raise exception + await gateway.handle_message(test_message) + + # Should not call send_message due to error + gateway.send_message.assert_not_called() + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_listen_text_message(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway listen with text message""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + gateway.running = True + + # Mock websocket + mock_ws = AsyncMock() + mock_ws.closed = False + gateway.ws = mock_ws + + # Mock handle_message + gateway.handle_message = AsyncMock() + + # Mock message + mock_msg = MagicMock() + mock_msg.type = WSMsgType.TEXT + mock_msg.data = '{"test": "message"}' + + # Mock receive to return message once, then close + mock_ws.receive.side_effect = [mock_msg, Exception("Test stop")] + + with pytest.raises(Exception, match="Test stop"): + await gateway.listen() + + gateway.handle_message.assert_called_once_with('{"test": "message"}') + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_listen_binary_message(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway listen with binary message""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + gateway.running = True + + # Mock websocket + mock_ws = AsyncMock() + mock_ws.closed = False + gateway.ws = mock_ws + + # Mock handle_message + gateway.handle_message = AsyncMock() + + # Mock message + mock_msg = MagicMock() + mock_msg.type = WSMsgType.BINARY + mock_msg.data = b'{"test": "binary"}' + + # Mock receive to return message once, then close + mock_ws.receive.side_effect = [mock_msg, Exception("Test stop")] + + with pytest.raises(Exception, match="Test stop"): + await gateway.listen() + + gateway.handle_message.assert_called_once_with('{"test": "binary"}') + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_listen_close_message(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway listen with close message""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + gateway.running = True + + # Mock websocket + mock_ws = AsyncMock() + mock_ws.closed = False + gateway.ws = mock_ws + + # Mock handle_message + gateway.handle_message = AsyncMock() + + # Mock message + mock_msg = MagicMock() + mock_msg.type = WSMsgType.CLOSE + + # Mock receive to return close message + mock_ws.receive.return_value = mock_msg + + await gateway.listen() + + # Should not call handle_message for close message + gateway.handle_message.assert_not_called() + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_shutdown(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway shutdown""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + mock_dispatcher_instance = AsyncMock() + mock_dispatcher.return_value = mock_dispatcher_instance + + gateway = ReverseGateway() + gateway.running = True + + # Mock disconnect + gateway.disconnect = AsyncMock() + + await gateway.shutdown() + + assert gateway.running is False + mock_dispatcher_instance.shutdown.assert_called_once() + gateway.disconnect.assert_called_once() + mock_client_instance.close.assert_called_once() + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + def test_reverse_gateway_stop(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway stop""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + gateway = ReverseGateway() + gateway.running = True + + gateway.stop() + + assert gateway.running is False + + +class TestReverseGatewayRun: + """Test cases for ReverseGateway run method""" + + @patch('trustgraph.rev_gateway.service.ConfigReceiver') + @patch('trustgraph.rev_gateway.service.MessageDispatcher') + @patch('pulsar.Client') + @pytest.mark.asyncio + async def test_reverse_gateway_run_successful_cycle(self, mock_pulsar_client, mock_dispatcher, mock_config_receiver): + """Test ReverseGateway run method with successful connect/listen cycle""" + mock_client_instance = MagicMock() + mock_pulsar_client.return_value = mock_client_instance + + mock_config_receiver_instance = AsyncMock() + mock_config_receiver.return_value = mock_config_receiver_instance + + gateway = ReverseGateway() + + # Mock methods + gateway.connect = AsyncMock(return_value=True) + gateway.listen = AsyncMock() + gateway.disconnect = AsyncMock() + gateway.shutdown = AsyncMock() + + # Stop after one iteration + call_count = 0 + async def mock_connect(): + nonlocal call_count + call_count += 1 + if call_count == 1: + return True + else: + gateway.running = False + return False + + gateway.connect = mock_connect + + await gateway.run() + + mock_config_receiver_instance.start.assert_called_once() + gateway.listen.assert_called_once() + gateway.disconnect.assert_called_once() + gateway.shutdown.assert_called_once() + + +class TestReverseGatewayArgs: + """Test cases for argument parsing and run function""" + + def test_parse_args_defaults(self): + """Test parse_args with default values""" + import sys + + # Mock sys.argv + original_argv = sys.argv + sys.argv = ['reverse-gateway'] + + try: + args = parse_args() + + assert args.websocket_uri is None + assert args.max_workers == 10 + assert args.pulsar_host is None + assert args.pulsar_api_key is None + assert args.pulsar_listener is None + finally: + sys.argv = original_argv + + def test_parse_args_custom_values(self): + """Test parse_args with custom values""" + import sys + + # Mock sys.argv + original_argv = sys.argv + sys.argv = [ + 'reverse-gateway', + '--websocket-uri', 'ws://custom:8080/ws', + '--max-workers', '20', + '--pulsar-host', 'pulsar://custom:6650', + '--pulsar-api-key', 'test-key', + '--pulsar-listener', 'test-listener' + ] + + try: + args = parse_args() + + assert args.websocket_uri == 'ws://custom:8080/ws' + assert args.max_workers == 20 + assert args.pulsar_host == 'pulsar://custom:6650' + assert args.pulsar_api_key == 'test-key' + assert args.pulsar_listener == 'test-listener' + finally: + sys.argv = original_argv + + @patch('trustgraph.rev_gateway.service.ReverseGateway') + @patch('asyncio.run') + def test_run_function(self, mock_asyncio_run, mock_gateway_class): + """Test run function""" + import sys + + # Mock sys.argv + original_argv = sys.argv + sys.argv = ['reverse-gateway', '--max-workers', '15'] + + try: + mock_gateway_instance = MagicMock() + mock_gateway_instance.url = "ws://localhost:7650/out" + mock_gateway_instance.pulsar_host = "pulsar://pulsar:6650" + mock_gateway_class.return_value = mock_gateway_instance + + run() + + mock_gateway_class.assert_called_once_with( + websocket_uri=None, + max_workers=15, + pulsar_host=None, + pulsar_api_key=None, + pulsar_listener=None + ) + mock_asyncio_run.assert_called_once_with(mock_gateway_instance.run()) + finally: + sys.argv = original_argv \ No newline at end of file