unittests

Author: daimor

.github/workflows/main.yml

@@ -41,6 +41,10 @@ jobs:
           python -m pip install -e .
           python -m pip install build
           python -m build
+      - name: Test
+        run: |
+          python -m pip install testcontainers-iris
+          pytest -vs --container containers.intersystems.com/intersystems/iris-community:latest-preview
       - name: Create Beta Release
         id: create_release
         if: github.event_name == 'push'

tests/conftest.py

@@ -0,0 +1,90 @@
+import pytest
+from testcontainers.iris import IRISContainer
+
+
+def pytest_addoption(parser):
+    group = parser.getgroup("iris")
+
+    group.addoption(
+        "--embedded",
+        action="store_true",
+        help="Use embedded mode",
+    )
+
+    group.addoption(
+        "--iris-host",
+        action="store",
+        default="localhost",
+        help="Hostname",
+    )
+
+    group.addoption(
+        "--iris-port",
+        action="store",
+        default=1972,
+        type=int,
+        help="Port",
+    )
+
+    group.addoption(
+        "--iris-namespace",
+        action="store",
+        default="USER",
+        help="Namespace",
+    )
+
+    group.addoption(
+        "--iris-username",
+        action="store",
+        default="_SYSTEM",
+        help="Username",
+    )
+
+    group.addoption(
+        "--iris-password",
+        action="store",
+        default="SYS",
+        help="Password",
+    )
+
+    group.addoption(
+        "--container",
+        action="store",
+        default=None,
+        type=str,
+        help="Docker image with IRIS",
+    )
+
+
+def pytest_configure(config: pytest.Config):
+    global iris
+    iris = None
+    if not config.option.container:
+        return
+    config.option.embedded = False
+    print("Starting IRIS container:", config.option.container)
+    try:
+        iris = IRISContainer(
+            config.option.container,
+            username="test",
+            password="test",
+            namespace="TEST",
+        )
+        iris.start()
+        print("Started on port:", iris.get_exposed_port(1972))
+        print(iris.get_connection_url())
+        config.option.iris_host = "localhost"
+        config.option.iris_username = iris.username
+        config.option.iris_password = iris.password
+        config.option.iris_namespace = iris.namespace
+        config.option.iris_port = int(iris.get_exposed_port(1972))
+    except Exception as ex:
+        iris = None
+        pytest.exit("Failed to start IRIS container: " + str(ex))
+
+
+def pytest_unconfigure(config):
+    global iris
+    if iris:
+        print("Stopping IRIS container")
+        iris.stop()

tests/test_server.py

@@ -0,0 +1,54 @@
+import sys
+import pytest
+from mcp.client.session import ClientSession
+from mcp.client.stdio import StdioServerParameters, stdio_client
+from mcp_server_iris.server import server
+
+
+@pytest.fixture
+async def iris_config(request):
+    return {
+        "IRIS_HOSTNAME": request.config.option.iris_host,
+        "IRIS_PORT": str(request.config.option.iris_port),
+        "IRIS_NAMESPACE": request.config.option.iris_namespace,
+        "IRIS_USERNAME": request.config.option.iris_username,
+        "IRIS_PASSWORD": request.config.option.iris_password,
+    }
+
+
+def test_server_initialization():
+    """Test that the server initializes correctly."""
+    assert server.name == "InterSystems IRIS MCP Server"
+
+
+@pytest.mark.asyncio
+async def test_list_tools():
+    """Test that list_tools returns expected tools."""
+    tools = await server.list_tools()
+    assert len(tools) > 0
+
+
+@pytest.mark.asyncio
+async def test_params_in_tool():
+    # print(f"Tools: {tools}")
+    tools = await server.list_tools()
+    execute_sql_tool = next(t for t in tools if t.name == "execute_sql")
+    props = list(execute_sql_tool.inputSchema["properties"].keys())
+    assert len(props) == 2
+    assert props[0] == "query"
+    assert props[1] == "params"
+
+
+@pytest.mark.asyncio
+async def test_call_tool_execute_sql(iris_config):
+    """Test calling execute_sql with a query."""
+    async with stdio_client(StdioServerParameters(command=sys.executable, args=["-m", "mcp_server_iris"], env=iris_config)) as (read, write):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            _ = await session.call_tool(
+                "execute_sql",
+                {
+                    "query": "select $namespace, $zversion",
+                    "params": [],
+                },
+            )