diff --git a/gns3server/api/routes/controller/links.py b/gns3server/api/routes/controller/links.py index 8c438de28..67041eee2 100644 --- a/gns3server/api/routes/controller/links.py +++ b/gns3server/api/routes/controller/links.py @@ -24,7 +24,7 @@ import aiohttp from fastapi import APIRouter, Depends, Request, status from fastapi.responses import StreamingResponse from fastapi.encoders import jsonable_encoder -from typing import List +from typing import List, Union from uuid import UUID from gns3server.controller import Controller @@ -289,15 +289,16 @@ async def stream_pcap(request: Request, link: Link = Depends(dep_link)) -> Strea @router.get( "/{link_id}/iface", - response_model=dict[str, schemas.UdpPort | schemas.EthernetPort], + response_model=Union[schemas.UDPPortInfo, schemas.EthernetPortInfo], dependencies=[Depends(has_privilege("Link.Audit"))] ) -async def get_iface(link: Link = Depends(dep_link)) -> dict[str, schemas.UdpPort | schemas.EthernetPort]: +async def get_iface(link: Link = Depends(dep_link)) -> Union[schemas.UDPPortInfo, schemas.EthernetPortInfo]: """ Return iface info for links to Cloud or NAT devices. Required privilege: Link.Audit """ + ifaces_info = {} for node_data in link._nodes: node = node_data["node"] @@ -305,17 +306,12 @@ async def get_iface(link: Link = Depends(dep_link)) -> dict[str, schemas.UdpPort continue port_number = node_data["port_number"] - compute = node.compute project_id = link.project.id - response = await compute.get( - f"/projects/{project_id}/{node.node_type}/nodes/{node.id}" - ) - node_info = response.json - - if "ports_mapping" not in node_info: + response = await compute.get(f"/projects/{project_id}/{node.node_type}/nodes/{node.id}") + if "ports_mapping" not in response.json: continue - ports_mapping = node_info["ports_mapping"] + ports_mapping = response.json["ports_mapping"] for port in ports_mapping: port_num = port.get("port_number") @@ -323,20 +319,20 @@ async def get_iface(link: Link = Depends(dep_link)) -> dict[str, schemas.UdpPort if port_num and int(port_num) == int(port_number): port_type = port.get("type", "") if "udp" in port_type.lower(): - ifaces_info[node.id] = { + ifaces_info = { + "node_id": node.id, "type": f"{port_type}", - "rhost": port["rhost"], "lport": port["lport"], - "rport": port["rport"], + "rhost": port["rhost"], + "rport": port["rport"] } else: - ifaces_info[node.id] = { + ifaces_info = { + "node_id": node.id, "type": f"{port_type}", "interface": port["interface"], } if not ifaces_info: - raise ControllerError( - "Link not connected to Cloud/NAT" - ) + raise ControllerError("Link not connected to Cloud/NAT") return ifaces_info diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index a4fb297d6..a23d61d96 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -20,7 +20,7 @@ from .common import ErrorMessage from .version import Version # Controller schemas -from .controller.links import LinkCreate, LinkUpdate, Link, UdpPort, EthernetPort +from .controller.links import LinkCreate, LinkUpdate, Link, UDPPortInfo, EthernetPortInfo from .controller.computes import ComputeCreate, ComputeUpdate, ComputeVirtualBoxVM, ComputeVMwareVM, ComputeDockerImage, AutoIdlePC, Compute from .controller.templates import TemplateCreate, TemplateUpdate, TemplateUsage, Template from .controller.images import Image, ImageType diff --git a/gns3server/schemas/controller/links.py b/gns3server/schemas/controller/links.py index ac82c0d76..cdcf6047f 100644 --- a/gns3server/schemas/controller/links.py +++ b/gns3server/schemas/controller/links.py @@ -94,21 +94,22 @@ class Link(LinkBase): ) -class UdpPort(BaseModel): +class UDPPortInfo(BaseModel): """ UDP port information. """ + node_id: UUID lport: int rhost: str rport: int type: str - -class EthernetPort(BaseModel): +class EthernetPortInfo(BaseModel): """ Ethernet port information. """ + node_id: UUID interface: str type: str diff --git a/tests/api/routes/controller/test_links.py b/tests/api/routes/controller/test_links.py index a5aea4d5e..5068fc8ea 100644 --- a/tests/api/routes/controller/test_links.py +++ b/tests/api/routes/controller/test_links.py @@ -16,6 +16,7 @@ # along with this program. If not, see . import pytest +import uuid import pytest_asyncio from typing import Tuple @@ -424,16 +425,18 @@ class TestLinkRoutes: assert response.status_code == status.HTTP_200_OK assert response.json() == FILTERS + async def test_get_udp_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None: """ Test getting UDP tunnel interface information from a link. """ + link = Link(project) project._links = {link.id: link} cloud_node = MagicMock() cloud_node.node_type = "cloud" - cloud_node.id = "cloud-node-id" + cloud_node.id = str(uuid.uuid4()) cloud_node.name = "Cloud1" compute = MagicMock() @@ -456,29 +459,28 @@ class TestLinkRoutes: link._nodes = [{"node": cloud_node, "port_number": 1}] response = await client.get(app.url_path_for("get_iface", project_id=project.id, link_id=link.id)) - + assert response.status_code == status.HTTP_200_OK result = response.json() - - assert "cloud-node-id" in result - - udp_info = result["cloud-node-id"] - assert udp_info["lport"] == 20000 - assert udp_info["rhost"] == "127.0.0.1" - assert udp_info["rport"] == 30000 - assert udp_info["type"] == "udp" + assert result["node_id"] == cloud_node.id + assert result["lport"] == 20000 + assert result["rhost"] == "127.0.0.1" + assert result["rport"] == 30000 + assert result["type"] == "udp" + + async def test_get_ethernet_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None: """ Test getting ethernet interface information from a link. """ link = Link(project) project._links = {link.id: link} - + cloud_node = MagicMock() cloud_node.node_type = "cloud" - cloud_node.id = "cloud-node-id" + cloud_node.id = str(uuid.uuid4()) cloud_node.name = "Cloud1" - + compute = MagicMock() response = MagicMock() response.json = { @@ -493,16 +495,13 @@ class TestLinkRoutes: } compute.get = AsyncioMagicMock(return_value=response) cloud_node.compute = compute - + link._nodes = [{"node": cloud_node, "port_number": 1}] - + response = await client.get(app.url_path_for("get_iface", project_id=project.id, link_id=link.id)) - + assert response.status_code == status.HTTP_200_OK result = response.json() - - assert "cloud-node-id" in result - - interface_info = result["cloud-node-id"] - assert interface_info["interface"] == "eth0" - assert interface_info["type"] == "ethernet" \ No newline at end of file + assert result["node_id"] == cloud_node.id + assert result["interface"] == "eth0" + assert result["type"] == "ethernet" \ No newline at end of file