Stricter checks to create/update an Ethernet switch and add tests

gns3server/api/routes/compute/ethernet_switch_nodes.py

@@ -150,12 +150,22 @@ def suspend_ethernet_switch(node: EthernetSwitch = Depends(dep_node)) -> None:
     pass
 
 
+@router.post("/{node_id}/reload", status_code=status.HTTP_204_NO_CONTENT)
+def reload_ethernet_switch(node: EthernetSwitch = Depends(dep_node)) -> None:
+    """
+    Reload an Ethernet switch.
+    This endpoint results in no action since Ethernet switch nodes are always on.
+    """
+
+    pass
+
+
 @router.post(
     "/{node_id}/adapters/{adapter_number}/ports/{port_number}/nio",
     status_code=status.HTTP_201_CREATED,
     response_model=schemas.UDPNIO,
 )
-async def create_nio(
+async def create_ethernet_switch_nio(
         *,
         adapter_number: int = Path(..., ge=0, le=0),
         port_number: int,
@@ -169,7 +179,7 @@ async def create_nio(
 
 
 @router.delete("/{node_id}/adapters/{adapter_number}/ports/{port_number}/nio", status_code=status.HTTP_204_NO_CONTENT)
-async def delete_nio(
+async def delete_ethernet_switch_nio(
         *,
         adapter_number: int = Path(..., ge=0, le=0),
         port_number: int,
@@ -185,7 +195,7 @@ async def delete_nio(
 
 
 @router.post("/{node_id}/adapters/{adapter_number}/ports/{port_number}/capture/start")
-async def start_capture(
+async def start_ethernet_switch_capture(
         *,
         adapter_number: int = Path(..., ge=0, le=0),
         port_number: int,
@@ -205,7 +215,7 @@ async def start_capture(
 @router.post(
     "/{node_id}/adapters/{adapter_number}/ports/{port_number}/capture/stop", status_code=status.HTTP_204_NO_CONTENT
 )
-async def stop_capture(
+async def stop_ethernet_switch_capture(
         *,
         adapter_number: int = Path(..., ge=0, le=0),
         port_number: int,

gns3server/compute/dynamips/nodes/ethernet_switch.py

@@ -166,7 +166,7 @@ class EthernetSwitch(Device):
         """
         if ports != self._ports:
             if len(self._nios) > 0 and len(ports) != len(self._ports):
-                raise NodeError("Can't modify a switch already connected.")
+                raise NodeError("Cannot change ports on a switch that is already connected.")
 
             port_number = 0
             for port in ports:
@@ -356,7 +356,7 @@ class EthernetSwitch(Device):
         elif settings["type"] == "dot1q":
             await self.set_dot1q_port(port_number, settings["vlan"])
         elif settings["type"] == "qinq":
-            await self.set_qinq_port(port_number, settings["vlan"], settings.get("ethertype"))
+            await self.set_qinq_port(port_number, settings["vlan"], settings.get("ethertype", "0x8100"))
 
     async def set_access_port(self, port_number, vlan_id):
         """
@@ -427,7 +427,7 @@ class EthernetSwitch(Device):
         await self._hypervisor.send(
             'ethsw set_qinq_port "{name}" {nio} {outer_vlan} {ethertype}'.format(
                 name=self._name, nio=nio, outer_vlan=outer_vlan, ethertype=ethertype if ethertype != "0x8100" else ""
-            )
+            ).strip()
         )
 
         log.info(

gns3server/schemas/compute/ethernet_switch_nodes.py

@@ -14,7 +14,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, validator
 from typing import Optional, List
 from uuid import UUID
 from enum import Enum
@@ -42,9 +42,17 @@ class EthernetSwitchPort(BaseModel):
     name: str
     port_number: int
     type: EthernetSwitchPortType = Field(..., description="Port type")
-    vlan: Optional[int] = Field(None, ge=1, description="VLAN number")
+    vlan: int = Field(..., ge=1, le=4094, description="VLAN number")
     ethertype: Optional[EthernetSwitchEtherType] = Field(None, description="QinQ Ethertype")
 
+    @validator("ethertype")
+    def validate_ethertype(cls, v, values):
+
+        if v is not None:
+            if "type" not in values or values["type"] != EthernetSwitchPortType.qinq:
+                raise ValueError("Ethertype is only for QinQ port type")
+        return v
+
 
 class TelnetConsoleType(str, Enum):
     """

tests/api/routes/compute/test_ethernet_switch_nodes.py

@@ -0,0 +1,454 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+import pytest_asyncio
+
+from fastapi import FastAPI, status
+from httpx import AsyncClient
+from tests.utils import asyncio_patch, AsyncioMagicMock
+from unittest.mock import call
+
+from gns3server.compute.project import Project
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest_asyncio.fixture
+async def ethernet_switch(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> dict:
+
+    params = {"name": "Ethernet Switch"}
+    with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.create") as mock:
+        response = await compute_client.post(
+            app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
+            json=params
+        )
+        assert mock.called
+        assert response.status_code == status.HTTP_201_CREATED
+
+    json_response = response.json()
+    node = compute_project.get_node(json_response["node_id"])
+    node._hypervisor = AsyncioMagicMock()
+    node._hypervisor.send = AsyncioMagicMock()
+    node._hypervisor.version = "0.2.16"
+    return json_response
+
+
+async def test_ethernet_switch_create(app: FastAPI, compute_client: AsyncClient, compute_project: Project) -> None:
+
+    params = {"name": "Ethernet Switch 1"}
+    with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.create") as mock:
+        response = await compute_client.post(
+            app.url_path_for("compute:create_ethernet_switch", project_id=compute_project.id),
+            json=params
+        )
+        assert mock.called
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["name"] == "Ethernet Switch 1"
+        assert response.json()["project_id"] == compute_project.id
+
+
+async def test_ethernet_switch_get(app: FastAPI, compute_client: AsyncClient, compute_project: Project, ethernet_switch: dict) -> None:
+
+    response = await compute_client.get(
+        app.url_path_for(
+            "compute:get_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"]
+        )
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["name"] == "Ethernet Switch"
+    assert response.json()["project_id"] == compute_project.id
+    assert response.json()["status"] == "started"
+
+
+async def test_ethernet_switch_duplicate(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        ethernet_switch: dict
+) -> None:
+
+    # create destination switch first
+    params = {"name": "Ethernet Switch 2"}
+    with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.create") as mock:
+        response = await compute_client.post(
+            app.url_path_for(
+                "compute:create_ethernet_switch",
+                project_id=compute_project.id),
+            json=params
+        )
+        assert mock.called
+        assert response.status_code == status.HTTP_201_CREATED
+
+    params = {"destination_node_id": response.json()["node_id"]}
+    response = await compute_client.post(
+        app.url_path_for(
+            "compute:duplicate_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"]), json=params
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+
+
+async def test_ethernet_switch_update(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        ethernet_switch: dict
+) -> None:
+
+    params = {
+        "name": "test",
+        "console_type": "telnet"
+    }
+
+    response = await compute_client.put(
+        app.url_path_for(
+            "compute:update_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"]),
+        json=params
+    )
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["name"] == "test"
+    node = compute_project.get_node(ethernet_switch["node_id"])
+    node._hypervisor.send.assert_called_with("ethsw rename \"Ethernet Switch\" \"test\"")
+
+
+async def test_ethernet_switch_update_ports(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        ethernet_switch: dict
+) -> None:
+
+    port_params = {
+        "ports_mapping": [
+            {
+                "name": "Ethernet0",
+                "port_number": 0,
+                "type": "qinq",
+                "vlan": 1
+            },
+            {
+                "name": "Ethernet1",
+                "port_number": 1,
+                "type": "qinq",
+                "vlan": 2,
+                "ethertype": "0x88A8"
+            },
+            {
+                "name": "Ethernet2",
+                "port_number": 2,
+                "type": "dot1q",
+                "vlan": 3,
+            },
+            {
+                "name": "Ethernet3",
+                "port_number": 3,
+                "type": "access",
+                "vlan": 4,
+            }
+        ],
+    }
+
+    response = await compute_client.put(
+        app.url_path_for(
+            "compute:update_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"]),
+        json=port_params
+    )
+    assert response.status_code == status.HTTP_200_OK
+
+    nio_params = {
+        "type": "nio_udp",
+        "lport": 4242,
+        "rport": 4343,
+        "rhost": "127.0.0.1"
+    }
+
+    for port_mapping in port_params["ports_mapping"]:
+        port_number = port_mapping["port_number"]
+        vlan = port_mapping["vlan"]
+        port_type = port_mapping["type"]
+        ethertype = port_mapping.get("ethertype", "")
+        url = app.url_path_for(
+            "compute:create_ethernet_switch_nio",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"],
+            adapter_number="0",
+            port_number=f"{port_number}"
+        )
+        await compute_client.post(url, json=nio_params)
+
+        node = compute_project.get_node(ethernet_switch["node_id"])
+        nio = node.get_nio(port_number)
+        calls = [
+            call.send(f'nio create_udp {nio.name} 4242 127.0.0.1 4343'),
+            call.send(f'ethsw add_nio "Ethernet Switch" {nio.name}'),
+            call.send(f'ethsw set_{port_type}_port "Ethernet Switch" {nio.name} {vlan} {ethertype}'.strip())
+        ]
+        node._hypervisor.send.assert_has_calls(calls)
+        node._hypervisor.send.reset_mock()
+
+
+@pytest.mark.parametrize(
+    "ports_settings",
+    (
+            (
+                    {
+                        "name": "Ethernet0",
+                        "port_number": 0,
+                        "type": "dot42q",  # invalid port type
+                        "vlan": 1,
+                    }
+            ),
+            (
+                    {
+                        "name": "Ethernet0",
+                        "port_number": 0,
+                        "type": "access",  # missing vlan field
+                    }
+            ),
+            (
+                    {
+                        "name": "Ethernet0",
+                        "port_number": 0,
+                        "type": "dot1q",
+                        "vlan": 1,
+                        "ethertype": "0x88A8"  # EtherType is only for QinQ
+                    }
+            ),
+            (
+                    {
+                        "name": "Ethernet0",
+                        "port_number": 0,
+                        "type": "qinq",
+                        "vlan": 1,
+                        "ethertype": "0x4242"  # not a valid EtherType
+                    }
+            ),
+            (
+                    {
+                        "name": "Ethernet0",
+                        "port_number": 0,
+                        "type": "access",
+                        "vlan": 0,  # minimum vlan number is 1
+                    }
+            ),
+            (
+                    {
+                        "name": "Ethernet0",
+                        "port_number": 0,
+                        "type": "access",
+                        "vlan": 4242,  # maximum vlan number is 4094
+                    }
+            ),
+    )
+)
+async def test_ethernet_switch_update_ports_invalid(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        ethernet_switch: dict,
+        ports_settings: dict,
+) -> None:
+
+    port_params = {
+        "ports_mapping": [ports_settings]
+    }
+
+    response = await compute_client.put(
+        app.url_path_for(
+            "compute:update_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"]),
+        json=port_params
+    )
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+
+
+async def test_ethernet_switch_delete(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    response = await compute_client.delete(
+        app.url_path_for(
+            "compute:delete_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"]
+        )
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_ethernet_switch_start(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    response = await compute_client.post(
+        app.url_path_for(
+            "compute:start_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"])
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_ethernet_switch_stop(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    response = await compute_client.post(
+        app.url_path_for(
+            "compute:stop_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"])
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_ethernet_switch_suspend(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    response = await compute_client.post(
+        app.url_path_for(
+            "compute:suspend_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"])
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_ethernet_switch_reload(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    response = await compute_client.post(
+        app.url_path_for(
+            "compute:reload_ethernet_switch",
+            project_id=ethernet_switch["project_id"],
+            node_id=ethernet_switch["node_id"])
+    )
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+
+async def test_ethernet_switch_create_udp(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        ethernet_switch: dict
+) -> None:
+
+    params = {
+        "type": "nio_udp",
+        "lport": 4242,
+        "rport": 4343,
+        "rhost": "127.0.0.1"
+    }
+
+    url = app.url_path_for(
+        "compute:create_ethernet_switch_nio",
+        project_id=ethernet_switch["project_id"],
+        node_id=ethernet_switch["node_id"],
+        adapter_number="0",
+        port_number="0"
+    )
+    response = await compute_client.post(url, json=params)
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json()["type"] == "nio_udp"
+
+    node = compute_project.get_node(ethernet_switch["node_id"])
+    nio = node.get_nio(0)
+    calls = [
+        call.send(f'nio create_udp {nio.name} 4242 127.0.0.1 4343'),
+        call.send(f'ethsw add_nio "Ethernet Switch" {nio.name}'),
+        call.send(f'ethsw set_access_port "Ethernet Switch" {nio.name} 1')
+    ]
+    node._hypervisor.send.assert_has_calls(calls)
+
+
+async def test_ethernet_switch_delete_nio(
+        app: FastAPI,
+        compute_client: AsyncClient,
+        compute_project: Project,
+        ethernet_switch: dict
+) -> None:
+
+    params = {
+        "type": "nio_udp",
+        "lport": 4242,
+        "rport": 4343,
+        "rhost": "127.0.0.1"
+    }
+
+    url = app.url_path_for(
+        "compute:create_ethernet_switch_nio",
+        project_id=ethernet_switch["project_id"],
+        node_id=ethernet_switch["node_id"],
+        adapter_number="0",
+        port_number="0"
+    )
+    await compute_client.post(url, json=params)
+
+    node = compute_project.get_node(ethernet_switch["node_id"])
+    node._hypervisor.send.reset_mock()
+    nio = node.get_nio(0)
+
+    url = app.url_path_for(
+        "compute:delete_ethernet_switch_nio",
+        project_id=ethernet_switch["project_id"],
+        node_id=ethernet_switch["node_id"],
+        adapter_number="0",
+        port_number="0"
+    )
+    response = await compute_client.delete(url)
+    assert response.status_code == status.HTTP_204_NO_CONTENT
+
+    calls = [
+        call(f'ethsw remove_nio "Ethernet Switch" {nio.name}'),
+        call(f'nio delete {nio.name}')
+    ]
+    node._hypervisor.send.assert_has_calls(calls)
+
+
+async def test_ethernet_switch_start_capture(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    params = {
+        "capture_file_name": "test.pcap",
+        "data_link_type": "DLT_EN10MB"
+    }
+
+    url = app.url_path_for("compute:start_ethernet_switch_capture",
+                           project_id=ethernet_switch["project_id"],
+                           node_id=ethernet_switch["node_id"],
+                           adapter_number="0",
+                           port_number="0")
+
+    with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.start_capture") as mock:
+        response = await compute_client.post(url, json=params)
+        assert response.status_code == status.HTTP_200_OK
+        assert mock.called
+        assert "test.pcap" in response.json()["pcap_file_path"]
+
+
+async def test_ethernet_switch_stop_capture(app: FastAPI, compute_client: AsyncClient, ethernet_switch: dict) -> None:
+
+    url = app.url_path_for("compute:stop_ethernet_switch_capture",
+                           project_id=ethernet_switch["project_id"],
+                           node_id=ethernet_switch["node_id"],
+                           adapter_number="0",
+                           port_number="0")
+
+    with asyncio_patch("gns3server.compute.dynamips.nodes.ethernet_switch.EthernetSwitch.stop_capture") as mock:
+        response = await compute_client.post(url)
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+        assert mock.called