Merge branch 'main' into darabos-tweaks

.github/workflows/test.yaml

@@ -0,0 +1,44 @@
+name: test
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+            enable-cache: true
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          eval `ssh-agent -s`
+          ssh-add - <<< '${{ secrets.LYNXSCRIBE_DEPLOY_KEY }}'
+          uv pip install -e lynxkite-core/[dev] lynxkite-app/[dev] lynxkite-graph-analytics/[dev] lynxkite-lynxscribe/ lynxkite-pillow-example/
+        env:
+            UV_SYSTEM_PYTHON: 1
+
+      - name: Run core tests
+        run: |
+          cd lynxkite-core
+          pytest
+
+      - name: Run app tests
+        run: |
+            cd lynxkite-app
+            pytest
+    
+      - name: Run graph analytics tests  
+        run: |
+            cd lynxkite-graph-analytics
+            pytest

README.md

@@ -23,7 +23,8 @@ Install everything like this:
 ```bash
 uv venv
 source .venv/bin/activate
-uv pip install -e lynxkite-core/ lynxkite-app/ lynxkite-graph-analytics/ lynxkite-lynxscribe/ lynxkite-pillow-example/
+# The [dev] tag is only needed if you intend on running tests
+uv pip install -e lynxkite-core/[dev] lynxkite-app/[dev] lynxkite-graph-analytics/[dev] lynxkite-lynxscribe/ lynxkite-pillow-example/
 ```
 
 This also builds the frontend, hopefully very quickly. To run it:
