Add unit tests for business operation, process, and service components; update .gitignore to exclude coverage files

Author: grongierisc

.gitignore

@@ -18,3 +18,4 @@ waitISC.log
 
 prof
 src/tests/bench/result.txt
+.coverage

src/tests/test_business_operation.py

@@ -0,0 +1,69 @@
+import pytest
+from unittest.mock import MagicMock, patch
+from iop._business_operation import _BusinessOperation
+from registerFiles.message import SimpleMessage
+
+@pytest.fixture
+def operation():
+    op = _BusinessOperation()
+    op.iris_handle = MagicMock()
+    return op
+
+def test_message_handling(operation):
+    # Test on_message
+    request = SimpleMessage(integer=1, string='test')
+    assert operation.on_message(request) is None
+    
+    # Test deprecated OnMessage
+    assert operation.OnMessage(request) is None
+
+def test_keepalive(operation):
+    assert operation.on_keepalive() is None
+
+def test_adapter_handling():
+    # Test adapter setup with mock IRIS adapter
+    op = _BusinessOperation()
+    mock_current = MagicMock()
+    mock_partner = MagicMock()
+    
+    # Setup mock IRIS adapter
+    mock_partner._IsA.return_value = True
+    mock_partner.GetModule.return_value = "some.module"
+    mock_partner.GetClassname.return_value = "SomeAdapter"
+    
+    with patch('importlib.import_module') as mock_import:
+        mock_module = MagicMock()
+        mock_import.return_value = mock_module
+        op._set_iris_handles(mock_current, mock_partner)
+        
+        assert op.iris_handle == mock_current
+
+def test_dispatch_methods(operation):
+    # Test dispatch initialization
+    operation.DISPATCH = [("MessageType1", "handle_type1")]
+    mock_host = MagicMock()
+    
+    operation._dispatch_on_init(mock_host)
+    
+    # Test message dispatch
+    request = SimpleMessage(integer=1, string='test')
+    operation._dispatch_on_message(request)
+    
+    # Verify internal method calls
+    operation.iris_handle.dispatchOnMessage.assert_not_called()
+
+def test_dispatch_with_custom_handlers():
+    class CustomOperation(_BusinessOperation):
+        def handle_simple(self, request: SimpleMessage):
+            return SimpleMessage(integer=request.integer + 1, string="handled")
+    
+    operation = CustomOperation()
+    operation._dispatch_on_init(MagicMock())
+    operation.iris_handle = MagicMock()
+    
+    request = SimpleMessage(integer=1, string='test')
+    response = operation._dispach_message(request)
+    
+    assert isinstance(response, SimpleMessage)
+    assert response.integer == 2
+    assert response.string == "handled"

src/tests/test_business_process.py

