Merge branch '3.1' into update-dependencies

tests/api/routes/compute/test_docker_nodes.py

@@ -211,6 +211,34 @@ class TestDockerNodesRoutes:
         assert response.status_code == status.HTTP_201_CREATED
         assert response.json()["type"] == "nio_udp"
 
+    async def test_docker_nio_batch_create(self, app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
+        """
+        Exercise the project-wide batch NIO endpoint: bind two NIOs on the same
+        docker node in a single request (the path used during project open).
+        """
+
+        params = {
+            "nios": [
+                {
+                    "node_id": vm["node_id"],
+                    "adapter_number": 0,
+                    "port_number": 0,
+                    "nio": {"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"},
+                },
+                {
+                    "node_id": vm["node_id"],
+                    "adapter_number": 1,
+                    "port_number": 0,
+                    "nio": {"type": "nio_udp", "lport": 4243, "rport": 4344, "rhost": "127.0.0.1"},
+                },
+            ]
+        }
+
+        url = app.url_path_for("compute:create_batch_nios", project_id=vm["project_id"])
+        response = await compute_client.post(url, json=params)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["added"] == 2
+
 
     async def test_docker_update_nio(self, app: FastAPI, compute_client: AsyncClient, vm: dict) -> None:
 

tests/api/routes/compute/test_projects.py

@@ -224,3 +224,121 @@ class TestComputeProjectRoutes:
                                                       project_id=project.id,
                                                       file_path=file_path), content=b"world")
         assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+class TestBatchNIOEdgeCases:
+
+    @pytest.mark.asyncio
+    async def test_dynamips_router_dispatch_to_slot_add_nio_binding(self):
+        """_add_nio_binding dispatches Dynamips router to slot_add_nio_binding."""
+        from unittest.mock import AsyncMock, MagicMock
+        from gns3server.api.routes.compute.projects import _add_nio_binding
+
+        node = MagicMock()
+        type(node.manager).__name__ = "Dynamips"
+        node.slot_add_nio_binding = AsyncMock()
+        nio = MagicMock()
+
+        await _add_nio_binding(node, 0, 0, nio)
+        node.slot_add_nio_binding.assert_called_once_with(0, 0, nio)
+
+    @pytest.mark.asyncio
+    async def test_dynamips_switch_dispatch_to_add_nio(self):
+        """_add_nio_binding dispatches Dynamips switch to add_nio."""
+        from unittest.mock import AsyncMock, MagicMock
+        from gns3server.api.routes.compute.projects import _add_nio_binding
+
+        node = MagicMock()
+        type(node.manager).__name__ = "Dynamips"
+        del node.slot_add_nio_binding  # no slot_add_nio → switch path
+        node.add_nio = AsyncMock()
+        nio = MagicMock()
+
+        await _add_nio_binding(node, 0, 0, nio)
+        node.add_nio.assert_called_once_with(nio, 0)
+
+    @pytest.mark.asyncio
+    async def test_dynamips_create_nio_is_async_and_needs_await(self):
+        """
+        Dynamips.create_nio is async (returns a coroutine) unlike the sync
+        base version.  The batch handler must await it.
+        """
+        import inspect
+        import asyncio as _asyncio
+
+        class _FakeDynamips:
+            async def create_nio(self, node, nio_settings):
+                return {"type": "nio_udp", "node": node}
+
+        class _FakeBase:
+            def create_nio(self, nio_settings):
+                return {"type": "nio_udp"}
+
+        dyn = _FakeDynamips()
+        base = _FakeBase()
+        assert len(inspect.signature(dyn.create_nio).parameters) == 2   # Dynamips
+        assert len(inspect.signature(base.create_nio).parameters) == 1  # standard
+        assert inspect.iscoroutinefunction(dyn.create_nio)
+        assert not inspect.iscoroutinefunction(base.create_nio)
+
+        # Verify the batch logic: 2 params → await, 1 param → no await
+        d_result = await dyn.create_nio("r1", {"type": "nio_udp"})
+        b_result = base.create_nio({"type": "nio_udp"})
+        assert d_result["node"] == "r1"
+        assert b_result["type"] == "nio_udp"
+
+    @pytest.mark.asyncio
+    async def test_qemu_dispatch_to_adapter_add_nio_binding(self):
+        """_add_nio_binding dispatches Qemu to adapter_add_nio_binding."""
+        from unittest.mock import AsyncMock, MagicMock
+        from gns3server.api.routes.compute.projects import _add_nio_binding
+
+        node = MagicMock()
+        type(node.manager).__name__ = "Qemu"
+        node.adapter_add_nio_binding = AsyncMock()
+        nio = MagicMock()
+
+        await _add_nio_binding(node, 0, 0, nio)
+        node.adapter_add_nio_binding.assert_called_once_with(0, nio)
+
+    @pytest.mark.asyncio
+    async def test_iou_dispatch_to_adapter_add_nio_binding(self):
+        """_add_nio_binding dispatches IOU to adapter_add_nio_binding(adapter, port, nio)."""
+        from unittest.mock import AsyncMock, MagicMock
+        from gns3server.api.routes.compute.projects import _add_nio_binding
+
+        node = MagicMock()
+        type(node.manager).__name__ = "IOU"
+        node.adapter_add_nio_binding = AsyncMock()
+        nio = MagicMock()
+
+        await _add_nio_binding(node, 1, 2, nio)
+        node.adapter_add_nio_binding.assert_called_once_with(1, 2, nio)
+
+    @pytest.mark.asyncio
+    async def test_vpcs_dispatch_to_port_add_nio_binding(self):
+        """_add_nio_binding dispatches VPCS to port_add_nio_binding."""
+        from unittest.mock import AsyncMock, MagicMock
+        from gns3server.api.routes.compute.projects import _add_nio_binding
+
+        node = MagicMock()
+        type(node.manager).__name__ = "VPCS"
+        node.port_add_nio_binding = AsyncMock()
+        nio = MagicMock()
+
+        await _add_nio_binding(node, 0, 3, nio)
+        node.port_add_nio_binding.assert_called_once_with(3, nio)
+
+    @pytest.mark.asyncio
+    async def test_builtin_dispatch_to_add_nio(self):
+        """_add_nio_binding dispatches Builtin nodes to add_nio(nio, port)."""
+        from unittest.mock import AsyncMock, MagicMock
+        from gns3server.api.routes.compute.projects import _add_nio_binding
+
+        node = MagicMock()
+        type(node.manager).__name__ = "Builtin"
+        node.add_nio = AsyncMock()
+        nio = MagicMock()
+
+        await _add_nio_binding(node, 0, 0, nio)
+        node.add_nio.assert_called_once_with(nio, 0)