@@ -40,6 +41,16 @@ cd lynxkite-app/web
 npm run dev
 ```
 
+## Executing tests
+
+Just go into each directory and execute `pytest`.
+
+```bash
+# Same thing for lynxkite-core and lynxkite-graph-analytics
+$ cd lynxkite-app
+$ pytest
+```
+
 ## Documentation
 
 To work on the documentation:

lynxkite-app/pyproject.toml

@@ -13,6 +13,11 @@ dependencies = [
     "sse-starlette>=2.2.1",
 ]
 
+[project.optional-dependencies]
+dev = [
+    "pytest",
+]
+
 [tool.uv.sources]
 lynxkite-core = { path = "../lynxkite-core" }
 

lynxkite-app/src/lynxkite/app/crdt.py

@@ -29,7 +29,11 @@ def ws_exception_handler(exception, log):
 
 
 class WebsocketServer(pycrdt_websocket.WebsocketServer):
-    async def init_room(self, name):
+    async def init_room(self, name: str) -> pycrdt_websocket.YRoom:
+        """Initialize a room for the workspace with the given name.
+
+        The workspace is loaded from "crdt_data" if it exists there, or from "data", or a new workspace is created.
+        """
         path = CRDT_PATH / f"{name}.crdt"
         assert path.is_relative_to(CRDT_PATH)
         ystore = pycrdt_websocket.ystore.FileYStore(path)
@@ -49,6 +53,8 @@ class WebsocketServer(pycrdt_websocket.WebsocketServer):
             ws["edges"] = pycrdt.Array()
         if "env" not in ws:
             ws["env"] = "unset"
+            # We have two possible sources of truth for the workspaces, the YStore and the JSON files.
+            # In case we didn't find the workspace in the YStore, we try to load it from the JSON files.
             try_to_load_workspace(ws, name)
         room = pycrdt_websocket.YRoom(
             ystore=ystore, ydoc=ydoc, exception_handler=ws_exception_handler
@@ -62,6 +68,12 @@ class WebsocketServer(pycrdt_websocket.WebsocketServer):
         return room
 
     async def get_room(self, name: str) -> pycrdt_websocket.YRoom:
+        """Get a room by name.
+
+        This method overrides the parent get_room method. The original creates an empty room,
+        with no associated Ydoc. Instead, we want to initialize the the room with a Workspace
+        object.
+        """
         if name not in self.rooms:
             self.rooms[name] = await self.init_room(name)
         room = self.rooms[name]
@@ -83,21 +95,43 @@ def clean_input(ws_pyd):
                 delattr(node, key)
 
 
-def crdt_update(crdt_obj, python_obj, boxes=set()):
+def crdt_update(
+    crdt_obj: pycrdt.Map | pycrdt.Array,
+    python_obj: dict | list,
+    non_collaborative_fields: set[str] = set(),
+):
+    """Update a CRDT object to match a Python object.
+
+    The types between the CRDT object and the Python object must match. If the Python object
+    is a dict, the CRDT object must be a Map. If the Python object is a list, the CRDT object
+    must be an Array.
+
+    Args:
+        crdt_obj: The CRDT object, that will be updated to match the Python object.
+        python_obj: The Python object to update with.
+        non_collaborative_fields: List of fields to treat as a black box. Black boxes are
+        updated as a whole, instead of having a fine-grained data structure to edit
+        collaboratively. Useful for complex fields that contain auto-generated data or
+        metadata.
+        The default is an empty set.
+
+    Raises:
+        ValueError: If the Python object provided is not a dict or list.
+    """
     if isinstance(python_obj, dict):
         for key, value in python_obj.items():
-            if key in boxes:
+            if key in non_collaborative_fields:
                 crdt_obj[key] = value
             elif isinstance(value, dict):
                 if crdt_obj.get(key) is None:
                     crdt_obj[key] = pycrdt.Map()
-                crdt_update(crdt_obj[key], value, boxes)
+                crdt_update(crdt_obj[key], value, non_collaborative_fields)
             elif isinstance(value, list):
                 if crdt_obj.get(key) is None:
                     crdt_obj[key] = pycrdt.Array()
-                crdt_update(crdt_obj[key], value, boxes)
+                crdt_update(crdt_obj[key], value, non_collaborative_fields)
             elif isinstance(value, enum.Enum):
-                crdt_obj[key] = str(value)
+                crdt_obj[key] = str(value.value)
             else:
                 crdt_obj[key] = value
     elif isinstance(python_obj, list):
@@ -105,12 +139,14 @@ def crdt_update(crdt_obj, python_obj, boxes=set()):
             if isinstance(value, dict):
                 if i >= len(crdt_obj):
                     crdt_obj.append(pycrdt.Map())
-                crdt_update(crdt_obj[i], value, boxes)
+                crdt_update(crdt_obj[i], value, non_collaborative_fields)
             elif isinstance(value, list):
                 if i >= len(crdt_obj):
                     crdt_obj.append(pycrdt.Array())
-                crdt_update(crdt_obj[i], value, boxes)
+                crdt_update(crdt_obj[i], value, non_collaborative_fields)
             else:
+                if isinstance(value, enum.Enum):
+                    value = str(value.value)
                 if i >= len(crdt_obj):
                     crdt_obj.append(value)
                 else:
@@ -119,18 +155,34 @@ def crdt_update(crdt_obj, python_obj, boxes=set()):
         raise ValueError("Invalid type:", python_obj)
 
 
-def try_to_load_workspace(ws, name):
+def try_to_load_workspace(ws: pycrdt.Map, name: str):
+    """Load the workspace `name`, if it exists, and update the `ws` CRDT object to match its contents.
+
+    Args:
+        ws: CRDT object to udpate with the workspace contents.
+        name: Name of the workspace to load.
+    """
     json_path = f"data/{name}"
     if os.path.exists(json_path):
         ws_pyd = workspace.load(json_path)
-        crdt_update(ws, ws_pyd.model_dump(), boxes={"display"})
+        # We treat the display field as a black box, since it is a large
+        # dictionary that is meant to change as a whole.
+        crdt_update(ws, ws_pyd.model_dump(), non_collaborative_fields={"display"})
 
 
 last_known_versions = {}
 delayed_executions = {}
 
 
-async def workspace_changed(name, changes, ws_crdt):
+async def workspace_changed(name: str, changes: pycrdt.MapEvent, ws_crdt: pycrdt.Map):
+    """Callback to react to changes in the workspace.
+
+
+    Args:
+        name: Name of the workspace.
+        changes: Changes performed to the workspace.
+        ws_crdt: CRDT object representing the workspace.
+    """
     ws_pyd = workspace.Workspace.model_validate(ws_crdt.to_py())
     # Do not trigger execution for superficial changes.
     # This is a quick solution until we build proper caching.
@@ -154,22 +206,35 @@ async def workspace_changed(name, changes, ws_crdt):
         await execute(name, ws_crdt, ws_pyd)
 
 
-async def execute(name, ws_crdt, ws_pyd, delay=0):
+async def execute(
+    name: str, ws_crdt: pycrdt.Map, ws_pyd: workspace.Workspace, delay: int = 0
+):
+    """Execute the workspace and update the CRDT object with the results.
+
+    Args:
+        name: Name of the workspace.
+        ws_crdt: CRDT object representing the workspace.
+        ws_pyd: Workspace object to execute.
+        delay: Wait time before executing the workspace. The default is 0.
+    """
     if delay:
         try:
             await asyncio.sleep(delay)
         except asyncio.CancelledError:
             return
     path = DATA_PATH / name
-    assert path.is_relative_to(DATA_PATH)
+    assert path.is_relative_to(DATA_PATH),"Provided workspace path is invalid"
+    # Save user changes before executing, in case the execution fails.
     workspace.save(ws_pyd, path)
     await workspace.execute(ws_pyd)
     workspace.save(ws_pyd, path)
+    # Execution happened on the Python object, we need to replicate
+    # the results to the CRDT object.
     with ws_crdt.doc.transaction():
         for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):
             if "data" not in nc:
                 nc["data"] = pycrdt.Map()
-            # Display is added as an opaque Box.
+            # Display is added as a non collaborative field.
             nc["data"]["display"] = np.data.display
             nc["data"]["error"] = np.data.error
 

lynxkite-app/tests/test_crdt.py

@@ -0,0 +1,72 @@
+from enum import Enum
+import pycrdt
+import pytest
+from lynxkite.app.crdt import crdt_update
+
+
+@pytest.fixture
+def empty_dict_workspace():
+    ydoc = pycrdt.Doc()
+    ydoc["workspace"] = ws = pycrdt.Map()
+    yield ws
+
+
+@pytest.fixture
+def empty_list_workspace():
+    ydoc = pycrdt.Doc()
+    ydoc["workspace"] = ws = pycrdt.Array()
+    yield ws
+
+
+class MyEnum(Enum):
+    VALUE = 1
+
+
+@pytest.mark.parametrize(
+    "python_obj,expected",
+    [
+        (
+            {
+                "key1": "value1",
+                "key2": {
+                    "nested_key1": "nested_value1",
+                    "nested_key2": ["nested_value2"],
+                    "nested_key3": MyEnum.VALUE,
+                },
+            },
+            {
+                "key1": "value1",
+                "key2": {
+                    "nested_key1": "nested_value1",
+                    "nested_key2": ["nested_value2"],
+                    "nested_key3": "1",
+                },
+            },
+        )
+    ],
+)
+def test_crdt_update_with_dict(empty_dict_workspace, python_obj, expected):
+    crdt_update(empty_dict_workspace, python_obj)
+    assert empty_dict_workspace.to_py() == expected
+
+
+@pytest.mark.parametrize(
+    "python_obj,expected",
+    [
+        (
+            [
+                "value1",
+                {"nested_key1": "nested_value1", "nested_key2": ["nested_value2"]},
+                MyEnum.VALUE,
+            ],
+            [
+                "value1",
+                {"nested_key1": "nested_value1", "nested_key2": ["nested_value2"]},
+                "1",
+            ],
+        ),
+    ],
+)
+def test_crdt_update_with_list(empty_list_workspace, python_obj, expected):
+    crdt_update(empty_list_workspace, python_obj)
+    assert empty_list_workspace.to_py() == expected

lynxkite-app/tests/test_main.py

@@ -0,0 +1,77 @@
+import uuid
+from fastapi.testclient import TestClient
+from lynxkite.app.main import app, detect_plugins, DATA_PATH
+import os
+
+
+client = TestClient(app)
+
+
+def test_detect_plugins_with_plugins():
+    # This test assumes that these plugins are installed as part of the testing process.
+    plugins = detect_plugins()
+    assert all(
+        plugin in plugins.keys()
+        for plugin in [
+            "lynxkite_plugins.graph_analytics",
+            "lynxkite_plugins.lynxscribe",
+            "lynxkite_plugins.pillow_example",
+        ]
+    )
+
+
+def test_get_catalog():
+    response = client.get("/api/catalog")
+    assert response.status_code == 200
+
+
+def test_save_and_load():
+    save_request = {
+        "path": "test",
+        "ws": {
+            "env": "test",
+            "nodes": [
+                {
+                    "id": "Node_1",
+                    "type": "basic",
+                    "data": {
+                        "display": None,
+                        "error": "Unknown operation.",
+                        "title": "Test node",
+                        "params": {"param1": "value"},
+                    },
+                    "position": {"x": -493.5496596237119, "y": 20.90123252513356},
+                }
+            ],
+            "edges": [],
+        },
+    }
+    response = client.post("/api/save", json=save_request)
+    saved_ws = response.json()
+    assert response.status_code == 200
+    response = client.get("/api/load?path=test")
+    assert response.status_code == 200
+    assert saved_ws == response.json()
+
+
+def test_list_dir():
+    test_dir = str(uuid.uuid4())
+    test_dir_full_path = DATA_PATH / test_dir
+    test_dir_full_path.mkdir(exist_ok=True)
+    test_file = test_dir_full_path / "test_file.txt"
+    test_file.touch()
+    response = client.get(f"/api/dir/list?path={str(test_dir)}")
+    assert response.status_code == 200
+    assert len(response.json()) == 1
+    assert response.json()[0]["name"] == f"{test_dir}/test_file.txt"
+    assert response.json()[0]["type"] == "workspace"
+    test_file.unlink()
+    test_dir_full_path.rmdir()
+
+
+def test_make_dir():
+    dir_name = str(uuid.uuid4())
+    response = client.post("/api/dir/mkdir", json={"path": dir_name})
+    assert response.status_code == 200
+    assert os.path.exists(DATA_PATH / dir_name)
+    os.rmdir(DATA_PATH / dir_name)

lynxkite-core/pyproject.toml

@@ -6,3 +6,8 @@ readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
 ]
+
+[project.optional-dependencies]
+dev = [
+    "pytest",
+]

lynxkite-core/src/lynxkite/core/executors/one_by_one.py

@@ -51,6 +51,8 @@ def get_stages(ws, catalog):
     nodes = {n.id: n for n in ws.nodes}
     batch_inputs = {}
     inputs = {}
+    # For each edge in the workspacce, we record the inputs (sources)
+    # required for each node (target).
     for edge in ws.edges:
         inputs.setdefault(edge.target, []).append(edge.source)
         node = nodes[edge.target]
@@ -93,7 +95,7 @@ async def await_if_needed(obj):
     return obj
 
 
-async def execute(ws, catalog, cache=None):
+async def execute(ws: workspace.Workspace, catalog, cache=None):
     nodes = {n.id: n for n in ws.nodes}
     contexts = {n.id: Context(node=n) for n in ws.nodes}
     edges = {n.id: [] for n in ws.nodes}

lynxkite-core/src/lynxkite/core/ops.py

@@ -64,6 +64,7 @@ class Parameter(BaseConfig):
 class Input(BaseConfig):
     name: str
     type: Type
+    # TODO: Make position an enum with the possible values.
     position: str = "left"
 
 
@@ -90,6 +91,7 @@ class Op(BaseConfig):
     params: dict[str, Parameter]
     inputs: dict[str, Input]
     outputs: dict[str, Output]
+    # TODO: Make type an enum with the possible values.
     type: str = "basic"  # The UI to use for this operation.
 
     def __call__(self, *inputs, **params):
@@ -209,16 +211,3 @@ def op_registration(env: str):
 
 def passive_op_registration(env: str):
     return functools.partial(register_passive_op, env)
-
-
-def register_area(env, name, params=[]):
-    """A node that represents an area. It can contain other nodes, but does not restrict movement in any way."""
-    op = Op(
-        func=no_op,
-        name=name,
-        params={p.name: p for p in params},
-        inputs={},
-        outputs={},
-        type="area",
-    )
-    CATALOGS[env][name] = op

lynxkite-core/src/lynxkite/core/workspace.py

@@ -29,6 +29,8 @@ class WorkspaceNodeData(BaseConfig):
 
 
 class WorkspaceNode(BaseConfig):
+    # The naming of these attributes matches the ones for the NodeBase type in React flow
+    # modyfing them will break the frontend.
     id: str
     type: str
     data: WorkspaceNodeData
@@ -44,6 +46,13 @@ class WorkspaceEdge(BaseConfig):
 
 
 class Workspace(BaseConfig):
+    """A workspace is a representation of a computational graph that consists of nodes and edges.
+
+    Each node represents an operation or task, and the edges represent the flow of data between
+    the nodes. Each workspace is associated with an environment, which determines the operations
+    that can be performed in the workspace and the execution method for the operations.
+    """
+
     env: str = ""
     nodes: list[WorkspaceNode] = dataclasses.field(default_factory=list)
     edges: list[WorkspaceEdge] = dataclasses.field(default_factory=list)
@@ -55,6 +64,7 @@ async def execute(ws: Workspace):
 
 
 def save(ws: Workspace, path: str):
+    """Persist a workspace to a local file in JSON format."""
     j = ws.model_dump_json(indent=2)
     dirname, basename = os.path.split(path)
     # Create temp file in the same directory to make sure it's on the same filesystem.
@@ -66,7 +76,17 @@ def save(ws: Workspace, path: str):
     os.replace(temp_name, path)
 
 
-def load(path: str):
+def load(path: str) -> Workspace:
+    """Load a workspace from a file.
+
+    After loading the workspace, the metadata of the workspace is updated.
+
+    Args:
+        path (str): The path to the file to load the workspace from.
+
+    Returns:
+        Workspace: The loaded workspace object, with updated metadata.
+    """
     with open(path) as f:
         j = f.read()
     ws = Workspace.model_validate_json(j)
@@ -75,13 +95,26 @@ def load(path: str):
     return ws
 
 
-def _update_metadata(ws):
+def _update_metadata(ws: Workspace) -> Workspace:
+    """Update the metadata of the given workspace object.
+
+    The metadata is the information about the operations that the nodes in the workspace represent,
+    like the parameters and their possible values.
+    This information comes from the catalog of operations for the environment of the workspace.
+
+    Args:
+        ws: The workspace object to update.
+
+    Returns:
+        Workspace: The updated workspace object.
+    """
     catalog = ops.CATALOGS.get(ws.env, {})
     nodes = {node.id: node for node in ws.nodes}
     done = set()
     while len(done) < len(nodes):
         for node in ws.nodes:
             if node.id in done:
+                # TODO: Can nodes with the same ID reference different operations?
                 continue
             data = node.data
             op = catalog.get(data.title)

lynxkite-core/tests/test_ops.py

@@ -0,0 +1,85 @@
+import inspect
+from lynxkite.core import ops
+import enum
+
+
+def test_op_decorator_no_params_no_types_default_positions():
+    @ops.op(env="test", name="add", view="basic", outputs=["result"])
+    def add(a, b):
+        return a + b
+
+    assert add.__op__.name == "add"
+    assert add.__op__.params == {}
+    assert add.__op__.inputs == {
+        "a": ops.Input(name="a", type=inspect._empty, position="left"),
+        "b": ops.Input(name="b", type=inspect._empty, position="left"),
+    }
+    assert add.__op__.outputs == {
+        "result": ops.Output(name="result", type=None, position="right")
+    }
+    assert add.__op__.type == "basic"
+    assert ops.CATALOGS["test"]["add"] == add.__op__
+
+
+def test_op_decorator_custom_positions():
+    @ops.input_position(a="right", b="top")
+    @ops.output_position(result="bottom")
+    @ops.op(env="test", name="add", view="basic", outputs=["result"])
+    def add(a, b):
+        return a + b
+
+    assert add.__op__.name == "add"
+    assert add.__op__.params == {}
+    assert add.__op__.inputs == {
+        "a": ops.Input(name="a", type=inspect._empty, position="right"),
+        "b": ops.Input(name="b", type=inspect._empty, position="top"),
+    }
+    assert add.__op__.outputs == {
+        "result": ops.Output(name="result", type=None, position="bottom")
+    }
+    assert add.__op__.type == "basic"
+    assert ops.CATALOGS["test"]["add"] == add.__op__
+
+
+def test_op_decorator_with_params_and_types_():
+    @ops.op(env="test", name="multiply", view="basic", outputs=["result"])
+    def multiply(a: int, b: float = 2.0, *, param: str = "param"):
+        return a * b
+
+    assert multiply.__op__.name == "multiply"
+    assert multiply.__op__.params == {
+        "param": ops.Parameter(name="param", default="param", type=str)
+    }
+    assert multiply.__op__.inputs == {
+        "a": ops.Input(name="a", type=int, position="left"),
+        "b": ops.Input(name="b", type=float, position="left"),
+    }
+    assert multiply.__op__.outputs == {
+        "result": ops.Output(name="result", type=None, position="right")
+    }
+    assert multiply.__op__.type == "basic"
+    assert ops.CATALOGS["test"]["multiply"] == multiply.__op__
+
+
+def test_op_decorator_with_complex_types():
+    class Color(enum.Enum):
+        RED = 1
+        GREEN = 2
+        BLUE = 3
+
+    @ops.op(env="test", name="color_op", view="basic", outputs=["result"])
+    def complex_op(color: Color, color_list: list[Color], color_dict: dict[str, Color]):
+        return color.name
+
+    assert complex_op.__op__.name == "color_op"
+    assert complex_op.__op__.params == {}
+    assert complex_op.__op__.inputs == {
+        "color": ops.Input(name="color", type=Color, position="left"),
+        "color_list": ops.Input(name="color_list", type=list[Color], position="left"),
+        "color_dict": ops.Input(name="color_dict", type=dict[str, Color], position="left"),
+    }
+    assert complex_op.__op__.type == "basic"
+    assert complex_op.__op__.outputs == {
+        "result": ops.Output(name="result", type=None, position="right")
+    }
+    assert ops.CATALOGS["test"]["color_op"] == complex_op.__op__

lynxkite-core/tests/test_workspace.py

@@ -0,0 +1,101 @@
+import os
+import pytest
+import tempfile
+from lynxkite.core import workspace
+
+
+def test_save_load():
+    ws = workspace.Workspace(env="test")
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="1",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Node 1", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="2",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Node 2", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    ws.edges.append(
+        workspace.WorkspaceEdge(
+            id="edge1",
+            source="1",
+            target="2",
+            sourceHandle="",
+            targetHandle="",
+        )
+    )
+    path = os.path.join(tempfile.gettempdir(), "test_workspace.json")
+
+    try:
+        workspace.save(ws, path)
+        assert os.path.exists(path)
+        loaded_ws = workspace.load(path)
+        assert loaded_ws.env == ws.env
+        assert len(loaded_ws.nodes) == len(ws.nodes)
+        assert len(loaded_ws.edges) == len(ws.edges)
+        sorted_ws_nodes = sorted(ws.nodes, key=lambda x: x.id)
+        sorted_loaded_ws_nodes = sorted(loaded_ws.nodes, key=lambda x: x.id)
+        # We do manual assertion on each attribute because metadata is added at
+        # loading time, which makes the objects different.
+        for node, loaded_node in zip(sorted_ws_nodes, sorted_loaded_ws_nodes):
+            assert node.id == loaded_node.id
+            assert node.type == loaded_node.type
+            assert node.data.title == loaded_node.data.title
+            assert node.data.params == loaded_node.data.params
+            assert node.position.x == loaded_node.position.x
+            assert node.position.y == loaded_node.position.y
+        sorted_ws_edges = sorted(ws.edges, key=lambda x: x.id)
+        sorted_loaded_ws_edges = sorted(loaded_ws.edges, key=lambda x: x.id)
+        for edge, loaded_edge in zip(sorted_ws_edges, sorted_loaded_ws_edges):
+            assert edge.id == loaded_edge.id
+            assert edge.source == loaded_edge.source
+            assert edge.target == loaded_edge.target
+            assert edge.sourceHandle == loaded_edge.sourceHandle
+            assert edge.targetHandle == loaded_edge.targetHandle
+    finally:
+        os.remove(path)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def populate_ops_catalog():
+    from lynxkite.core import ops
+
+    ops.register_passive_op("test", "Test Operation", [])
+
+
+def test_update_metadata():
+    ws = workspace.Workspace(env="test")
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="1",
+            type="basic",
+            data=workspace.WorkspaceNodeData(title="Test Operation", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="2",
+            type="basic",
+            data=workspace.WorkspaceNodeData(title="Unknown Operation", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    updated_ws = workspace._update_metadata(ws)
+    assert updated_ws.nodes[0].data.meta.name == "Test Operation"
+    assert updated_ws.nodes[0].data.error is None
+    assert not hasattr(updated_ws.nodes[1].data, "meta")
+    assert updated_ws.nodes[1].data.error == "Unknown operation."
+
+
+def test_update_metadata_with_empty_workspace():
+    ws = workspace.Workspace(env="test")
+    updated_ws = workspace._update_metadata(ws)
+    assert len(updated_ws.nodes) == 0

lynxkite-graph-analytics/pyproject.toml

@@ -16,9 +16,16 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+dev = [
+    "pytest",
+    "pytest-asyncio",
+]
 gpu = [
     "nx-cugraph-cu12>=24.12.0",
 ]
 
 [tool.uv.sources]
 lynxkite-core = { path = "../lynxkite-core" }
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"

lynxkite-graph-analytics/src/lynxkite_plugins/graph_analytics/lynxkite_ops.py

@@ -1,7 +1,7 @@
 """Graph analytics operations. To be split into separate files when we have more."""
 
 import os
-from lynxkite.core import ops
+from lynxkite.core import ops, workspace
 from collections import deque
 import dataclasses
 import functools
@@ -124,7 +124,7 @@ def disambiguate_edges(ws):
 
 @ops.register_executor(ENV)
 async def execute(ws):
-    catalog = ops.CATALOGS[ENV]
+    catalog: dict[str, ops.Op] = ops.CATALOGS[ENV]
     disambiguate_edges(ws)
     outputs = {}
     failed = 0
@@ -135,12 +135,17 @@ async def execute(ws):
             # TODO: Take the input/output handles into account.
             inputs = [edge.source for edge in ws.edges if edge.target == node.id]
             if all(input in outputs for input in inputs):
+                # All inputs for this node are ready, we can compute the output.
                 inputs = [outputs[input] for input in inputs]
                 data = node.data
-                op = catalog[data.title]
                 params = {**data.params}
-                # Convert inputs.
+                op = catalog.get(data.title)
+                if not op:
+                    data.error = "Operation not found in catalog"
+                    failed += 1
+                    continue
                 try:
+                    # Convert inputs types  to match operation signature.
                     for i, (x, p) in enumerate(zip(inputs, op.inputs.values())):
                         if p.type == nx.Graph and isinstance(x, Bundle):
                             inputs[i] = x.to_nx()
@@ -222,6 +227,7 @@ def create_scale_free_graph(*, nodes: int = 10):
 @op("Compute PageRank")
 @nx_node_attribute_func("pagerank")
 def compute_pagerank(graph: nx.Graph, *, damping=0.85, iterations=100):
+    # TODO: This requires scipy to be installed.
     return nx.pagerank(graph, alpha=damping, max_iter=iterations)
 
 

lynxkite-graph-analytics/src/lynxkite_plugins/graph_analytics/pytorch_model_ops.py

@@ -72,4 +72,4 @@ ops.register_passive_op(
     inputs=[ops.Input(name="input", position="top", type="tensor")],
     outputs=[ops.Output(name="output", position="bottom", type="tensor")],
     params=[ops.Parameter.basic("times", 1, int)],
-)
+)

lynxkite-graph-analytics/tests/test_lynxkite_ops.py

@@ -0,0 +1,94 @@
+import pandas as pd
+import pytest
+import networkx as nx
+
+from lynxkite.core import workspace
+from lynxkite_plugins.graph_analytics.lynxkite_ops import Bundle, execute, op
+
+
+async def test_execute_operation_not_in_catalog():
+    ws = workspace.Workspace(env="test")
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="1",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Non existing op", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    await execute(ws)
+    assert ws.nodes[0].data.error == "Operation not found in catalog"
+
+
+async def test_execute_operation_inputs_correct_cast():
+    # Test that the automatic casting of operation inputs works correctly.
+
+    @op("Create Bundle")
+    def create_bundle() -> Bundle:
+        df = pd.DataFrame({"source": [1, 2, 3], "target": [4, 5, 6]})
+        return Bundle(dfs={"edges": df})
+
+    @op("Bundle to Graph")
+    def bundle_to_graph(graph: nx.Graph) -> nx.Graph:
+        return graph
+
+    @op("Graph to Bundle")
+    def graph_to_bundle(bundle: Bundle) -> pd.DataFrame:
+        return list(bundle.dfs.values())[0]
+
+    @op("Dataframe to Bundle")
+    def dataframe_to_bundle(bundle: Bundle) -> Bundle:
+        return bundle
+
+    ws = workspace.Workspace(env="test")
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="1",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Create Bundle", params={}),
+            position=workspace.Position(x=0, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="2",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Bundle to Graph", params={}),
+            position=workspace.Position(x=100, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="3",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Graph to Bundle", params={}),
+            position=workspace.Position(x=200, y=0),
+        )
+    )
+    ws.nodes.append(
+        workspace.WorkspaceNode(
+            id="4",
+            type="node_type",
+            data=workspace.WorkspaceNodeData(title="Dataframe to Bundle", params={}),
+            position=workspace.Position(x=300, y=0),
+        )
+    )
+    ws.edges = [
+        workspace.WorkspaceEdge(
+            id="1", source="1", target="2", sourceHandle="1", targetHandle="2"
+        ),
+        workspace.WorkspaceEdge(
+            id="2", source="2", target="3", sourceHandle="2", targetHandle="3"
+        ),
+        workspace.WorkspaceEdge(
+            id="3", source="3", target="4", sourceHandle="3", targetHandle="4"
+        ),
+    ]
+
+    await execute(ws)
+
+    assert all([node.data.error is None for node in ws.nodes])
+
+
+if __name__ == "__main__":
+    pytest.main()