@@ -0,0 +1,85 @@
+import pytest
+from unittest.mock import MagicMock, patch
+from iop._business_process import _BusinessProcess
+from registerFiles.message import SimpleMessage, PickledMessage, FullMessage
+
+@pytest.fixture
+def process():
+    proc = _BusinessProcess()
+    proc.iris_handle = MagicMock()
+    return proc
+
+def test_message_handling(process):
+    # Test on_message
+    request = SimpleMessage(integer=1, string='test')
+    assert process.on_message(request) is None
+    
+    # Test on_request
+    assert process.on_request(request) is None
+    
+    # Test on_response
+    response = SimpleMessage(integer=2, string='response')
+    call_request = SimpleMessage(integer=3, string='call_request')
+    call_response = SimpleMessage(integer=4, string='call_response')
+    completion_key = "test_key"
+    
+    assert process.on_response(
+        request, response, call_request, call_response, completion_key
+    ) == response
+    
+    # Test on_complete
+    assert process.on_complete(request, response) == response
+
+def test_async_operations(process):
+    # Test send_request_async
+    target = "target_service"
+    request = SimpleMessage(integer=1, string='test')
+    process.send_request_async(target, request)
+    process.iris_handle.dispatchSendRequestAsync.assert_called_once()
+
+    # Test set_timer
+    timeout = 1000
+    completion_key = "timer_key"
+    process.set_timer(timeout, completion_key)
+    process.iris_handle.dispatchSetTimer.assert_called_once_with(
+        timeout, completion_key
+    )
+
+def test_persistent_properties():
+    # Test persistent property handling
+    class ProcessWithProperties(_BusinessProcess):
+        PERSISTENT_PROPERTY_LIST = ["test_prop"]
+        def __init__(self):
+            super().__init__()
+            self.test_prop = "test_value"
+
+    process = ProcessWithProperties()
+    mock_host = MagicMock()
+    
+    # Test save properties
+    process._save_persistent_properties(mock_host)
+    mock_host.setPersistentProperty.assert_called_once_with("test_prop", "test_value")
+    
+    # Test restore properties
+    mock_host.getPersistentProperty.return_value = "restored_value"
+    process._restore_persistent_properties(mock_host)
+    assert process.test_prop == "restored_value"
+
+def test_dispatch_methods(process):
+    mock_host = MagicMock()
+    request = SimpleMessage(integer=1, string='test')
+    response = SimpleMessage(integer=2, string='response')
+
+    # Test dispatch methods
+    process._dispatch_on_init(mock_host)
+    process._dispatch_on_connected(mock_host)
+    process._dispatch_on_request(mock_host, request)
+    process._dispatch_on_response(
+        mock_host, request, response, request, response, "completion_key"
+    )
+    process._dispatch_on_tear_down(mock_host)
+
+def test_reply(process):
+    response = SimpleMessage(integer=1, string='test')
+    process.reply(response)
+    process.iris_handle.dispatchReply.assert_called_once()

src/tests/test_business_service.py

@@ -0,0 +1,63 @@
+import pytest
+from unittest.mock import MagicMock, patch
+from iop._business_service import _BusinessService
+from registerFiles.message import SimpleMessage
+
+@pytest.fixture
+def service():
+    svc = _BusinessService()
+    svc.iris_handle = MagicMock()
+    return svc
+
+def test_process_input(service):
+    # Test on_process_input
+    message = SimpleMessage(integer=1, string='test')
+    assert service.on_process_input(message) is None
+    
+    # Test deprecated OnProcessInput
+    assert service.OnProcessInput(message) is None
+
+def test_adapter_handling():
+    # Test adapter setup with mock IRIS adapter
+    svc = _BusinessService()
+    mock_current = MagicMock()
+    mock_partner = MagicMock()
+    
+    # Setup mock IRIS adapter
+    mock_partner._IsA.return_value = True
+    mock_partner.GetModule.return_value = "some.module"
+    mock_partner.GetClassname.return_value = "SomeAdapter"
+    
+    with patch('importlib.import_module') as mock_import:
+        mock_module = MagicMock()
+        mock_import.return_value = mock_module
+        svc._set_iris_handles(mock_current, mock_partner)
+        
+        assert svc.iris_handle == mock_current
+        assert svc.Adapter is not None
+        assert svc.adapter is not None
+
+def test_dispatch_on_process_input(service):
+    message = SimpleMessage(integer=1, string='test')
+    service._dispatch_on_process_input(message)
+    
+    # Verify the message was processed
+    service.iris_handle.dispatchOnProcessInput.assert_not_called()
+
+def test_custom_service():
+    class CustomService(_BusinessService):
+        def on_process_input(self, message):
+            return SimpleMessage(integer=message.integer * 2, string=f"processed_{message.string}")
+    
+    service = CustomService()
+    service.iris_handle = MagicMock()
+    
+    input_msg = SimpleMessage(integer=5, string='test')
+    result = service.on_process_input(input_msg)
+    
+    assert isinstance(result, SimpleMessage)
+    assert result.integer == 10
+    assert result.string == "processed_test"
+
+def test_wait_for_next_call_interval(service):
+    assert service._wait_for_next_call_interval is False

src/tests/test_iop_director.py

@@ -52,3 +52,95 @@ def test_component_with_iris_classname_and_body(self):
     def test_component_with_nonexistent_iris_classname(self):
         with pytest.raises(RuntimeError):
             _Director.test_component('test', classname='iris.test', body='test')
+
+class TestBusinessService:
+    def test_get_business_service(self):
+        director = _Director()
+        iris.cls("IOP.Director").dispatchCreateBusinessService = MagicMock(return_value=MagicMock())
+        service = director.get_business_service("test")
+        assert service is not None
+        
+    def test_get_business_service_force_session(self):
+        director = _Director()
+        mock_service = MagicMock()
+        mock_service.iris_handle = MagicMock()
+        iris.cls("IOP.Director").dispatchCreateBusinessService = MagicMock(return_value=mock_service)
+        service = director.get_business_service("test", force_session_id=True)
+        assert service.iris_handle.ForceSessionId.called
+
+    def test_create_business_service(self):
+        iris.cls("IOP.Director").dispatchCreateBusinessService = MagicMock(return_value="test")
+        result = _Director.create_business_service("test")
+        assert result == "test"
+
+    def test_create_python_business_service(self):
+        mock_obj = MagicMock()
+        mock_obj.GetClass = MagicMock(return_value="test_class")
+        iris.cls("IOP.Director").dispatchCreateBusinessService = MagicMock(return_value=mock_obj)
+        result = _Director.create_python_business_service("test")
+        assert result == "test_class"
+
+class TestProductionManagement:
+    @pytest.fixture(autouse=True)
+    def setup_mocks(self):
+        self.start_mock = MagicMock()
+        self.stop_mock = MagicMock()
+        self.restart_mock = MagicMock()
+        self.update_mock = MagicMock()
+        iris.cls('Ens.Director').StartProduction = self.start_mock
+        iris.cls('Ens.Director').StopProduction = self.stop_mock
+        iris.cls('Ens.Director').RestartProduction = self.restart_mock
+        iris.cls('Ens.Director').UpdateProduction = self.update_mock
+
+    def test_start_production(self):
+        _Director.start_production("test_prod")
+        self.start_mock.assert_called_once_with("test_prod")
+
+    def test_stop_production(self):
+        _Director.stop_production()
+        self.stop_mock.assert_called_once()
+
+    def test_restart_production(self):
+        _Director.restart_production()
+        self.restart_mock.assert_called_once()
+
+    def test_shutdown_production(self):
+        _Director.shutdown_production()
+        self.stop_mock.assert_called_once_with(10, 1)
+
+    def test_update_production(self):
+        _Director.update_production()
+        self.update_mock.assert_called_once()
+
+    def test_list_productions(self):
+        iris.cls('IOP.Director').dispatchListProductions = MagicMock(return_value=["prod1", "prod2"])
+        result = _Director.list_productions()
+        assert result == ["prod1", "prod2"]
+
+    def test_status_production(self):
+        mock_status = {'Production': 'test_prod', 'Status': 'running'}
+        iris.cls('IOP.Director').StatusProduction = MagicMock(return_value=mock_status)
+        result = _Director.status_production()
+        assert result == mock_status
+
+class TestLogging:
+    def test_format_log(self):
+        test_row = [1, 'Config1', 'Job1', 'Msg1', 'Session1', 'Source1', 'Method1', 
+                    'Stack1', 'Text1', '2023-01-01', 'TraceCat1', 1]
+        result = _Director.format_log(test_row)
+        assert 'Assert' in result
+        assert 'Config1' in result
+
+    def test_format_log_different_types(self):
+        types = {1: 'Assert', 2: 'Error', 3: 'Warning', 4: 'Info', 5: 'Trace', 6: 'Alert'}
+        for type_num, type_str in types.items():
+            test_row = [1, 'Config1', 'Job1', 'Msg1', 'Session1', 'Source1', 'Method1',
+                       'Stack1', 'Text1', '2023-01-01', 'TraceCat1', type_num]
+            result = _Director.format_log(test_row)
+            assert type_str in result
+
+    @pytest.mark.asyncio
+    async def test_log_production_async(self):
+        handler = MagicMock()
+        handler.sigint_log = True
+        await _Director._log_production_async(handler)