diff --git a/ai-requirements.txt b/ai-requirements.txt index 135af4c09..8b46655a4 100644 --- a/ai-requirements.txt +++ b/ai-requirements.txt @@ -30,7 +30,8 @@ tiktoken>=0.8.0 langsmith>=0.7.7 # Network Automation -netmiko>=4.7.0 +paramiko>=5.0.0 # install version >= 5.0.0 to address CVE-2026-44405 +netmiko>=4.8.0 nornir>=3.5.0 nornir-netmiko>=1.0.1 nornir-utils>=0.2.0 diff --git a/docs/features/mcp-service.md b/docs/features/mcp-service.md index 5390211a3..4963ad748 100644 --- a/docs/features/mcp-service.md +++ b/docs/features/mcp-service.md @@ -187,7 +187,9 @@ All subsequent tool handler REST API calls use this JWT → zero extra bcrypt |------|-------------| | `drawing_list` | List drawings on canvas | | `drawing_get` | Get drawing details | -| `drawing_create` | Create drawing (SVG label/shape/image) | +| `drawing_create_text` | Create a text label (SVG assembled server-side, Web UI compatible) | +| `drawing_create_rectangle` | Create a rectangle annotation | +| `drawing_create_ellipse` | Create an ellipse/circle annotation | | `drawing_update` | Update drawing (position, rotation, SVG) | | `drawing_delete` | Delete a drawing | diff --git a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py index 148ccac63..822c54b20 100644 --- a/gns3server/agent/gns3_copilot/agent/gns3_copilot.py +++ b/gns3server/agent/gns3_copilot/agent/gns3_copilot.py @@ -77,7 +77,10 @@ from gns3server.agent.gns3_copilot.skills.registry import get_prompt from gns3server.agent.gns3_copilot.tools_v2 import ( ExecuteMultipleDeviceCommands, ExecuteMultipleDeviceConfigCommands, + GNS3CreateEllipseDrawingTool, GNS3CreateNodeTool, + GNS3CreateRectangleDrawingTool, + GNS3CreateTextDrawingTool, GNS3LinkTool, GNS3PacketFilterTool, GNS3StartNodeTool, @@ -103,6 +106,9 @@ TEACHING_ASSISTANT_MODE_TOOLS = [ GNS3TemplateTool(), # Get GNS3 node templates GNS3CreateNodeTool(), # Create new nodes in GNS3 GNS3LinkTool(), # Create links between nodes + GNS3CreateTextDrawingTool(), # Add text labels to the canvas + GNS3CreateRectangleDrawingTool(), # Add rectangle annotations to the canvas + GNS3CreateEllipseDrawingTool(), # Add ellipse annotations to the canvas GNS3StartNodeTool(), # Start GNS3 nodes (returns immediately) GNS3WaitTool(), # Wait for nodes to boot (pair with start_gns3_node) GNS3UpdateNodeNameTool(), # Update node name @@ -118,6 +124,9 @@ LAB_AUTOMATION_ASSISTANT_MODE_TOOLS = [ GNS3TemplateTool(), # Get GNS3 node templates GNS3CreateNodeTool(), # Create new nodes in GNS3 GNS3LinkTool(), # Create links between nodes + GNS3CreateTextDrawingTool(), # Add text labels to the canvas + GNS3CreateRectangleDrawingTool(), # Add rectangle annotations to the canvas + GNS3CreateEllipseDrawingTool(), # Add ellipse annotations to the canvas GNS3StartNodeTool(), # Start GNS3 nodes (returns immediately) GNS3WaitTool(), # Wait for nodes to boot (pair with start_gns3_node) GNS3StopNodeTool(), # Stop GNS3 nodes diff --git a/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py b/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py index ba3fca6c7..44e03688b 100644 --- a/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py +++ b/gns3server/agent/gns3_copilot/gns3_client/api_handlers.py @@ -55,6 +55,7 @@ from typing import Any from gns3server.agent.gns3_copilot.gns3_client.connector import Gns3Connector from gns3server.services import access_ticket_service from gns3server.services.access_tickets import DEFAULT_TICKET_TTL +from gns3server.utils.drawings import build_ellipse_svg, build_rectangle_svg, build_text_svg log = logging.getLogger(__name__) @@ -1076,3 +1077,95 @@ def marker_definition_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) # action == "delete" conn.http_call("delete", url) return {"message": f"Marker definition '{def_name}' deleted", "project_id": project_id, "def_name": def_name} + + +# ── Drawing handlers ────────────────────────────────────────────────────── +# SVG assembly and validation live in gns3server.utils.drawings (Web UI +# compatible format); these handlers only carry params and do the REST call. + + +def create_text_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + """Create a canvas text label; the SVG is assembled from validated params.""" + + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + try: + svg, width, height, line_height = build_text_svg( + params.get("text"), + font_size=params.get("font_size", 13), + color=params.get("color", "#000000"), + bold=params.get("bold", False), + font_family=params.get("font_family", "monospace"), + ) + except ValueError as e: + return {"error": str(e)} + conn = _get_connector(gns3_ctx) + data = {"svg": svg, "x": params.get("x", 0), "y": params.get("y", 0), "z": params.get("z", 1)} + result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json() + return { + "message": "Text drawing created", + "drawing": result, + "width": width, + "height": height, + "line_height": line_height, + } + + +def create_rectangle_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + """Create a canvas rectangle; the SVG is assembled from validated params.""" + + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + width = params.get("width") + height = params.get("height") + if width is None or height is None: + return {"error": "width and height are required"} + try: + svg, width, height = build_rectangle_svg( + width, + height, + fill=params.get("fill", "#FFFFFF"), + fill_opacity=params.get("fill_opacity", 1.0), + stroke=params.get("stroke"), + stroke_width=params.get("stroke_width", 1), + dashed=params.get("dashed", False), + dasharray=params.get("dasharray"), + rx=params.get("rx", 0), + ) + except ValueError as e: + return {"error": str(e)} + conn = _get_connector(gns3_ctx) + data = {"svg": svg, "x": params.get("x", 0), "y": params.get("y", 0), "z": params.get("z", 0)} + result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json() + return {"message": "Rectangle drawing created", "drawing": result, "width": width, "height": height} + + +def create_ellipse_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + """Create a canvas ellipse from a bounding box; the SVG is assembled from validated params.""" + + project_id = params.get("project_id") + if not project_id: + return {"error": "project_id is required"} + width = params.get("width") + height = params.get("height") + if width is None or height is None: + return {"error": "width and height are required"} + try: + svg, width, height = build_ellipse_svg( + width, + height, + fill=params.get("fill", "#FFFFFF"), + fill_opacity=params.get("fill_opacity", 1.0), + stroke=params.get("stroke"), + stroke_width=params.get("stroke_width", 1), + dashed=params.get("dashed", False), + dasharray=params.get("dasharray"), + ) + except ValueError as e: + return {"error": str(e)} + conn = _get_connector(gns3_ctx) + data = {"svg": svg, "x": params.get("x", 0), "y": params.get("y", 0), "z": params.get("z", 0)} + result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json() + return {"message": "Ellipse drawing created", "drawing": result, "width": width, "height": height} diff --git a/gns3server/agent/gns3_copilot/tools_v2/__init__.py b/gns3server/agent/gns3_copilot/tools_v2/__init__.py index 0818bd81c..078105ada 100644 --- a/gns3server/agent/gns3_copilot/tools_v2/__init__.py +++ b/gns3server/agent/gns3_copilot/tools_v2/__init__.py @@ -40,6 +40,7 @@ Main modules: - vpcs_tools_netmiko: VPCS device configuration tool using Netmiko - gns3_create_node: GNS3 node creation tool - gns3_create_link: GNS3 link creation tool +- gns3_create_drawing: GNS3 canvas drawing tools (text, rectangle, ellipse) - gns3_start_node: GNS3 node startup tool (immediate return, no boot wait) - gns3_wait: Wait timer tool (pair with node start to let devices boot) - gns3_get_node_temp: GNS3 template retrieval tool @@ -54,6 +55,11 @@ Author: Yue Guobin (岳国宾) # Import main tool classes from .config_tools_nornir import ExecuteMultipleDeviceConfigCommands from .display_tools_nornir import ExecuteMultipleDeviceCommands +from .gns3_create_drawing import ( + GNS3CreateEllipseDrawingTool, + GNS3CreateRectangleDrawingTool, + GNS3CreateTextDrawingTool, +) from .gns3_create_link import GNS3LinkTool from .gns3_create_node import GNS3CreateNodeTool from .gns3_get_node_temp import GNS3TemplateTool @@ -81,7 +87,10 @@ __url__ = "https://github.com/yueguobin/gns3-copilot" __all__ = [ "ExecuteMultipleDeviceCommands", "ExecuteMultipleDeviceConfigCommands", + "GNS3CreateEllipseDrawingTool", "GNS3CreateNodeTool", + "GNS3CreateRectangleDrawingTool", + "GNS3CreateTextDrawingTool", "GNS3LinkTool", "GNS3PacketFilterTool", "GNS3StartNodeQuickTool", diff --git a/gns3server/agent/gns3_copilot/tools_v2/gns3_create_drawing.py b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_drawing.py new file mode 100644 index 000000000..7f3f688fe --- /dev/null +++ b/gns3server/agent/gns3_copilot/tools_v2/gns3_create_drawing.py @@ -0,0 +1,162 @@ +# SPDX-License-Identifier: GPL-3.0-or-later +# +# GNS3-Copilot - AI-powered Network Lab Assistant for GNS3 +# +# This file is part of GNS3-Copilot project. +# +# GNS3-Copilot 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. +# +# GNS3-Copilot 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 GNS3-Copilot. If not, see . +# +# Copyright (C) 2025 Yue Guobin (岳国宾) +# Author: Yue Guobin (岳国宾) +# +# Project Home: https://github.com/yueguobin/gns3-copilot +# + +""" +GNS3 canvas drawing tools for text, rectangle and ellipse annotations. + +Thin LangChain wrappers over the shared REST handlers in +``gns3_client.api_handlers``; the Web UI compatible SVG is assembled and +validated by ``gns3server.utils.drawings`` (single implementation shared +with the MCP drawing tools). +""" + +import json +import logging +from typing import Any + +from langchain.tools import BaseTool +from langchain_core.callbacks import CallbackManagerForToolRun + +from gns3server.agent.gns3_copilot.gns3_client.api_handlers import ( + build_gns3_ctx, + create_ellipse_drawing_handler, + create_rectangle_drawing_handler, + create_text_drawing_handler, +) + +# Configure logging +logger = logging.getLogger(__name__) + + +class _GNS3CreateDrawingToolBase(BaseTool): + """Shared plumbing: parse JSON input, build ctx, dispatch to a handler.""" + + def _dispatch(self, input_data: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + raise NotImplementedError + + def _run( + self, + tool_input: str, + run_manager: CallbackManagerForToolRun | None = None, + ) -> dict[str, Any]: + try: + input_data = json.loads(tool_input) + + gns3_ctx = build_gns3_ctx() + if gns3_ctx is None: + logger.error("Failed to create GNS3 connector") + return {"error": "Failed to connect to GNS3 server. Please check your configuration."} + + return self._dispatch(input_data, gns3_ctx) + except json.JSONDecodeError as e: + logger.error("Invalid JSON input: %s", e) + return {"error": f"Invalid JSON input: {e}"} + except Exception as e: + logger.exception("Drawing creation failed") + return {"error": str(e)} + + +class GNS3CreateTextDrawingTool(_GNS3CreateDrawingToolBase): + """ + A LangChain tool to add a text label to the GNS3 project canvas. + + **Input**: + A JSON object with project_id, x, y (top-left corner on the canvas) + and text; optional font_size (1-200, default 13), color (#RRGGBB, + default #000000), bold (default false), font_family (default + monospace) and z (default 1, keeps text above links). + Newlines in text start additional lines; XML characters are escaped + automatically; the computed width/height are returned. + + **Output**: + A dictionary with the created drawing, its drawing_id and the + computed width/height/line_height, or an "error" key on failure. + """ + + name: str = "create_gns3_text_drawing" + description: str = """ + Adds a text label to the GNS3 project canvas. + Input: JSON with project_id, x, y (top-left corner), text and optional + font_size, color, bold, font_family, z. + Returns: the created drawing with its computed width and height. + """ + + def _dispatch(self, input_data: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + return create_text_drawing_handler(input_data, gns3_ctx) + + +class GNS3CreateRectangleDrawingTool(_GNS3CreateDrawingToolBase): + """ + A LangChain tool to add a rectangle to the GNS3 project canvas. + + **Input**: + A JSON object with project_id, x, y (top-left corner), width and + height (1-2000); optional fill (#RRGGBB or "none", default #FFFFFF), + fill_opacity (0-1, default 1.0), stroke (#RRGGBB, no border by + default), stroke_width (1-20), dashed (bool), dasharray (like "4,2", + requires stroke) and rx (corner radius, default 0). z defaults to 0 + so shapes stay behind nodes and text. + + **Output**: + A dictionary with the created drawing, or an "error" key on failure. + """ + + name: str = "create_gns3_rectangle_drawing" + description: str = """ + Adds a rectangle to the GNS3 project canvas. + Input: JSON with project_id, x, y, width, height and optional fill, + fill_opacity, stroke, stroke_width, dashed, dasharray, rx, z. + Returns: the created drawing with its width and height. + """ + + def _dispatch(self, input_data: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + return create_rectangle_drawing_handler(input_data, gns3_ctx) + + +class GNS3CreateEllipseDrawingTool(_GNS3CreateDrawingToolBase): + """ + A LangChain tool to add an ellipse (or circle) to the GNS3 canvas. + + **Input**: + A JSON object with project_id, x, y (bounding box top-left corner), + width and height (2-2000, equal values give a circle); the same + optional style fields as the rectangle tool. cx/cy are derived from + the bounding box; for odd width/height the drawing is 1 px smaller + than requested and the actual size is returned. + + **Output**: + A dictionary with the created drawing, or an "error" key on failure. + """ + + name: str = "create_gns3_ellipse_drawing" + description: str = """ + Adds an ellipse or circle to the GNS3 project canvas. + Input: JSON with project_id, x, y, width, height (bounding box) and + optional fill, fill_opacity, stroke, stroke_width, dashed, dasharray, z. + Returns: the created drawing with its actual width and height. + """ + + def _dispatch(self, input_data: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: + return create_ellipse_drawing_handler(input_data, gns3_ctx) diff --git a/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py b/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py index 50456cbf5..6e20dc72d 100644 --- a/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py +++ b/gns3server/agent/gns3_copilot/utils/gns3_drawing_utils.py @@ -33,6 +33,8 @@ Supports ellipse and rectangle shapes for two-node annotations. import math from typing import Any, Literal +from gns3server.utils.drawings import build_ellipse_svg, build_rectangle_svg, build_text_svg + # Default parameters DEFAULT_DEVICE_WIDTH = 50 DEFAULT_DEVICE_HEIGHT = 50 @@ -144,10 +146,16 @@ def calculate_two_node_shape( center_y = (node1_center_y + node2_center_y) / 2 color_scheme = _get_color_scheme(area_name) - text_svg = generate_text_svg(area_name, color_scheme) - text_svg_width = len(area_name) * 8 + 20 - text_svg_height = DEFAULT_FONT_SIZE + 16 + # Shared builder: Web UI compatible SVG plus the computed label size + # (the old len * 8 + 20 estimate ignored the Web UI's pt-based rendering). + text_svg, text_svg_width, text_svg_height, _ = build_text_svg( + area_name, + font_size=DEFAULT_FONT_SIZE, + color=color_scheme["stroke"], + bold=True, + font_family="TypeWriter", + ) if shape_type == "ellipse": rx = distance / 2 @@ -236,8 +244,9 @@ def generate_ellipse_svg( svg_width: int, svg_height: int, ) -> str: - """Generate SVG for ellipse.""" - return f"""""" + """Generate SVG for ellipse (delegates to the shared Web UI compatible builder).""" + svg, _, _ = build_ellipse_svg(rx * 2, ry * 2, fill=color_scheme["fill"], fill_opacity=color_scheme["fill_opacity"]) + return svg def generate_rectangle_svg( @@ -245,16 +254,21 @@ def generate_rectangle_svg( height: int, color_scheme: dict[str, Any], ) -> str: - """Generate SVG for rectangle.""" - return f"""""" + """Generate SVG for rectangle (delegates to the shared Web UI compatible builder).""" + svg, _, _ = build_rectangle_svg(width, height, fill=color_scheme["fill"], fill_opacity=color_scheme["fill_opacity"]) + return svg def generate_text_svg(text: str, color_scheme: dict[str, Any]) -> str: - """Generate SVG for text label.""" - text_width = len(text) * 8 + 20 - text_height = DEFAULT_FONT_SIZE + 16 - - return f"""{text}""" + """Generate SVG for text label (delegates to the shared Web UI compatible builder).""" + svg, _, _, _ = build_text_svg( + text, + font_size=DEFAULT_FONT_SIZE, + color=color_scheme["stroke"], + bold=True, + font_family="TypeWriter", + ) + return svg def _hsv_to_hex(h: int, s: int, v: int) -> str: diff --git a/gns3server/agent/mcp/__init__.py b/gns3server/agent/mcp/__init__.py index 27cce38b9..57b45e5ea 100644 --- a/gns3server/agent/mcp/__init__.py +++ b/gns3server/agent/mcp/__init__.py @@ -45,8 +45,11 @@ from sqlalchemy.ext.asyncio import AsyncSession from gns3server.agent.gns3_copilot.gns3_client.api_handlers import ( available_filters_handler, + create_ellipse_drawing_handler, create_link_handler, create_node_handler, + create_rectangle_drawing_handler, + create_text_drawing_handler, delete_link_handler, delete_node_file_handler, delete_node_handler, @@ -109,7 +112,6 @@ from .device_config import ( vpcs_config_set_handler, ) from .drawings import ( - create_drawing_handler, delete_drawing_handler, get_drawing_handler, get_drawings_handler, @@ -1539,41 +1541,142 @@ async def drawing_list( @mcp.tool() -async def drawing_create( +async def drawing_create_text( project_id: Annotated[str, Field(description="Project name or UUID")], - svg: Annotated[str, Field(description="SVG content for the drawing")], - x: Annotated[int, Field(description="X coordinate (default: 0)")] = 0, - y: Annotated[int, Field(description="Y coordinate (default: 0)")] = 0, - z: Annotated[int, Field(description="Z layer (default: 0)")] = 0, - locked: Annotated[bool, Field(description="Lock the drawing (default: false)")] = False, - rotation: Annotated[int, Field(description="Rotation angle in degrees, -359 to 359 (default: 0)")] = 0, + x: Annotated[int, Field(description="X coordinate of the text top-left corner on the canvas")], + y: Annotated[int, Field(description="Y coordinate of the text top-left corner on the canvas")], + text: Annotated[str, Field(description="Label text; newlines (\\n) start additional lines")], + font_size: Annotated[int, Field(description="Font size, 1-200 (default: 13)")] = 13, + color: Annotated[str, Field(description="Text color, hex like #RRGGBB (default: #000000)")] = "#000000", + bold: Annotated[bool, Field(description="Bold text (default: false)")] = False, + font_family: Annotated[str, Field(description="Font family name (default: monospace)")] = "monospace", + z: Annotated[int, Field(description="Z layer; 1 keeps text above links (default: 1)")] = 1, ) -> list[dict[str, Any]]: - """Create a new drawing (label, shape, or image) on a project canvas. + """Add a text label to a project canvas. - GNS3 SVG rendering notes: - - MUST have a solid fill color (e.g. fill=\"#FF0000\") to render. - fill=\"none\" or fill=\"transparent\" will be invisible in the GUI. - - works correctly with or without fill. - - and work normally. + The SVG is assembled server-side in the Web UI compatible format + (validation, escaping and sizing included). - SVG examples: - Text label: R1 - Rectangle: - Ellipse: - Line: - Dashed line: + - XML characters (& < >) are escaped automatically. + - Multi-line: embed newlines (\\n) in text; each line renders below the previous one. + - The drawing width/height are computed from the text and returned, so labels + can be laid out without guessing the selection box size. + - Constraints: color must be hex like #RRGGBB; font_family accepts plain names + (letters, digits, spaces, . ' -); text capped at 500 characters. + + Example: bold red label at (100, 80): + drawing_create_text(project_id="demo", x=100, y=80, text="R1 management", color="#CC0000", bold=True) """ return await asyncio.to_thread( _run_handler_sync, - create_drawing_handler, + create_text_drawing_handler, { "project_id": project_id, - "svg": svg, + "text": text, "x": x, "y": y, "z": z, - "locked": locked, - "rotation": rotation, + "font_size": font_size, + "color": color, + "bold": bold, + "font_family": font_family, + }, + ) + + +@mcp.tool() +async def drawing_create_rectangle( + project_id: Annotated[str, Field(description="Project name or UUID")], + x: Annotated[int, Field(description="X coordinate of the rectangle top-left corner on the canvas")], + y: Annotated[int, Field(description="Y coordinate of the rectangle top-left corner on the canvas")], + width: Annotated[int, Field(description="Rectangle width in px, 1-2000")], + height: Annotated[int, Field(description="Rectangle height in px, 1-2000")], + fill: Annotated[str, Field(description='Fill color, hex like #RRGGBB or "none" (default: #FFFFFF)')] = "#FFFFFF", + fill_opacity: Annotated[float, Field(description="Fill opacity, 0-1 (default: 1.0)")] = 1.0, + stroke: Annotated[str | None, Field(description="Border color, hex like #RRGGBB; omit for no border")] = None, + stroke_width: Annotated[int, Field(description="Border width, 1-20 (default: 1)")] = 1, + dashed: Annotated[bool, Field(description="Dashed border, uses the default 10,6 pattern (default: false)")] = False, + dasharray: Annotated[str | None, Field(description='Custom dash pattern like "4,2"; requires stroke')] = None, + rx: Annotated[int, Field(description="Corner radius, 0 to min(width, height)/2 (default: 0)")] = 0, + z: Annotated[int, Field(description="Z layer; 0 keeps shapes behind nodes and text (default: 0)")] = 0, +) -> list[dict[str, Any]]: + """Add a rectangle to a project canvas. + + The SVG is assembled server-side in the Web UI compatible format + (validation included). + + - fill "none" renders an outline-only box (stroke recommended then). + - Border attributes are only emitted when stroke is given. + - dasharray must be plain "on,off" numbers; the Qt-remapped patterns + ("25, 25", "5, 25", ...) are rejected because the Web UI rewrites them. + + Example: semi-transparent outline box at (200, 150): + drawing_create_rectangle(project_id="demo", x=200, y=150, width=400, height=200, + fill="none", stroke="#333333", dashed=True) + """ + return await asyncio.to_thread( + _run_handler_sync, + create_rectangle_drawing_handler, + { + "project_id": project_id, + "x": x, + "y": y, + "z": z, + "width": width, + "height": height, + "fill": fill, + "fill_opacity": fill_opacity, + "stroke": stroke, + "stroke_width": stroke_width, + "dashed": dashed, + "dasharray": dasharray, + "rx": rx, + }, + ) + + +@mcp.tool() +async def drawing_create_ellipse( + project_id: Annotated[str, Field(description="Project name or UUID")], + x: Annotated[int, Field(description="X coordinate of the bounding box top-left corner on the canvas")], + y: Annotated[int, Field(description="Y coordinate of the bounding box top-left corner on the canvas")], + width: Annotated[int, Field(description="Bounding box width in px, 2-2000; equal to height gives a circle")], + height: Annotated[int, Field(description="Bounding box height in px, 2-2000")], + fill: Annotated[str, Field(description='Fill color, hex like #RRGGBB or "none" (default: #FFFFFF)')] = "#FFFFFF", + fill_opacity: Annotated[float, Field(description="Fill opacity, 0-1 (default: 1.0)")] = 1.0, + stroke: Annotated[str | None, Field(description="Border color, hex like #RRGGBB; omit for no border")] = None, + stroke_width: Annotated[int, Field(description="Border width, 1-20 (default: 1)")] = 1, + dashed: Annotated[bool, Field(description="Dashed border, uses the default 10,6 pattern (default: false)")] = False, + dasharray: Annotated[str | None, Field(description='Custom dash pattern like "4,2"; requires stroke')] = None, + z: Annotated[int, Field(description="Z layer; 0 keeps shapes behind nodes and text (default: 0)")] = 0, +) -> list[dict[str, Any]]: + """Add an ellipse (or circle) to a project canvas. + + The SVG is assembled server-side in the Web UI compatible format. cx/cy are + derived from the bounding box (the Web UI treats them as offsets inside the + drawing, not canvas coordinates); for odd width/height the drawing is 1 px + smaller than requested and the actual size is returned. + + Example: circle of diameter 160 at (300, 300): + drawing_create_ellipse(project_id="demo", x=300, y=300, width=160, height=160, + fill="#5AA9DD", fill_opacity=0.8) + """ + return await asyncio.to_thread( + _run_handler_sync, + create_ellipse_drawing_handler, + { + "project_id": project_id, + "x": x, + "y": y, + "z": z, + "width": width, + "height": height, + "fill": fill, + "fill_opacity": fill_opacity, + "stroke": stroke, + "stroke_width": stroke_width, + "dashed": dashed, + "dasharray": dasharray, }, ) diff --git a/gns3server/agent/mcp/drawings.py b/gns3server/agent/mcp/drawings.py index 7ca6daeda..7c459fe6a 100644 --- a/gns3server/agent/mcp/drawings.py +++ b/gns3server/agent/mcp/drawings.py @@ -51,24 +51,6 @@ def get_drawings_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di return {"drawings": drawings, "count": len(drawings)} -def create_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: - project_id = params.get("project_id") - svg = params.get("svg") - if not project_id or not svg: - return {"error": "project_id and svg are required"} - conn = _get_connector(gns3_ctx) - data = { - "svg": svg, - "x": params.get("x", 0), - "y": params.get("y", 0), - "z": params.get("z", 0), - "locked": params.get("locked", False), - "rotation": params.get("rotation", 0), - } - result = conn.http_call("post", f"{conn.base_url}/projects/{project_id}/drawings", json_data=data).json() - return {"message": "Drawing created", "drawing": result} - - def get_drawing_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: project_id = params.get("project_id") drawing_id = params.get("drawing_id") diff --git a/gns3server/api/routes/controller/acl.py b/gns3server/api/routes/controller/acl.py index a454a1d61..49f814966 100644 --- a/gns3server/api/routes/controller/acl.py +++ b/gns3server/api/routes/controller/acl.py @@ -80,7 +80,11 @@ def _iter_route_paths(routes: Sequence[BaseRoute], prefix: str = "", include_mou yield from _iter_route_paths(mounted_routes, _join_paths(prefix, route.path), include_mounted_routes) -@router.get("/endpoints", dependencies=[Depends(has_privilege("ACE.Audit"))]) +@router.get( + "/endpoints", + response_model=List[schemas.ACEEndpoint], + dependencies=[Depends(has_privilege("ACE.Audit"))], +) async def endpoints( users_repo: UsersRepository = Depends(get_repository(UsersRepository)), rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), diff --git a/gns3server/api/routes/controller/api_keys.py b/gns3server/api/routes/controller/api_keys.py index 42f1d66c2..c3297e612 100644 --- a/gns3server/api/routes/controller/api_keys.py +++ b/gns3server/api/routes/controller/api_keys.py @@ -50,12 +50,12 @@ def _generate_api_key(api_key_id: UUID | None = None) -> tuple[str, str, str, UU return raw_key, key_hash, key_prefix, api_key_id -@router.post("", status_code=status.HTTP_201_CREATED) +@router.post("", response_model=schemas.ApiKeyCreated, status_code=status.HTTP_201_CREATED) async def create_api_key( api_key_data: schemas.ApiKeyCreate, current_user: schemas.User = Depends(get_current_active_user), api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)), -) -> dict: +) -> schemas.ApiKeyCreated: """Create a new API key. The full key is returned only once.""" raw_key, key_hash, key_prefix, new_key_id = _generate_api_key() @@ -66,42 +66,42 @@ async def create_api_key( key_hash=key_hash, key_prefix=key_prefix, ) - return { - "api_key_id": str(db_key.api_key_id), - "api_key": raw_key, - "name": db_key.name, - "key_prefix": db_key.key_prefix, - "created_at": db_key.created_at.isoformat() if db_key.created_at else None, - } + return schemas.ApiKeyCreated( + api_key_id=db_key.api_key_id, + api_key=raw_key, + name=db_key.name, + key_prefix=db_key.key_prefix, + created_at=db_key.created_at.isoformat() if db_key.created_at else None, + ) -@router.get("") +@router.get("", response_model=list[schemas.ApiKey]) async def list_api_keys( current_user: schemas.User = Depends(get_current_active_user), api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)), -) -> list[dict]: +) -> list[schemas.ApiKey]: """List all API keys for the current user.""" keys = await api_keys_repo.get_api_keys_by_user(current_user.user_id) return [ - { - "api_key_id": str(k.api_key_id), - "name": k.name, - "key_prefix": k.key_prefix, - "created_at": k.created_at.isoformat() if k.created_at else None, - "last_used_at": k.last_used_at.isoformat() if k.last_used_at else None, - "revoked": k.revoked, - } + schemas.ApiKey( + api_key_id=k.api_key_id, + name=k.name, + key_prefix=k.key_prefix, + created_at=k.created_at.isoformat() if k.created_at else None, + last_used_at=k.last_used_at.isoformat() if k.last_used_at else None, + revoked=k.revoked, + ) for k in keys ] -@router.post("/{api_key_id}/revoke", status_code=status.HTTP_200_OK) +@router.post("/{api_key_id}/revoke", response_model=schemas.ApiKeyMessage, status_code=status.HTTP_200_OK) async def revoke_api_key( api_key_id: UUID, current_user: schemas.User = Depends(get_current_active_user), api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)), -) -> dict: +) -> schemas.ApiKeyMessage: """Revoke an API key. It will immediately stop working, but can be restored.""" key = await api_keys_repo.get_api_key(api_key_id) @@ -111,15 +111,15 @@ async def revoke_api_key( raise HTTPException(status_code=403, detail="Cannot modify another user's API key") await api_keys_repo.revoke_api_key(api_key_id) - return {"message": f"API key '{key.name}' revoked"} + return schemas.ApiKeyMessage(message=f"API key '{key.name}' revoked") -@router.post("/{api_key_id}/restore", status_code=status.HTTP_200_OK) +@router.post("/{api_key_id}/restore", response_model=schemas.ApiKeyMessage, status_code=status.HTTP_200_OK) async def restore_api_key( api_key_id: UUID, current_user: schemas.User = Depends(get_current_active_user), api_keys_repo: ApiKeysRepository = Depends(get_repository(ApiKeysRepository)), -) -> dict: +) -> schemas.ApiKeyMessage: """Restore a previously revoked API key.""" key = await api_keys_repo.get_api_key(api_key_id) @@ -129,7 +129,7 @@ async def restore_api_key( raise HTTPException(status_code=403, detail="Cannot modify another user's API key") await api_keys_repo.restore_api_key(api_key_id) - return {"message": f"API key '{key.name}' restored"} + return schemas.ApiKeyMessage(message=f"API key '{key.name}' restored") @router.delete("/{api_key_id}", status_code=status.HTTP_204_NO_CONTENT) diff --git a/gns3server/api/routes/controller/appliances.py b/gns3server/api/routes/controller/appliances.py index 1d3f0e413..c20554a9f 100644 --- a/gns3server/api/routes/controller/appliances.py +++ b/gns3server/api/routes/controller/appliances.py @@ -82,6 +82,8 @@ def get_appliance(appliance_id: UUID) -> schemas.Appliance: @router.post( "/{appliance_id}/version", + response_model=schemas.Appliance, + response_model_exclude_unset=True, status_code=status.HTTP_201_CREATED, dependencies=[Depends(has_privilege("Appliance.Allocate"))], ) diff --git a/gns3server/api/routes/controller/computes.py b/gns3server/api/routes/controller/computes.py index a33ebc9f6..796a72063 100644 --- a/gns3server/api/routes/controller/computes.py +++ b/gns3server/api/routes/controller/computes.py @@ -212,7 +212,7 @@ async def vmware_vms(compute_id: Union[str, UUID]) -> List[schemas.ComputeVMware return result -@router.post("/{compute_id}/dynamips/auto_idlepc") +@router.post("/{compute_id}/dynamips/auto_idlepc", response_model=schemas.IdlePC) async def dynamips_autoidlepc(compute_id: Union[str, UUID], auto_idle_pc: schemas.AutoIdlePC): """ Find a suitable Idle-PC value for a given IOS image. This may take a few minutes. diff --git a/gns3server/api/routes/controller/controller.py b/gns3server/api/routes/controller/controller.py index cef3ee77e..a6a499372 100644 --- a/gns3server/api/routes/controller/controller.py +++ b/gns3server/api/routes/controller/controller.py @@ -162,7 +162,12 @@ async def update_iou_license(iou_license: schemas.IOULicense) -> schemas.IOULice return current_iou_license -@router.get("/statistics", dependencies=[Depends(get_current_active_user)]) +@router.get( + "/statistics", + response_model=schemas.ServerStatistics, + response_model_exclude_unset=True, + dependencies=[Depends(get_current_active_user)], +) async def statistics() -> dict: """ Return server statistics including compute resources, projects, and nodes. diff --git a/gns3server/api/routes/controller/gns3vm.py b/gns3server/api/routes/controller/gns3vm.py index e0ae90eba..21d2cdf49 100644 --- a/gns3server/api/routes/controller/gns3vm.py +++ b/gns3server/api/routes/controller/gns3vm.py @@ -30,7 +30,7 @@ from gns3server.controller import Controller router = APIRouter() -@router.get("/engines") +@router.get("/engines", response_model=List[schemas.GNS3VMEngineInfo]) async def get_engines() -> List[dict]: """ Return the list of supported engines for the GNS3VM. @@ -40,7 +40,7 @@ async def get_engines() -> List[dict]: return gns3_vm.engine_list() -@router.get("/engines/{engine}/vms") +@router.get("/engines/{engine}/vms", response_model=List[schemas.GNS3VMInfo]) async def get_vms(engine: str) -> List[dict]: """ Return all the available VMs for a specific virtualization engine. diff --git a/gns3server/api/routes/controller/images.py b/gns3server/api/routes/controller/images.py index c82f2464a..4593a4f97 100644 --- a/gns3server/api/routes/controller/images.py +++ b/gns3server/api/routes/controller/images.py @@ -329,7 +329,12 @@ async def prune_images( ) -@router.post("/install", status_code=status.HTTP_200_OK, dependencies=[Depends(has_privilege("Image.Allocate"))]) +@router.post( + "/install", + response_model=schemas.ImageInstallResult, + status_code=status.HTTP_200_OK, + dependencies=[Depends(has_privilege("Image.Allocate"))], +) async def install_images( images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)), templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)), diff --git a/gns3server/api/routes/controller/links.py b/gns3server/api/routes/controller/links.py index 83c0899fd..d17119624 100644 --- a/gns3server/api/routes/controller/links.py +++ b/gns3server/api/routes/controller/links.py @@ -20,7 +20,7 @@ API routes for links. import logging import os -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional from uuid import UUID, uuid4 import aiohttp @@ -205,11 +205,11 @@ async def create_links(project_id: UUID, links_create: List[schemas.LinkCreate]) @router.get( "/{link_id}/available_filters", - response_model=List[schemas.LinkFilterDefinition], + response_model=List[schemas.LinkFilterInfo], response_model_exclude_unset=True, dependencies=[Depends(has_privilege("Link.Audit"))], ) -async def get_filters(link: Link = Depends(dep_link)) -> List[schemas.LinkFilterDefinition]: +async def get_filters(link: Link = Depends(dep_link)) -> List[dict]: """ Return all filters available for a given link. @@ -353,6 +353,7 @@ async def stop_capture(link: Link = Depends(dep_link)) -> None: @router.post( "/{link_id}/capture/wireshark/restart", + response_model=schemas.LinkWiresharkRestart, status_code=status.HTTP_200_OK, dependencies=[Depends(has_privilege("Link.Capture"))], ) @@ -510,7 +511,12 @@ async def web_wireshark_websocket( pass -@router.get("/{link_id}/markers", dependencies=[Depends(has_privilege("Link.Audit"))]) +@router.get( + "/{link_id}/markers", + response_model=Dict[str, schemas.Marker], + response_model_exclude_unset=True, + dependencies=[Depends(has_privilege("Link.Audit"))], +) async def get_markers(link: Link = Depends(dep_link)) -> dict: """ Return all traffic-insight markers configured on this link. @@ -522,7 +528,11 @@ async def get_markers(link: Link = Depends(dep_link)) -> dict: @router.post( - "/{link_id}/markers", status_code=status.HTTP_201_CREATED, dependencies=[Depends(has_privilege("Link.Modify"))] + "/{link_id}/markers", + response_model=schemas.Marker, + response_model_exclude_unset=True, + status_code=status.HTTP_201_CREATED, + dependencies=[Depends(has_privilege("Link.Modify"))], ) async def create_marker(marker_data: schemas.MarkerCreate, link: Link = Depends(dep_link)) -> dict: """ @@ -567,7 +577,12 @@ async def delete_marker(marker_name: str, link: Link = Depends(dep_link)) -> Non await link.stop_marker(marker_name) -@router.put("/{link_id}/markers/{marker_name}", dependencies=[Depends(has_privilege("Link.Modify"))]) +@router.put( + "/{link_id}/markers/{marker_name}", + response_model=schemas.Marker, + response_model_exclude_unset=True, + dependencies=[Depends(has_privilege("Link.Modify"))], +) async def update_marker(marker_name: str, marker_data: schemas.MarkerUpdate, link: Link = Depends(dep_link)) -> dict: """ Update a traffic-insight marker (change BPF, tag, or enabled). diff --git a/gns3server/api/routes/controller/nodes.py b/gns3server/api/routes/controller/nodes.py index 04dbd332a..988524835 100644 --- a/gns3server/api/routes/controller/nodes.py +++ b/gns3server/api/routes/controller/nodes.py @@ -573,7 +573,11 @@ async def get_node_links(node: Node = Depends(dep_node)) -> List[schemas.Link]: return links -@router.get("/{node_id}/dynamips/auto_idlepc", dependencies=[Depends(has_privilege("Node.Audit"))]) +@router.get( + "/{node_id}/dynamips/auto_idlepc", + response_model=schemas.IdlePC, + dependencies=[Depends(has_privilege("Node.Audit"))], +) async def auto_idlepc(node: Node = Depends(dep_node)) -> dict: """ Compute an Idle-PC value for a Dynamips node diff --git a/gns3server/api/routes/controller/projects.py b/gns3server/api/routes/controller/projects.py index aceb39811..3736aba39 100644 --- a/gns3server/api/routes/controller/projects.py +++ b/gns3server/api/routes/controller/projects.py @@ -31,7 +31,7 @@ import gns3server.utils.zipfile_zstd as zipfile log = logging.getLogger() -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional from uuid import UUID from fastapi import ( @@ -237,7 +237,9 @@ async def delete_project( await rbac_repo.delete_all_ace_starting_with_path(f"/projects/{project.id}") -@router.get("/{project_id}/stats", dependencies=[Depends(has_privilege("Project.Audit"))]) +@router.get( + "/{project_id}/stats", response_model=schemas.ProjectStats, dependencies=[Depends(has_privilege("Project.Audit"))] +) def get_project_stats(project: Project = Depends(dep_project)) -> dict: """ Return a project statistics. @@ -248,7 +250,12 @@ def get_project_stats(project: Project = Depends(dep_project)) -> dict: return project.stats() -@router.get("/{project_id}/markers", dependencies=[Depends(has_privilege("Project.Audit"))]) +@router.get( + "/{project_id}/markers", + response_model=Dict[str, schemas.ProjectMarker], + response_model_exclude_unset=True, + dependencies=[Depends(has_privilege("Project.Audit"))], +) def get_project_markers(project: Project = Depends(dep_project)) -> dict: """ Return all traffic-insight markers across every link in the project. @@ -279,6 +286,7 @@ async def _replay_response(awaitable): @router.get( "/{project_id}/markers/tags/{tag}/replay/range", + response_model=schemas.ReplayTimeline, dependencies=[Depends(has_privilege("Project.Audit"))], ) async def replay_tag_range( @@ -319,6 +327,7 @@ async def replay_tag_range( @router.get( "/{project_id}/markers/tags/{tag}/replay/frames", + response_model=schemas.ReplayFrames, dependencies=[Depends(has_privilege("Project.Audit"))], ) async def replay_tag_frames( @@ -359,6 +368,8 @@ async def replay_tag_frames( @router.get( "/{project_id}/markers/tags/{tag}/replay/frame/detail", + response_model=schemas.ReplayFrameDetail, + response_model_exclude_unset=True, dependencies=[Depends(has_privilege("Project.Audit"))], ) async def replay_tag_frame_detail( @@ -397,7 +408,12 @@ async def replay_tag_frame_detail( # --------------------------------------------------------------------------- -@router.get("/{project_id}/marker-definitions", dependencies=[Depends(has_privilege("Project.Audit"))]) +@router.get( + "/{project_id}/marker-definitions", + response_model=Dict[str, schemas.MarkerDefinitionWithLinks], + response_model_exclude_unset=True, + dependencies=[Depends(has_privilege("Project.Audit"))], +) def get_marker_definitions(project: Project = Depends(dep_project)) -> dict: """ Return all project-level marker definitions with their bound link IDs. @@ -419,6 +435,8 @@ def get_marker_definitions(project: Project = Depends(dep_project)) -> dict: @router.post( "/{project_id}/marker-definitions", + response_model=schemas.MarkerDefinition, + response_model_exclude_unset=True, status_code=status.HTTP_201_CREATED, dependencies=[Depends(has_privilege("Project.Modify"))], ) @@ -446,7 +464,12 @@ async def create_marker_definition( return project.marker_definitions.get(name, {}) -@router.put("/{project_id}/marker-definitions/{def_name}", dependencies=[Depends(has_privilege("Project.Modify"))]) +@router.put( + "/{project_id}/marker-definitions/{def_name}", + response_model=schemas.MarkerDefinition, + response_model_exclude_unset=True, + dependencies=[Depends(has_privilege("Project.Modify"))], +) async def update_marker_definition( def_name: str, def_data: schemas.MarkerDefinitionCreate, project: Project = Depends(dep_project) ) -> dict: diff --git a/gns3server/api/routes/controller/symbols.py b/gns3server/api/routes/controller/symbols.py index a9c80741a..afa004398 100644 --- a/gns3server/api/routes/controller/symbols.py +++ b/gns3server/api/routes/controller/symbols.py @@ -21,7 +21,7 @@ API routes for symbols. import logging import os -from typing import List +from typing import Dict, List from fastapi import APIRouter, Depends, Request, Response, status from fastapi.responses import FileResponse @@ -40,7 +40,7 @@ log = logging.getLogger(__name__) router = APIRouter() -@router.get("", dependencies=[Depends(has_privilege("Symbol.Audit"))]) +@router.get("", response_model=List[schemas.SymbolInfo], dependencies=[Depends(has_privilege("Symbol.Audit"))]) def get_symbols() -> List[dict]: """ Return all symbols. @@ -83,6 +83,7 @@ async def get_symbol(symbol_id: str, request: Request) -> Response: @router.get( "/{symbol_id:path}/dimensions", + response_model=schemas.SymbolDimensions, responses={404: {"model": schemas.ErrorMessage, "description": "Could not find symbol"}}, dependencies=[Depends(has_privilege("Symbol.Audit"))], ) @@ -102,7 +103,11 @@ async def get_symbol_dimensions(symbol_id: str) -> dict: raise ControllerNotFoundError(f"Could not get symbol file: {e}") -@router.get("/default_symbols", dependencies=[Depends(has_privilege("Symbol.Audit"))]) +@router.get( + "/default_symbols", + response_model=Dict[str, Dict[str, str]], + dependencies=[Depends(has_privilege("Symbol.Audit"))], +) def get_default_symbols() -> dict: """ Return all default symbols. diff --git a/gns3server/controller/node.py b/gns3server/controller/node.py index 5df81d8b1..6fd35d653 100644 --- a/gns3server/controller/node.py +++ b/gns3server/controller/node.py @@ -16,7 +16,6 @@ # along with this program. If not, see . import asyncio -import contextlib import copy import html import logging @@ -139,7 +138,6 @@ class Node: self._node_directory = None self._status = "stopped" self._status_changed = asyncio.Event() - self._transition = None self._template_id = template_id self._x = 0 self._y = 0 @@ -907,19 +905,6 @@ class Node: async def destroy(self): await self.delete() - @contextlib.asynccontextmanager - async def _transitioning(self, transition): - """ - Report a transitional status while a lifecycle call is in flight - """ - - self._transition = transition - self.project.emit_notification("node.updated", self.asdict()) - try: - yield - finally: - self._transition = None - async def wait_for_status(self, status): """ Wait until the controller side status of the node matches @@ -954,19 +939,18 @@ class Node: f"Cannot start node '{self._name}': {len(failed_links)} deferred link(s) " "could not be restored. Please try again." ) - async with self._transitioning("starting"): - try: - # For IOU: we need to send the licence everytime we start a node - if self.node_type == "iou": - license_check = self._project.controller.iou_license.get("license_check", True) - iourc_content = self._project.controller.iou_license.get("iourc_content", None) - await self.post( - "/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content} - ) - else: - await self.post("/start", data=data, timeout=240) - except asyncio.TimeoutError: - raise ControllerTimeoutError(f"Timeout when starting {self._name}") + try: + # For IOU: we need to send the licence everytime we start a node + if self.node_type == "iou": + license_check = self._project.controller.iou_license.get("license_check", True) + iourc_content = self._project.controller.iou_license.get("iourc_content", None) + await self.post( + "/start", timeout=240, data={"license_check": license_check, "iourc_content": iourc_content} + ) + else: + await self.post("/start", data=data, timeout=240) + except asyncio.TimeoutError: + raise ControllerTimeoutError(f"Timeout when starting {self._name}") self._confirm_status("started") # A started node may have brought kernel anchors into existence that # a link needs to finish wiring (the Ethernet-switch fast path joins @@ -983,16 +967,15 @@ class Node: if self.missing_image: return stopped = False - async with self._transitioning("stopping"): - try: - await self.post("/stop", timeout=240, dont_connect=True) - stopped = True - except (ComputeError, ControllerError): - # We don't care if a node is down at this step - if strict: - raise - except asyncio.TimeoutError: - raise ControllerTimeoutError(f"Timeout when stopping {self._name}") + try: + await self.post("/stop", timeout=240, dont_connect=True) + stopped = True + except (ComputeError, ControllerError): + # We don't care if a node is down at this step + if strict: + raise + except asyncio.TimeoutError: + raise ControllerTimeoutError(f"Timeout when stopping {self._name}") if stopped: self._confirm_status("stopped") @@ -1388,7 +1371,7 @@ class Node: additional_data = { "project_id": self._project.id, "command_line": self._command_line, - "status": self._transition or self._status, + "status": self._status, "console_host": str(self._compute.console_host), "node_directory": self._node_directory, "ports": [port.asdict() for port in self.ports], diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index 7419f911c..74ddfe4af 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -26,10 +26,20 @@ from .controller.computes import ( ComputeUpdate, ComputeVirtualBoxVM, ComputeVMwareVM, + IdlePC, ) from .controller.drawings import Drawing -from .controller.gns3vm import GNS3VM -from .controller.images import Image, ImageSyncJob, ImageSyncRequest, ImageType, ImageUpload +from .controller.gns3vm import GNS3VM, GNS3VMEngineInfo, GNS3VMInfo +from .controller.images import ( + Image, + ImageInstallCreated, + ImageInstallResult, + ImageInstallSkipped, + ImageSyncJob, + ImageSyncRequest, + ImageType, + ImageUpload, +) # Controller schemas from .controller.links import ( @@ -38,15 +48,20 @@ from .controller.links import ( LinkBatchResult, LinkCapture, LinkCreate, - LinkFilterDefinition, + LinkFilterInfo, LinkFilterParameter, LinkFilters, LinkFilterType, LinkIfaceInfo, LinkUpdate, + LinkWiresharkRestart, + Marker, MarkerCreate, + MarkerDefinition, MarkerDefinitionCreate, + MarkerDefinitionWithLinks, MarkerUpdate, + ProjectMarker, UDPPortInfo, ) from .controller.nodes import Node, NodeBatchResult, NodeCapture, NodeCreate, NodeDuplicate, NodeUpdate @@ -57,6 +72,7 @@ from .controller.projects import ( ProjectCreate, ProjectDuplicate, ProjectFile, + ProjectStats, ProjectUpdate, ) from .controller.templates import Template, TemplateCreate, TemplateUpdate, TemplateUsage @@ -124,12 +140,32 @@ from .compute.vmware_nodes import VMware, VMwareCreate, VMwareUpdate from .compute.vpcs_nodes import VPCS, VPCSCreate, VPCSUpdate from .controller.capabilities import Capabilities from .controller.iou_license import IOULicense +from .controller.marker_replay import ( + ReplayFrame, + ReplayFrameDetail, + ReplayFrames, + ReplayFrameSource, + ReplaySource, + ReplayTimeline, + ReplayTreeNode, +) from .controller.netmiko import NetmikoDeviceType, NetmikoDeviceTypeList from .controller.notifications import Notification, NotificationAction from .controller.pools import Resource, ResourceCreate, ResourcePool, ResourcePoolCreate, ResourcePoolUpdate -from .controller.rbac import ACE, ACECreate, ACEUpdate, Privilege, Role, RoleCreate, RoleUpdate +from .controller.rbac import ACE, ACECreate, ACEEndpoint, ACEUpdate, Privilege, Role, RoleCreate, RoleUpdate from .controller.settings import SettingsResponse, SettingsUpdate, SettingsUpdateResponse from .controller.snapshots import Snapshot, SnapshotCreate +from .controller.statistics import ( + ComputeStatistics, + ComputeStatisticsEntry, + LinkStatistics, + NodeStatistics, + ProjectCountStatistics, + ServerStatistics, + WebWiresharkContainer, + WebWiresharkStatistics, +) +from .controller.symbols import SymbolDimensions, SymbolInfo from .controller.templates.cloud_templates import CloudTemplate, CloudTemplateUpdate from .controller.templates.docker_templates import DockerTemplate, DockerTemplateUpdate from .controller.templates.dynamips_templates import ( @@ -158,7 +194,7 @@ from .controller.templates.vmware_templates import VMwareTemplate, VMwareTemplat # Controller template schemas from .controller.templates.vpcs_templates import VPCSTemplate, VPCSTemplateUpdate -from .controller.tokens import ApiKeyCreate, RefreshTokenRequest, Token +from .controller.tokens import ApiKey, ApiKeyCreate, ApiKeyCreated, ApiKeyMessage, RefreshTokenRequest, Token # Schemas for both controller and compute from .qemu_disk_image import QemuDiskImageCreate, QemuDiskImageFormat, QemuDiskImageUpdate @@ -172,12 +208,16 @@ __all__ = [ "UDPNIO", "VPCS", "ACECreate", + "ACEEndpoint", "ACEUpdate", "ATMSwitch", "ATMSwitchCreate", "ATMSwitchUpdate", "AnchorNIO", + "ApiKey", "ApiKeyCreate", + "ApiKeyCreated", + "ApiKeyMessage", "Appliance", "ApplianceVersion", "ApplianceVersionCreate", @@ -212,6 +252,8 @@ __all__ = [ "Compute", "ComputeCreate", "ComputeDockerImage", + "ComputeStatistics", + "ComputeStatisticsEntry", "ComputeUpdate", "ComputeVMwareVM", "ComputeVirtualBoxVM", @@ -243,13 +285,19 @@ __all__ = [ "FrameRelaySwitch", "FrameRelaySwitchCreate", "FrameRelaySwitchUpdate", + "GNS3VMEngineInfo", + "GNS3VMInfo", "IOUCreate", "IOULicense", "IOUStart", "IOUTemplate", "IOUTemplateUpdate", "IOUUpdate", + "IdlePC", "Image", + "ImageInstallCreated", + "ImageInstallResult", + "ImageInstallSkipped", "ImageSyncJob", "ImageSyncRequest", "ImageType", @@ -265,15 +313,20 @@ __all__ = [ "LinkBatchResult", "LinkCapture", "LinkCreate", - "LinkFilterDefinition", + "LinkFilterInfo", "LinkFilterParameter", "LinkFilterType", "LinkFilters", "LinkIfaceInfo", + "LinkStatistics", "LinkUpdate", + "LinkWiresharkRestart", "LoggedInUserUpdate", + "Marker", "MarkerCreate", + "MarkerDefinition", "MarkerDefinitionCreate", + "MarkerDefinitionWithLinks", "MarkerRebuild", "MarkerToggle", "MarkerUpdate", @@ -287,6 +340,7 @@ __all__ = [ "NodeCreate", "NodeDuplicate", "NodeFile", + "NodeStatistics", "NodeUpdate", "Notification", "NotificationAction", @@ -295,9 +349,12 @@ __all__ = [ "Privilege", "Project", "ProjectCompression", + "ProjectCountStatistics", "ProjectCreate", "ProjectDuplicate", "ProjectFile", + "ProjectMarker", + "ProjectStats", "ProjectUpdate", "Qemu", "QemuCreate", @@ -309,6 +366,13 @@ __all__ = [ "QemuUpdate", "RefreshTokenRequest", "RenameSession", + "ReplayFrame", + "ReplayFrameDetail", + "ReplayFrameSource", + "ReplayFrames", + "ReplaySource", + "ReplayTimeline", + "ReplayTreeNode", "Resource", "ResourceCreate", "ResourcePool", @@ -318,11 +382,14 @@ __all__ = [ "RoleCreate", "RoleUpdate", "ServerConfig", + "ServerStatistics", "SettingsResponse", "SettingsUpdate", "SettingsUpdateResponse", "Snapshot", "SnapshotCreate", + "SymbolDimensions", + "SymbolInfo", "Template", "TemplateCreate", "TemplateUpdate", @@ -350,4 +417,6 @@ __all__ = [ "VirtualBoxTemplate", "VirtualBoxTemplateUpdate", "VirtualBoxUpdate", + "WebWiresharkContainer", + "WebWiresharkStatistics", ] diff --git a/gns3server/schemas/controller/computes.py b/gns3server/schemas/controller/computes.py index 5742e4391..229a97660 100644 --- a/gns3server/schemas/controller/computes.py +++ b/gns3server/schemas/controller/computes.py @@ -153,3 +153,11 @@ class AutoIdlePC(BaseModel): model_config = ConfigDict( json_schema_extra={"example": {"platform": "c7200", "image": "/path/to/c7200_image.bin", "ram": 256}} ) + + +class IdlePC(BaseModel): + """ + Result of an Idle-PC computation. + """ + + idlepc: Optional[str] = Field(None, description="Idle-PC value, null when none was found") diff --git a/gns3server/schemas/controller/gns3vm.py b/gns3server/schemas/controller/gns3vm.py index 71a975d43..2c39a5657 100644 --- a/gns3server/schemas/controller/gns3vm.py +++ b/gns3server/schemas/controller/gns3vm.py @@ -55,3 +55,24 @@ class GNS3VM(BaseModel): vcpus: Optional[int] = Field(None, description="Number of CPUs to allocate for the GNS3 VM") ram: Optional[int] = Field(None, description="Amount of memory to allocate for the GNS3 VM") port: Optional[int] = Field(None, gt=0, le=65535) + + +class GNS3VMEngineInfo(BaseModel): + """ + A virtualization engine supported by the GNS3 VM. + """ + + engine_id: str = Field(..., description="Engine identifier") + name: str = Field(..., description="Engine display name") + description: str = Field(..., description="Engine description (may contain HTML)") + support_when_exit: bool = Field(..., description="The engine supports the 'when exit' action") + support_headless: bool = Field(..., description="The engine supports headless mode") + support_ram: bool = Field(..., description="The engine supports RAM and vCPUs allocation") + + +class GNS3VMInfo(BaseModel): + """ + A virtual machine available on a virtualization engine. + """ + + vmname: str = Field(..., description="Virtual machine name") diff --git a/gns3server/schemas/controller/images.py b/gns3server/schemas/controller/images.py index 92ba88775..17eb73ad3 100644 --- a/gns3server/schemas/controller/images.py +++ b/gns3server/schemas/controller/images.py @@ -16,7 +16,8 @@ from datetime import datetime from enum import Enum -from typing import Annotated, Literal, Optional +from typing import Annotated, List, Literal, Optional +from uuid import UUID from pydantic import BaseModel, ConfigDict, Field @@ -100,3 +101,32 @@ class ImageCompatibility(BaseModel): class ImageCompatibilityCatalog(BaseModel): image_sizes: list[int] has_unknown_sizes: bool + + +class ImageInstallCreated(BaseModel): + """ + A template created from an image by the automatic installation. + """ + + template_id: UUID + name: Optional[str] = None + version: Optional[str] = None + template_type: Optional[str] = None + + +class ImageInstallSkipped(BaseModel): + """ + An image or template that the automatic installation skipped. + """ + + name: Optional[str] = None + reason: str + + +class ImageInstallResult(BaseModel): + """ + Result of the automatic template installation from images. + """ + + created: List[ImageInstallCreated] + skipped: List[ImageInstallSkipped] diff --git a/gns3server/schemas/controller/links.py b/gns3server/schemas/controller/links.py index c18547c5e..e1ac4993c 100644 --- a/gns3server/schemas/controller/links.py +++ b/gns3server/schemas/controller/links.py @@ -139,29 +139,6 @@ class LinkFilters(BaseModel): ) -class LinkFilterParameter(BaseModel): - """ - Parameter of a packet filter. - """ - - name: str - type: Literal["int", "text", "str"] - minimum: Optional[int] = None - maximum: Optional[int] = None - unit: Optional[str] = None - - -class LinkFilterDefinition(BaseModel): - """ - Packet filter available on a link. - """ - - type: LinkFilterType - name: str - description: str - parameters: List[LinkFilterParameter] - - class LinkBase(BaseModel): """ Link data. @@ -409,3 +386,81 @@ class MarkerDefinitionCreate(BaseModel): @classmethod def _both_to_none(cls, v): return None if v == "both" else v + + +class Marker(BaseModel): + """ + A traffic-insight marker attached to a link. + """ + + bpf: str = Field(..., description="libpcap BPF expression") + tag: Optional[int] = Field(None, description="Correlation id shared by markers replayed together") + enabled: bool = Field(True, description="Whether the marker is capturing") + color: Optional[str] = Field(None, description="Hex color used by the Web UI") + highlight_duration: Optional[int] = Field(None, description="Web UI highlight duration in milliseconds") + capture_node_id: Optional[str] = Field(None, description="Node whose uBridge hosts the marker") + direction: Optional[str] = Field(None, description="'tx' or 'rx', null for both directions") + data_link_type: Optional[str] = Field(None, description="pcap link-layer type of the capture file") + inherited_from: Optional[str] = Field(None, description="Project marker definition this marker was copied from") + + +class ProjectMarker(Marker): + """ + A marker in the project-wide aggregation, with the link and node it belongs to. + """ + + link_id: str + node_id: Optional[str] = None + + +class MarkerDefinition(BaseModel): + """ + A project-level marker definition inherited by every link. + """ + + bpf: str = Field(..., description="libpcap BPF expression") + tag: Optional[int] = Field(None, description="Correlation id shared by markers replayed together") + direction: Optional[str] = Field(None, description="'tx' or 'rx', null for both directions") + color: Optional[str] = Field(None, description="Hex color used by the Web UI") + highlight_duration: Optional[int] = Field(None, description="Web UI highlight duration in milliseconds") + data_link_type: Optional[str] = Field(None, description="pcap link-layer type of inherited markers") + paused: bool = Field(False, description="Whether the inherited markers are paused") + + +class MarkerDefinitionWithLinks(MarkerDefinition): + """ + A project-level marker definition with the links currently carrying a copy of it. + """ + + link_ids: List[str] = Field(..., description="Links carrying an inherited copy of the definition") + + +class LinkWiresharkRestart(BaseModel): + """ + Result of restarting the Wireshark window of a link capture. + """ + + status: str + + +class LinkFilterParameter(BaseModel): + """ + A parameter of a packet filter. + """ + + name: str + type: str + minimum: Optional[int] = None + maximum: Optional[int] = None + unit: Optional[str] = None + + +class LinkFilterInfo(BaseModel): + """ + A packet filter that can be applied to a link. + """ + + type: str = Field(..., description="Filter identifier used as key in the link 'filters' property") + name: str + description: str + parameters: List[LinkFilterParameter] diff --git a/gns3server/schemas/controller/marker_replay.py b/gns3server/schemas/controller/marker_replay.py new file mode 100644 index 000000000..a07d31cd3 --- /dev/null +++ b/gns3server/schemas/controller/marker_replay.py @@ -0,0 +1,114 @@ +# +# Copyright (C) 2026 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 . + +from typing import List, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class ReplaySource(BaseModel): + """ + A capture source (one marker pcap) of a replay tag. + """ + + node_id: str + link_id: str + marker: str + data_link_type: str + count: int = Field(..., description="Total frames in the capture, unaffected by filters") + + +class ReplayFrame(BaseModel): + """ + One frame of the merged replay timeline. + """ + + ts: str = Field(..., description="Timestamp string to send back unchanged") + len: int = Field(..., description="Captured length in bytes") + node_id: str + link_id: str + marker: str + frame_number: int = Field(..., description="1-based frame number in the source pcap") + src: Optional[str] = None + dst: Optional[str] = None + proto: Optional[str] = None + info: Optional[str] = None + bg: Optional[str] = Field(None, description="Wireshark coloring rule background") + fg: Optional[str] = Field(None, description="Wireshark coloring rule foreground") + + +class ReplayTimeline(BaseModel): + """ + Replay timeline of a tag. + """ + + tag: int + start: Optional[str] = Field(None, description="Timestamp of the first matching frame") + end: Optional[str] = Field(None, description="Timestamp of the last matching frame") + frame_count: int + sources: List[ReplaySource] + frames: List[ReplayFrame] + + +class ReplayFrames(BaseModel): + """ + Frames of a replay time window. + """ + + frames: List[ReplayFrame] + + +class ReplayFrameSource(BaseModel): + """ + The source pcap and frame a decoded frame comes from. + """ + + node_id: str + link_id: str + marker: str + frame_number: int + + +class ReplayTreeNode(BaseModel): + """ + A node of the decoded protocol tree. + + Keys sharkd adds in the future are passed through unchanged. + """ + + model_config = ConfigDict(extra="allow") + + element: Optional[str] = Field(None, description="Node type") + label: Optional[str] = Field(None, description="Display text") + name: Optional[str] = Field(None, description="Field name, e.g. 'ip.ttl'") + filter_expr: Optional[str] = Field(None, description="Display filter with the value baked in") + pos: Optional[int] = Field(None, description="Byte offset for hex highlighting") + size: Optional[int] = Field(None, description="Byte length for hex highlighting") + expert: Optional[str] = Field(None, description="Expert severity") + generated: Optional[bool] = Field(None, description="Generated by Wireshark") + children: Optional[List["ReplayTreeNode"]] = Field(None, description="Nested fields") + + +class ReplayFrameDetail(BaseModel): + """ + One decoded frame. + """ + + ts: str + source: ReplayFrameSource + field_count: int + hex: str = Field(..., description="Raw frame bytes as hex") + tree: List[ReplayTreeNode] diff --git a/gns3server/schemas/controller/projects.py b/gns3server/schemas/controller/projects.py index 1b84b5586..138fd71ea 100644 --- a/gns3server/schemas/controller/projects.py +++ b/gns3server/schemas/controller/projects.py @@ -127,3 +127,15 @@ class ProjectCompression(str, Enum): bzip2 = "bzip2" lzma = "lzma" zstd = "zstd" + + +class ProjectStats(BaseModel): + """ + Project object counts. + """ + + nodes: int + links: int + drawings: int + snapshots: int + markers: int diff --git a/gns3server/schemas/controller/rbac.py b/gns3server/schemas/controller/rbac.py index e02a46180..60ff78e96 100644 --- a/gns3server/schemas/controller/rbac.py +++ b/gns3server/schemas/controller/rbac.py @@ -109,3 +109,13 @@ class Role(DateTimeModelMixin, RoleBase): is_builtin: bool privileges: List[Privilege] model_config = ConfigDict(from_attributes=True) + + +class ACEEndpoint(BaseModel): + """ + An endpoint that can be used in an ACL entry. + """ + + endpoint: str = Field(..., description="Endpoint path to use as the ACE path") + name: str = Field(..., description="Human readable name of the endpoint") + endpoint_type: str = Field(..., description="Kind of resource the endpoint refers to") diff --git a/gns3server/schemas/controller/statistics.py b/gns3server/schemas/controller/statistics.py new file mode 100644 index 000000000..14ec8bc6d --- /dev/null +++ b/gns3server/schemas/controller/statistics.py @@ -0,0 +1,131 @@ +# +# Copyright (C) 2026 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 . + +from typing import Dict, List, Optional, Union + +from pydantic import BaseModel, Field + + +class ComputeStatistics(BaseModel): + """ + Resource usage reported by a compute. + + Fields come from the compute itself, so a compute running another build + may omit some of them. + """ + + memory_total: Optional[int] = None + memory_free: Optional[int] = None + memory_used: Optional[int] = None + swap_total: Optional[int] = None + swap_free: Optional[int] = None + swap_used: Optional[int] = None + cpu_usage_percent: Optional[int] = None + cpu_count: Optional[int] = None + cpu_count_physical: Optional[int] = None + cpu_model: Optional[str] = None + memory_usage_percent: Optional[int] = None + swap_usage_percent: Optional[int] = None + disk_usage_percent: Optional[int] = None + disk_total: Optional[int] = None + disk_used: Optional[int] = None + disk_free: Optional[int] = None + load_average: Optional[List[float]] = None + load_average_percent: Optional[List[float]] = None + + +class ComputeStatisticsEntry(BaseModel): + """ + Statistics of one compute. + """ + + compute_id: str + compute_name: str + statistics: ComputeStatistics + + +class ProjectCountStatistics(BaseModel): + """ + Project counts by status. + """ + + total: int + opened: int + closed: int + + +class NodeStatistics(BaseModel): + """ + Node counts across all projects. + """ + + total: int + open_project_nodes: int + closed_project_nodes: int + by_type: Dict[str, int] = Field(..., description="Node count by node type") + by_status: Dict[str, int] = Field(..., description="Node count by status, open projects only") + + +class LinkStatistics(BaseModel): + """ + Link counts across all projects. + """ + + total: int + capturing: int + + +class WebWiresharkContainer(BaseModel): + """ + A Web Wireshark container. + """ + + project_id: str + project_name: str + container_id: str + status: str + running: bool + active_sessions: Optional[int] = None + memory_limit: Optional[str] = None + cpu_limit: Optional[str] = None + pids_limit: Optional[Union[int, str]] = Field(None, description="Process limit, or 'unlimited'") + memory: Optional[str] = Field(None, description="Current memory usage") + cpu: Optional[str] = Field(None, description="Current CPU usage") + pids: Optional[int] = Field(None, description="Current process count") + + +class WebWiresharkStatistics(BaseModel): + """ + Web Wireshark container statistics across all opened projects. + """ + + total_containers: int + running_containers: int + active_sessions: int + containers: List[WebWiresharkContainer] + + +class ServerStatistics(BaseModel): + """ + Server statistics. + """ + + uptime_seconds: int + computes: List[ComputeStatisticsEntry] + projects: ProjectCountStatistics + nodes: NodeStatistics + links: LinkStatistics + webwireshark: WebWiresharkStatistics diff --git a/gns3server/schemas/controller/symbols.py b/gns3server/schemas/controller/symbols.py new file mode 100644 index 000000000..75f9e9f31 --- /dev/null +++ b/gns3server/schemas/controller/symbols.py @@ -0,0 +1,37 @@ +# +# Copyright (C) 2026 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 . + +from pydantic import BaseModel + + +class SymbolInfo(BaseModel): + """ + A symbol available on the controller. + """ + + symbol_id: str + filename: str + theme: str + builtin: bool + + +class SymbolDimensions(BaseModel): + """ + Symbol dimensions in pixels. + """ + + width: int + height: int diff --git a/gns3server/schemas/controller/tokens.py b/gns3server/schemas/controller/tokens.py index 1ecd6545a..7b8841b55 100644 --- a/gns3server/schemas/controller/tokens.py +++ b/gns3server/schemas/controller/tokens.py @@ -15,6 +15,7 @@ # along with this program. If not, see . from typing import Optional +from uuid import UUID from pydantic import BaseModel @@ -43,3 +44,30 @@ class ApiKeyCreate(BaseModel): """Schema for creating a new API key.""" name: str + + +class ApiKey(BaseModel): + """API key metadata. The secret is never returned after creation.""" + + api_key_id: UUID + name: str + key_prefix: str + created_at: Optional[str] = None + last_used_at: Optional[str] = None + revoked: bool + + +class ApiKeyCreated(BaseModel): + """A newly created API key. The full key is returned only once.""" + + api_key_id: UUID + api_key: str + name: str + key_prefix: str + created_at: Optional[str] = None + + +class ApiKeyMessage(BaseModel): + """Confirmation message returned when an API key is revoked or restored.""" + + message: str diff --git a/gns3server/utils/drawings.py b/gns3server/utils/drawings.py new file mode 100644 index 000000000..54303a843 --- /dev/null +++ b/gns3server/utils/drawings.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python +# +# Copyright (C) 2026 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 . + +""" +Web UI compatible drawing SVG assembly. + +The GNS3 Web UI does not render drawing SVG as-is: it re-parses it into its own +element model with ONE recognized child element per drawing (text, image, rect, +line, ellipse or path) and reads the root width/height for the selection box. +These helpers emit exactly that format, mirroring the attribute sets the Web UI +itself serializes, and validate inputs so callers fail loudly instead of hitting +the controller's silent drop of unparseable SVG strings of 500+ characters. + +Shared by the MCP tools (agent/mcp) and the GNS3 copilot tools (tools_v2). +""" + +import re +from math import ceil +from typing import Any +from xml.etree import ElementTree as ET +from xml.sax.saxutils import escape + +# The Web UI renders font-size as pt (1pt = 4/3 px) and monospace glyphs advance +# ~0.6em, so one character is ~0.8 x font_size in canvas pixels. Proportional +# fonts get a wider worst-case factor so the selection box does not clip the text. +_MONOSPACE_CHAR_FACTOR = 0.8 +_PROPORTIONAL_CHAR_FACTOR = 1.0 +# The Web UI stacks lines at dy=1.4em, i.e. 1.4 x 4/3 ~ 1.87 px per pt of font size. +_LINE_HEIGHT_FACTOR = 1.9 +_MAX_TEXT_LENGTH = 500 + +_COLOR_RE = re.compile(r"^#(?:[0-9A-Fa-f]{3,4}|[0-9A-Fa-f]{6}|[0-9A-Fa-f]{8})$") +_FONT_FAMILY_RE = re.compile(r"^[A-Za-z0-9 ,.'-]+$") +_DASHARRAY_RE = re.compile(r"^[0-9][0-9, ]*$") +# The Web UI remaps these Qt-style dash patterns to different values +# (QtDasharrayFixer); reject them so what you send is what renders. +_QT_REMAPPED_DASHARRAYS = {"25, 25", "5, 25", "5, 25, 25", "25, 25, 5, 25, 5"} +_DEFAULT_DASHARRAY = "10,6" + + +def _checked_text(text: Any) -> str: + if text is None: + raise ValueError("text is required") + if not isinstance(text, str): + raise ValueError("text must be a string") + text = text.replace("\r\n", "\n").replace("\r", "\n").strip("\n") + if not text: + raise ValueError("text is required") + if len(text) > _MAX_TEXT_LENGTH: + raise ValueError(f"text is too long (max {_MAX_TEXT_LENGTH} characters)") + if any(ord(ch) < 32 and ch != "\n" for ch in text) or "\x7f" in text: + raise ValueError("text contains control characters (only newlines are allowed)") + return text + + +def _checked_font_size(font_size: Any) -> int: + if isinstance(font_size, bool) or not isinstance(font_size, int) or not 1 <= font_size <= 200: + raise ValueError("font_size must be an integer between 1 and 200") + return font_size + + +def _checked_font_family(font_family: Any) -> str: + if not isinstance(font_family, str) or not _FONT_FAMILY_RE.match(font_family): + raise ValueError("font_family contains invalid characters (allowed: letters, digits, spaces, . ' -)") + return font_family + + +def _checked_color(value: Any, name: str) -> str: + if not isinstance(value, str) or not _COLOR_RE.match(value): + raise ValueError(f"{name} must be a hex value like #RRGGBB") + return value + + +def _checked_dimension(value: Any, name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 2000: + raise ValueError(f"{name} must be an integer between 1 and 2000") + return value + + +def _checked_radius(value: Any, max_radius: int) -> int: + if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= max_radius: + raise ValueError(f"rx must be an integer between 0 and {max_radius}") + return value + + +def _checked_style( + fill: Any, fill_opacity: Any, stroke: Any, stroke_width: Any, dashed: Any, dasharray: Any +) -> tuple[str, Any, str | None, int, str | None]: + """ + Validate the style block shared by rectangle and ellipse drawings. + + :returns: (fill, fill_opacity, stroke, stroke_width, dasharray-or-None) + """ + + if fill != "none" and (not isinstance(fill, str) or not _COLOR_RE.match(fill)): + raise ValueError('fill must be a hex value like #RRGGBB or "none"') + if isinstance(fill_opacity, bool) or not isinstance(fill_opacity, (int, float)) or not 0 <= fill_opacity <= 1: + raise ValueError("fill_opacity must be a number between 0 and 1") + + dash = dasharray if dasharray is not None else (_DEFAULT_DASHARRAY if dashed else None) + if dash is not None: + if not isinstance(dash, str) or not _DASHARRAY_RE.match(dash) or dash in _QT_REMAPPED_DASHARRAYS: + raise ValueError( + 'dasharray must be plain "on,off" number pairs and must not be one of ' + f"the Qt-remapped patterns {sorted(_QT_REMAPPED_DASHARRAYS)}" + ) + if stroke is None: + raise ValueError("stroke is required when dasharray is set") + + if stroke is not None: + stroke = _checked_color(stroke, "stroke") + if isinstance(stroke_width, bool) or not isinstance(stroke_width, int) or not 1 <= stroke_width <= 20: + raise ValueError("stroke_width must be an integer between 1 and 20") + + return fill, fill_opacity, stroke, stroke_width, dash + + +def _style_attrs(fill: str, fill_opacity: Any, stroke: str | None, stroke_width: int, dash: str | None) -> str: + """Render the validated style block as SVG attributes (Web UI attribute order).""" + + attrs = f'fill="{fill}" fill-opacity="{fill_opacity}"' + if stroke is not None: + attrs += f' stroke="{stroke}" stroke-width="{stroke_width}"' + if dash is not None: + attrs += f' stroke-dasharray="{dash}"' + return attrs + + +def build_text_svg( + text: Any, font_size: int = 13, color: str = "#000000", bold: bool = False, font_family: str = "monospace" +) -> tuple[str, int, int, int]: + """ + Build a Web UI compatible single SVG for a canvas label. + + The output mirrors what the Web UI itself serializes for text drawings: + root with one child and no x/y (the Web UI parser + ignores them and positions the text itself). + + :returns: (svg, width, height, line_height) with the computed drawing size + """ + + text = _checked_text(text) + font_size = _checked_font_size(font_size) + color = _checked_color(color, "color") + font_family = _checked_font_family(font_family) + + lines = text.split("\n") + factor = _MONOSPACE_CHAR_FACTOR if "monospace" in font_family.lower() else _PROPORTIONAL_CHAR_FACTOR + width = max(1, ceil(max(len(line) for line in lines) * font_size * factor)) + line_height = ceil(font_size * _LINE_HEIGHT_FACTOR) + height = len(lines) * line_height + weight = "bold" if bold else "normal" + svg = ( + f'' + f'{escape(text)}' + ) + ET.fromstring(svg) + return svg, width, height, line_height + + +def build_rectangle_svg( + width: int, + height: int, + fill: str = "#FFFFFF", + fill_opacity: float = 1.0, + stroke: str | None = None, + stroke_width: int = 1, + dashed: bool = False, + dasharray: str | None = None, + rx: int = 0, +) -> tuple[str, int, int]: + """ + Build a Web UI compatible single SVG. + + The rect sits at the drawing origin (the Web UI ignores rect x/y), so the + caller positions the drawing itself and the root size equals the rect size. + + :returns: (svg, width, height) + """ + + width = _checked_dimension(width, "width") + height = _checked_dimension(height, "height") + rx = _checked_radius(rx, min(width, height) // 2) + fill, fill_opacity, stroke, stroke_width, dash = _checked_style( + fill, fill_opacity, stroke, stroke_width, dashed, dasharray + ) + + svg = ( + f'' + f"' + ) + ET.fromstring(svg) + return svg, width, height + + +def build_ellipse_svg( + width: int, + height: int, + fill: str = "#FFFFFF", + fill_opacity: float = 1.0, + stroke: str | None = None, + stroke_width: int = 1, + dashed: bool = False, + dasharray: str | None = None, +) -> tuple[str, int, int]: + """ + Build a Web UI compatible single SVG from a bounding box. + + The radii and center are derived (rx = width // 2, cx = rx) because the Web UI + treats cx/cy as offsets inside the drawing box, not canvas coordinates. For + odd width/height the drawing is 1 px smaller than requested. + + :returns: (svg, width, height) with the actual (possibly rounded-down) size + """ + + width = _checked_dimension(width, "width") + height = _checked_dimension(height, "height") + rx, ry = width // 2, height // 2 + if rx < 1 or ry < 1: + raise ValueError("width and height must be at least 2") + fill, fill_opacity, stroke, stroke_width, dash = _checked_style( + fill, fill_opacity, stroke, stroke_width, dashed, dasharray + ) + + svg = ( + f'' + f"' + ) + ET.fromstring(svg) + return svg, 2 * rx, 2 * ry diff --git a/pyproject.toml b/pyproject.toml index 5ff415217..44df93d8d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -211,6 +211,8 @@ extend-immutable-calls = [ "gns3server/db/tasks.py" = ["RUF006", "RUF019"] "gns3server/db_migrations/versions/ec4b7b198555_convert_pickletype_fields_to_json.py" = ["S301", "S608"] "gns3server/logger.py" = ["PLW0127"] +# Parses self-generated SVG (validated inputs) like controller/drawing.py does. +"gns3server/utils/drawings.py" = ["S314"] "gns3server/schemas/controller/settings.py" = ["S105"] "gns3server/schemas/controller/templates/docker_templates.py" = ["W292"] "gns3server/schemas/controller/tokens.py" = ["S105"] @@ -234,7 +236,9 @@ extend-immutable-calls = [ "scripts/extract_mermaid.py" = ["C901", "PLW1510", "S603", "S607"] "scripts/tag.py" = ["S603", "S607"] "scripts/welcome.py" = ["C901", "E722", "PLW1510", "S108", "S112", "S603", "S605", "S606", "S607"] +"tests/agent/mcp/test_drawing_handlers.py" = ["S314"] "tests/agent/mcp/test_handlers.py" = ["B905"] +"tests/utils/test_drawings.py" = ["S314"] "tests/agent/mcp/test_tool_params.py" = ["C901"] "tests/api/test_openapi_spec.py" = ["S603"] "tests/api/routes/controller/test_images.py" = ["S324", "W292"] diff --git a/tests/agent/mcp/test_drawing_handlers.py b/tests/agent/mcp/test_drawing_handlers.py new file mode 100644 index 000000000..5a4498830 --- /dev/null +++ b/tests/agent/mcp/test_drawing_handlers.py @@ -0,0 +1,153 @@ +""" +MCP drawing handler tests: text/rectangle/ellipse creation via the shared +api_handlers layer (SVG building units live in tests/utils/test_drawings.py). +""" + +from unittest.mock import MagicMock, patch +from xml.etree import ElementTree as ET + +import pytest + +from gns3server.agent.gns3_copilot.gns3_client.api_handlers import ( + create_ellipse_drawing_handler, + create_rectangle_drawing_handler, + create_text_drawing_handler, +) + +AH = "gns3server.agent.gns3_copilot.gns3_client.api_handlers" + + +def _mock_conn(json_result=None): + """Create a mocked Gns3Connector with base_url and http_call.""" + conn = MagicMock() + conn.base_url = "http://192.168.1.3:3080/v3" + conn.http_call.return_value.json.return_value = json_result or {"status": "ok"} + return conn + + +@pytest.fixture +def ctx(): + return {"server_url": "http://192.168.1.3:3080", "jwt_token": "token", "jwt_username": "admin"} + + +class TestCreateTextDrawingHandler: + def test_success(self, ctx): + with patch(f"{AH}._get_connector") as m: + conn = _mock_conn({"drawing_id": "d1", "x": 100, "y": 80}) + m.return_value = conn + result = create_text_drawing_handler( + {"project_id": "p1", "text": "R1", "x": 100, "y": 80, "bold": True}, ctx + ) + assert result["drawing"]["drawing_id"] == "d1" + assert result["width"] == 21 + assert result["height"] == 25 + # z defaults to 1 (above links) + body = conn.http_call.call_args.kwargs["json_data"] + assert body["z"] == 1 + assert body["x"] == 100 + assert 'font-weight="bold"' in body["svg"] + + def test_missing_project_id(self, ctx): + assert "error" in create_text_drawing_handler({"text": "R1"}, ctx) + + def test_missing_text(self, ctx): + assert "error" in create_text_drawing_handler({"project_id": "p1"}, ctx) + + def test_blank_text(self, ctx): + assert "error" in create_text_drawing_handler({"project_id": "p1", "text": "\n\n"}, ctx) + + def test_newline_normalization(self, ctx): + with patch(f"{AH}._get_connector") as m: + conn = _mock_conn({"drawing_id": "d1"}) + m.return_value = conn + create_text_drawing_handler({"project_id": "p1", "text": "a\r\nb"}, ctx) + svg = conn.http_call.call_args.kwargs["json_data"]["svg"] + assert "\r" not in svg + + @pytest.mark.parametrize("font_size", [0, 201, "13", 13.5, True]) + def test_bad_font_size(self, ctx, font_size): + result = create_text_drawing_handler({"project_id": "p1", "text": "R1", "font_size": font_size}, ctx) + assert "error" in result + + @pytest.mark.parametrize("color", ["red", "123456", "#GGGGGG", "#12345", 'black" onload="x']) + def test_bad_color(self, ctx, color): + result = create_text_drawing_handler({"project_id": "p1", "text": "R1", "color": color}, ctx) + assert "error" in result + + @pytest.mark.parametrize("font_family", ['mono" onload="x', "sans", "a&b", ""]) + def test_bad_font_family(self, ctx, font_family): + result = create_text_drawing_handler({"project_id": "p1", "text": "R1", "font_family": font_family}, ctx) + assert "error" in result + + def test_xml_entities_survive_round_trip(self, ctx): + with patch(f"{AH}._get_connector") as m: + conn = _mock_conn({"drawing_id": "d1"}) + m.return_value = conn + create_text_drawing_handler({"project_id": "p1", "text": "R1 & "}, ctx) + svg = conn.http_call.call_args.kwargs["json_data"]["svg"] + assert "R1 & <backup>" in svg + ET.fromstring(svg) + + +class TestCreateRectangleDrawingHandler: + def test_success(self, ctx): + with patch(f"{AH}._get_connector") as m: + conn = _mock_conn({"drawing_id": "d2"}) + m.return_value = conn + result = create_rectangle_drawing_handler( + {"project_id": "p1", "x": 200, "y": 150, "width": 80, "height": 50, "fill": "#4A90D9"}, ctx + ) + assert result["drawing"]["drawing_id"] == "d2" + assert result["width"] == 80 + assert result["height"] == 50 + body = conn.http_call.call_args.kwargs["json_data"] + # z defaults to 0 (shapes behind text and nodes) + assert body["z"] == 0 + assert body["svg"] == ( + '' + '' + ) + + def test_missing_dimensions(self, ctx): + assert "error" in create_rectangle_drawing_handler({"project_id": "p1"}, ctx) + + def test_bad_fill_returns_error(self, ctx): + result = create_rectangle_drawing_handler({"project_id": "p1", "width": 80, "height": 50, "fill": "red"}, ctx) + assert "error" in result + + def test_qt_remapped_dasharray_rejected(self, ctx): + result = create_rectangle_drawing_handler( + {"project_id": "p1", "width": 80, "height": 50, "stroke": "#333333", "dasharray": "25, 25"}, ctx + ) + assert "error" in result + + +class TestCreateEllipseDrawingHandler: + def test_success(self, ctx): + with patch(f"{AH}._get_connector") as m: + conn = _mock_conn({"drawing_id": "d3"}) + m.return_value = conn + result = create_ellipse_drawing_handler( + {"project_id": "p1", "x": 300, "y": 300, "width": 160, "height": 160, "fill": "#5AA9DD"}, ctx + ) + assert result["drawing"]["drawing_id"] == "d3" + body = conn.http_call.call_args.kwargs["json_data"] + assert body["z"] == 0 + assert body["svg"] == ( + '' + '' + ) + + def test_odd_dimensions_report_actual_size(self, ctx): + with patch(f"{AH}._get_connector") as m: + conn = _mock_conn({"drawing_id": "d3"}) + m.return_value = conn + result = create_ellipse_drawing_handler({"project_id": "p1", "width": 101, "height": 61}, ctx) + assert (result["width"], result["height"]) == (100, 60) + + def test_too_small(self, ctx): + result = create_ellipse_drawing_handler({"project_id": "p1", "width": 1, "height": 60}, ctx) + assert "error" in result + + def test_missing_dimensions(self, ctx): + assert "error" in create_ellipse_drawing_handler({"project_id": "p1", "width": 100}, ctx) diff --git a/tests/agent/mcp/test_tool_params.py b/tests/agent/mcp/test_tool_params.py index e2287f8e1..b4c01d910 100644 --- a/tests/agent/mcp/test_tool_params.py +++ b/tests/agent/mcp/test_tool_params.py @@ -67,6 +67,9 @@ HANDLER_FILES = { "download_capture_file_handler": API_HANDLERS_FILE, "link_marker_handler": API_HANDLERS_FILE, "marker_definition_handler": API_HANDLERS_FILE, + "create_text_drawing_handler": API_HANDLERS_FILE, + "create_rectangle_drawing_handler": API_HANDLERS_FILE, + "create_ellipse_drawing_handler": API_HANDLERS_FILE, "list_templates_handler": "templates.py", "get_template_handler": "templates.py", "create_template_handler": "templates.py", @@ -80,7 +83,6 @@ HANDLER_FILES = { "delete_snapshot_handler": "snapshots.py", "restore_snapshot_handler": "snapshots.py", "get_drawings_handler": "drawings.py", - "create_drawing_handler": "drawings.py", "get_drawing_handler": "drawings.py", "update_drawing_handler": "drawings.py", "delete_drawing_handler": "drawings.py", diff --git a/tests/api/routes/controller/test_appliances.py b/tests/api/routes/controller/test_appliances.py index 193557e35..e2b853f5f 100644 --- a/tests/api/routes/controller/test_appliances.py +++ b/tests/api/routes/controller/test_appliances.py @@ -105,7 +105,7 @@ class TestApplianceRoutes: async def test_add_v8_version_appliance(self, app: FastAPI, client: AsyncClient) -> None: appliance_id = "1cfdf900-7c30-4cb7-8f03-3f61d2581633" # Empty VM appliance - new_version = {"name": "v8-99G", "category": "guest", "images": {"hda_disk_image": "empty99G.qcow2"}} + new_version = {"name": "v8-99G", "images": {"hda_disk_image": "empty99G.qcow2"}} response = await client.post( app.url_path_for("add_appliance_version", appliance_id=appliance_id), json=new_version ) diff --git a/tests/api/routes/controller/test_links.py b/tests/api/routes/controller/test_links.py index 54830923b..14e5b9699 100644 --- a/tests/api/routes/controller/test_links.py +++ b/tests/api/routes/controller/test_links.py @@ -648,12 +648,12 @@ class TestLinkRoutes: assert "LinkFilters" not in schemas filters_property = schemas["Link"]["properties"]["filters"] assert filters_property["anyOf"][0] == {"additionalProperties": True, "type": "object"} - assert schemas["LinkFilterDefinition"]["properties"]["parameters"]["items"] == { + assert schemas["LinkFilterInfo"]["properties"]["parameters"]["items"] == { "$ref": "#/components/schemas/LinkFilterParameter" } path = "/v3/projects/{project_id}/links/{link_id}/available_filters" responses = app.openapi()["paths"][path]["get"]["responses"]["200"]["content"]["application/json"]["schema"] - assert responses["items"] == {"$ref": "#/components/schemas/LinkFilterDefinition"} + assert responses["items"] == {"$ref": "#/components/schemas/LinkFilterInfo"} async def test_get_udp_interface(self, app: FastAPI, client: AsyncClient, project: Project) -> None: """ diff --git a/tests/api/routes/controller/test_markers.py b/tests/api/routes/controller/test_markers.py index b53ec8533..bc4fecf30 100644 --- a/tests/api/routes/controller/test_markers.py +++ b/tests/api/routes/controller/test_markers.py @@ -19,6 +19,8 @@ HTTP-route tests for the traffic-insight marker endpoints: per-link markers, project-level definitions, and the project-wide aggregation view. """ +from unittest.mock import DEFAULT + import pytest from fastapi import FastAPI, status from httpx import AsyncClient @@ -43,6 +45,25 @@ def _inherited(link, name="arp"): } +def _record_marker(link): + """Side effect for a mocked ``start_marker`` that stores the marker like the real one does.""" + + def side_effect(*, name, bpf, tag=None, color=None, highlight_duration=None, direction=None, **kwargs): + link._markers[name] = { + "bpf": bpf, + "tag": tag, + "enabled": True, + "color": color, + "highlight_duration": highlight_duration, + "capture_node_id": "node-id", + "direction": direction, + "data_link_type": kwargs.get("data_link_type", "DLT_EN10MB"), + } + return DEFAULT + + return side_effect + + class TestMarkerRoutes: # ----------------------------------------------------------------------- # Per-link markers @@ -53,13 +74,25 @@ class TestMarkerRoutes: link = UDPLink(project) project._links = {link.id: link} - with asyncio_patch("gns3server.controller.udp_link.UDPLink.start_marker") as mock: + with asyncio_patch( + "gns3server.controller.udp_link.UDPLink.start_marker", side_effect=_record_marker(link) + ) as mock: response = await client.post( app.url_path_for("create_marker", project_id=project.id, link_id=link.id), json={"bpf": "icmp", "tag": 3, "color": "#ff5722", "highlight_duration": 800}, ) assert response.status_code == status.HTTP_201_CREATED + assert response.json() == { + "bpf": "icmp", + "tag": 3, + "enabled": True, + "color": "#ff5722", + "highlight_duration": 800, + "capture_node_id": "node-id", + "direction": None, + "data_link_type": "DLT_EN10MB", + } mock.assert_called_once() _, kwargs = mock.call_args assert kwargs["bpf"] == "icmp" @@ -73,12 +106,15 @@ class TestMarkerRoutes: link = UDPLink(project) project._links = {link.id: link} - with asyncio_patch("gns3server.controller.udp_link.UDPLink.start_marker") as mock: + with asyncio_patch( + "gns3server.controller.udp_link.UDPLink.start_marker", side_effect=_record_marker(link) + ) as mock: response = await client.post( app.url_path_for("create_marker", project_id=project.id, link_id=link.id), json={"name": "web", "bpf": "tcp port 80"}, ) assert response.status_code == status.HTTP_201_CREATED + assert response.json()["bpf"] == "tcp port 80" _, kwargs = mock.call_args assert kwargs["name"] == "web" @@ -141,6 +177,16 @@ class TestMarkerRoutes: async def test_update_marker(self, app: FastAPI, client: AsyncClient, project: Project) -> None: link = UDPLink(project) + link._markers["web"] = { + "bpf": "udp port 53", + "tag": None, + "enabled": True, + "color": None, + "highlight_duration": 1500, + "capture_node_id": "node-id", + "direction": None, + "data_link_type": "DLT_EN10MB", + } project._links = {link.id: link} with asyncio_patch("gns3server.controller.udp_link.UDPLink.update_marker") as mock: @@ -149,6 +195,7 @@ class TestMarkerRoutes: json={"bpf": "udp port 53", "highlight_duration": 1500}, ) assert response.status_code == status.HTTP_200_OK + assert response.json() == link._markers["web"] _, kwargs = mock.call_args assert kwargs["bpf"] == "udp port 53" assert kwargs["highlight_duration"] == 1500 @@ -195,12 +242,27 @@ class TestMarkerRoutes: async def test_create_marker_definition(self, app: FastAPI, client: AsyncClient, project: Project) -> None: - with asyncio_patch("gns3server.controller.project.Project.create_marker_definition") as mock: + def store_definition(*, name, bpf, highlight_duration=None, **kwargs): + project._marker_definitions[name] = { + "bpf": bpf, + "tag": None, + "direction": None, + "color": None, + "highlight_duration": highlight_duration, + "data_link_type": "DLT_EN10MB", + "paused": False, + } + return DEFAULT + + with asyncio_patch( + "gns3server.controller.project.Project.create_marker_definition", side_effect=store_definition + ) as mock: response = await client.post( app.url_path_for("create_marker_definition", project_id=project.id), json={"name": "arp", "bpf": "arp", "highlight_duration": 1200}, ) assert response.status_code == status.HTTP_201_CREATED + assert response.json() == project._marker_definitions["arp"] _, kwargs = mock.call_args assert kwargs["name"] == "arp" assert kwargs["bpf"] == "arp" @@ -220,12 +282,15 @@ class TestMarkerRoutes: async def test_update_marker_definition(self, app: FastAPI, client: AsyncClient, project: Project) -> None: + project._marker_definitions["arp"] = {"bpf": "arp or rarp", "tag": None, "highlight_duration": 900} + with asyncio_patch("gns3server.controller.project.Project.update_marker_definition") as mock: response = await client.put( app.url_path_for("update_marker_definition", project_id=project.id, def_name="arp"), json={"bpf": "arp or rarp", "highlight_duration": 900}, ) assert response.status_code == status.HTTP_200_OK + assert response.json() == {"bpf": "arp or rarp", "tag": None, "highlight_duration": 900} _, kwargs = mock.call_args assert kwargs["bpf"] == "arp or rarp" assert kwargs["highlight_duration"] == 900 diff --git a/tests/api/routes/controller/test_nodes.py b/tests/api/routes/controller/test_nodes.py index be05b9cf3..facbf5e6e 100644 --- a/tests/api/routes/controller/test_nodes.py +++ b/tests/api/routes/controller/test_nodes.py @@ -486,7 +486,7 @@ class TestNodeRoutes: response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=node.id)) assert response.json()["status"] == "started" - async def test_start_node_reports_starting_while_in_flight( + async def test_start_node_keeps_real_status_while_in_flight( self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node ) -> None: @@ -498,10 +498,11 @@ class TestNodeRoutes: compute.post = post response = await client.post(app.url_path_for("start_node", project_id=project.id, node_id=node.id), json={}) assert response.status_code == status.HTTP_200_OK - assert seen == ["starting"] + # no transitional status is reported while the call is in flight + assert seen == ["stopped"] assert response.json()["status"] == "started" - async def test_stop_node_reports_stopping_while_in_flight( + async def test_stop_node_keeps_real_status_while_in_flight( self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute, node: Node ) -> None: @@ -514,7 +515,8 @@ class TestNodeRoutes: node._status = "started" response = await client.post(app.url_path_for("stop_node", project_id=project.id, node_id=node.id)) assert response.status_code == status.HTTP_200_OK - assert seen == ["stopping"] + # no transitional status is reported while the call is in flight + assert seen == ["started"] assert response.json()["status"] == "stopped" async def test_start_node_wait_timeout( diff --git a/tests/api/routes/controller/test_response_models.py b/tests/api/routes/controller/test_response_models.py new file mode 100644 index 000000000..9ea0f6dda --- /dev/null +++ b/tests/api/routes/controller/test_response_models.py @@ -0,0 +1,464 @@ +# +# Copyright (C) 2026 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 . + +""" +Runtime JSON shape of the endpoints that used to return untyped objects, and +the OpenAPI contract that now describes them. +""" + +import uuid +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import FastAPI, status +from httpx import AsyncClient + +from gns3server.controller import Controller, marker_replay +from gns3server.controller.link import Link +from gns3server.controller.project import Project +from gns3server.controller.udp_link import UDPLink +from tests.controller.test_marker_replay import _write_pcap + +pytestmark = pytest.mark.asyncio + +TYPED_OPERATIONS = [ + ("post", "/v3/access/api-keys"), + ("get", "/v3/access/api-keys"), + ("post", "/v3/access/api-keys/{api_key_id}/revoke"), + ("post", "/v3/access/api-keys/{api_key_id}/restore"), + ("get", "/v3/statistics"), + ("get", "/v3/projects/{project_id}/stats"), + ("post", "/v3/images/install"), + ("post", "/v3/appliances/{appliance_id}/version"), + ("get", "/v3/projects/{project_id}/markers"), + ("get", "/v3/projects/{project_id}/marker-definitions"), + ("post", "/v3/projects/{project_id}/marker-definitions"), + ("put", "/v3/projects/{project_id}/marker-definitions/{def_name}"), + ("get", "/v3/projects/{project_id}/markers/tags/{tag}/replay/range"), + ("get", "/v3/projects/{project_id}/markers/tags/{tag}/replay/frames"), + ("get", "/v3/projects/{project_id}/markers/tags/{tag}/replay/frame/detail"), + ("get", "/v3/projects/{project_id}/links/{link_id}/markers"), + ("post", "/v3/projects/{project_id}/links/{link_id}/markers"), + ("put", "/v3/projects/{project_id}/links/{link_id}/markers/{marker_name}"), + ("get", "/v3/symbols"), + ("get", "/v3/symbols/{symbol_id}/dimensions"), + ("get", "/v3/symbols/default_symbols"), + ("get", "/v3/projects/{project_id}/links/{link_id}/available_filters"), + ("post", "/v3/projects/{project_id}/links/{link_id}/capture/wireshark/restart"), + ("get", "/v3/projects/{project_id}/nodes/{node_id}/dynamips/auto_idlepc"), + ("post", "/v3/computes/{compute_id}/dynamips/auto_idlepc"), + ("get", "/v3/gns3vm/engines"), + ("get", "/v3/gns3vm/engines/{engine}/vms"), + ("get", "/v3/access/acl/endpoints"), +] + + +def _is_untyped(schema: dict) -> bool: + if not schema: + return True + if "$ref" in schema: + return False + if schema.get("type") == "array": + return _is_untyped(schema.get("items", {})) + if schema.get("type") == "object": + extra = schema.get("additionalProperties") + if "properties" in schema: + return False + return extra in (None, True, {}) + for key in ("anyOf", "oneOf"): + if key in schema: + return any(_is_untyped(s) for s in schema[key] if s.get("type") != "null") + return False + + +class TestOpenAPIContract: + async def test_no_untyped_success_response(self, app: FastAPI, client: AsyncClient) -> None: + + response = await client.get("/openapi.json") + assert response.status_code == status.HTTP_200_OK + paths = response.json()["paths"] + + untyped = [] + for method, path in TYPED_OPERATIONS: + operation = paths[path][method] + success = [r for code, r in operation["responses"].items() if code.startswith("2") and code != "204"] + assert success, f"{method.upper()} {path} declares no success response" + for r in success: + schema = r["content"]["application/json"]["schema"] + if _is_untyped(schema): + untyped.append(f"{method.upper()} {path}") + assert not untyped + + +class TestApiKeys: + async def _create(self, app: FastAPI, client: AsyncClient, name: str = "ci") -> dict: + + response = await client.post(app.url_path_for("create_api_key"), json={"name": name}) + assert response.status_code == status.HTTP_201_CREATED + return response.json() + + async def test_create(self, app: FastAPI, client: AsyncClient) -> None: + + body = await self._create(app, client, "created") + assert set(body) == {"api_key_id", "api_key", "name", "key_prefix", "created_at"} + assert body["name"] == "created" + assert body["api_key"].startswith(f"gns3_{body['api_key_id']}_") + assert body["key_prefix"] == body["api_key"][:8] + assert isinstance(body["created_at"], str) + uuid.UUID(body["api_key_id"]) + + async def test_list(self, app: FastAPI, client: AsyncClient) -> None: + + created = await self._create(app, client, "listed") + response = await client.get(app.url_path_for("list_api_keys")) + assert response.status_code == status.HTTP_200_OK + entry = next(k for k in response.json() if k["api_key_id"] == created["api_key_id"]) + assert set(entry) == {"api_key_id", "name", "key_prefix", "created_at", "last_used_at", "revoked"} + assert entry["name"] == "listed" + assert entry["last_used_at"] is None + assert entry["revoked"] is False + assert entry["created_at"] == created["created_at"] + + async def test_revoke_and_restore(self, app: FastAPI, client: AsyncClient) -> None: + + created = await self._create(app, client, "toggled") + key_id = created["api_key_id"] + + response = await client.post(app.url_path_for("revoke_api_key", api_key_id=key_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"message": "API key 'toggled' revoked"} + keys = (await client.get(app.url_path_for("list_api_keys"))).json() + assert next(k for k in keys if k["api_key_id"] == key_id)["revoked"] is True + + response = await client.post(app.url_path_for("restore_api_key", api_key_id=key_id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"message": "API key 'toggled' restored"} + keys = (await client.get(app.url_path_for("list_api_keys"))).json() + assert next(k for k in keys if k["api_key_id"] == key_id)["revoked"] is False + + async def test_revoke_unknown_key(self, app: FastAPI, client: AsyncClient) -> None: + + response = await client.post(app.url_path_for("revoke_api_key", api_key_id=str(uuid.uuid4()))) + assert response.status_code == status.HTTP_404_NOT_FOUND + + +COMPUTE_STATISTICS = { + "memory_total": 8000, + "memory_free": 4000, + "memory_used": 4000, + "swap_total": 1000, + "swap_free": 900, + "swap_used": 100, + "cpu_usage_percent": 12, + "cpu_count": 4, + "cpu_count_physical": None, + "cpu_model": "Test CPU", + "memory_usage_percent": 50, + "swap_usage_percent": 10, + "disk_usage_percent": 30, + "disk_total": 100000, + "disk_used": 30000, + "disk_free": 70000, + "load_average": [0.5, 0.4, 0.3], + "load_average_percent": [12.5, 10.0, 7.5], +} + + +class TestStatistics: + async def test_statistics_shape(self, app: FastAPI, client: AsyncClient, project: Project) -> None: + + webwireshark = { + "total_containers": 1, + "running_containers": 1, + "active_sessions": 0, + "containers": [ + { + "project_id": project.id, + "project_name": project.name, + "container_id": "abcdef123456", + "status": "running", + "running": True, + "active_sessions": 0, + "memory_limit": "unlimited", + "cpu_limit": "unlimited", + "pids_limit": "unlimited", + "memory": "10MiB / 1GiB", + "cpu": "0.5%", + "pids": 3, + }, + { + "project_id": project.id, + "project_name": project.name, + "container_id": "123456abcdef", + "status": "exited", + "running": False, + }, + ], + } + with patch( + "gns3server.api.routes.controller.controller.collect_webwireshark_stats", + new=AsyncMock(return_value=webwireshark), + ): + response = await client.get(app.url_path_for("statistics")) + assert response.status_code == status.HTTP_200_OK + body = response.json() + assert set(body) == {"uptime_seconds", "computes", "projects", "nodes", "links", "webwireshark"} + assert isinstance(body["uptime_seconds"], int) + assert body["projects"] == {"total": 1, "opened": 1, "closed": 0} + assert body["nodes"] == { + "total": 0, + "open_project_nodes": 0, + "closed_project_nodes": 0, + "by_type": {}, + "by_status": {}, + } + assert body["links"] == {"total": 0, "capturing": 0} + assert body["webwireshark"] == webwireshark + + async def test_statistics_compute_entry( + self, app: FastAPI, client: AsyncClient, controller: Controller, project: Project + ) -> None: + + compute = MagicMock() + compute.id = "local" + compute.name = "local" + reply = MagicMock() + reply.json = dict(COMPUTE_STATISTICS) + compute.get = AsyncMock(return_value=reply) + controller._computes = {"local": compute} + + response = await client.get(app.url_path_for("statistics")) + assert response.status_code == status.HTTP_200_OK + assert response.json()["computes"] == [ + {"compute_id": "local", "compute_name": "local", "statistics": COMPUTE_STATISTICS} + ] + + async def test_statistics_compute_missing_fields_stay_absent( + self, app: FastAPI, client: AsyncClient, controller: Controller, project: Project + ) -> None: + + compute = MagicMock() + compute.id = "old" + compute.name = "old" + reply = MagicMock() + reply.json = {"memory_total": 8000, "cpu_count": 2} + compute.get = AsyncMock(return_value=reply) + controller._computes = {"old": compute} + + response = await client.get(app.url_path_for("statistics")) + assert response.status_code == status.HTTP_200_OK + assert response.json()["computes"][0]["statistics"] == {"memory_total": 8000, "cpu_count": 2} + + async def test_project_stats(self, app: FastAPI, client: AsyncClient, project: Project) -> None: + + link = UDPLink(project) + link._markers["a"] = {"bpf": "icmp"} + link._markers["b"] = {"bpf": "arp"} + project._links = {link.id: link} + + response = await client.get(app.url_path_for("get_project_stats", project_id=project.id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"nodes": 0, "links": 1, "drawings": 0, "snapshots": 0, "markers": 2} + + +class TestSymbols: + async def test_list_shape(self, app: FastAPI, client: AsyncClient) -> None: + + response = await client.get(app.url_path_for("get_symbols")) + assert response.status_code == status.HTTP_200_OK + symbols = response.json() + assert symbols + for symbol in symbols: + assert set(symbol) == {"symbol_id", "filename", "theme", "builtin"} + assert isinstance(symbol["builtin"], bool) + + async def test_default_symbols(self, app: FastAPI, client: AsyncClient) -> None: + + response = await client.get(app.url_path_for("get_default_symbols")) + assert response.status_code == status.HTTP_200_OK + themes = response.json() + assert themes["Classic"]["firewall"] == ":/symbols/classic/firewall.svg" + assert themes == Controller.instance().symbols.default_symbols() + + async def test_dimensions(self, app: FastAPI, client: AsyncClient) -> None: + + response = await client.get( + app.url_path_for("get_symbol_dimensions", symbol_id=":/symbols/classic/firewall.svg") + ) + assert response.status_code == status.HTTP_200_OK + body = response.json() + assert set(body) == {"width", "height"} + assert isinstance(body["width"], int) + assert isinstance(body["height"], int) + + +class TestLinkRoutes: + async def test_wireshark_restart(self, app: FastAPI, client: AsyncClient, project: Project) -> None: + + link = Link(project) + project._links = {link.id: link} + with patch("gns3server.controller.link.Link._restart_web_wireshark", new=AsyncMock()): + response = await client.post(app.url_path_for("restart_wireshark", project_id=project.id, link_id=link.id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"status": "restarted"} + + async def test_available_filters_keep_optional_parameter_keys_absent( + self, app: FastAPI, client: AsyncClient, project: Project + ) -> None: + + link = Link(project) + project._links = {link.id: link} + with patch.object(Link, "_get_filter_node", return_value=object()): + response = await client.get(app.url_path_for("get_filters", project_id=project.id, link_id=link.id)) + assert response.status_code == status.HTTP_200_OK + by_type = {f["type"]: f for f in response.json()} + assert by_type["bpf"]["parameters"] == [{"name": "Filters", "type": "text"}] + assert by_type["delay"]["parameters"][0] == { + "name": "Latency", + "minimum": 1, + "maximum": 32767, + "unit": "ms", + "type": "int", + } + + async def test_link_markers_keep_missing_keys_absent( + self, app: FastAPI, client: AsyncClient, project: Project + ) -> None: + + link = UDPLink(project) + link._markers["legacy"] = {"bpf": "icmp", "tag": 1, "enabled": True} + link._markers["global-arp"] = {"bpf": "arp", "inherited_from": "arp", "capture_node_id": "n1"} + project._links = {link.id: link} + + response = await client.get(app.url_path_for("get_markers", project_id=project.id, link_id=link.id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == link._markers + + +class TestMarkerAggregation: + async def test_project_markers_and_definitions(self, app: FastAPI, client: AsyncClient, project: Project) -> None: + + link = UDPLink(project) + link._markers["web"] = { + "bpf": "tcp port 80", + "tag": 4, + "enabled": False, + "color": "#ff5722", + "highlight_duration": 800, + "capture_node_id": "n1", + "direction": "tx", + "data_link_type": "DLT_EN10MB", + } + link._markers["global-arp"] = { + "bpf": "arp", + "tag": None, + "enabled": True, + "color": None, + "highlight_duration": None, + "capture_node_id": "n1", + "direction": None, + "data_link_type": "DLT_EN10MB", + "inherited_from": "arp", + } + project._links = {link.id: link} + project._marker_definitions = { + "arp": { + "bpf": "arp", + "tag": None, + "direction": None, + "color": None, + "highlight_duration": None, + "data_link_type": "DLT_EN10MB", + "paused": False, + }, + "old": {"bpf": "icmp", "tag": 5}, + } + + response = await client.get(app.url_path_for("get_project_markers", project_id=project.id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == { + f"{link.id}/web": {**link._markers["web"], "link_id": link.id, "node_id": "n1"}, + f"{link.id}/global-arp": {**link._markers["global-arp"], "link_id": link.id, "node_id": "n1"}, + } + + response = await client.get(app.url_path_for("get_marker_definitions", project_id=project.id)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == { + "arp": {**project._marker_definitions["arp"], "link_ids": [link.id]}, + "old": {"bpf": "icmp", "tag": 5, "link_ids": []}, + } + + +class TestReplayFrameDetail: + async def test_frame_detail_passes_unknown_tree_keys_through( + self, app: FastAPI, client: AsyncClient, project: Project, monkeypatch + ) -> None: + + detail = { + "ts": "1693472000.000000", + "source": {"node_id": "n1", "link_id": "l1", "marker": "icmp", "frame_number": 1}, + "field_count": 3, + "hex": "0200", + "tree": [ + { + "element": "proto", + "label": "Internet Protocol", + "name": "ip", + "filter_expr": "ip", + "pos": 14, + "size": 20, + "children": [{"label": "TTL: 64", "name": "ip.ttl", "expert": "Chat", "generated": True}], + "future_key": {"a": 1}, + } + ], + } + monkeypatch.setattr(marker_replay, "decode_frame", AsyncMock(return_value=detail)) + + response = await client.get( + app.url_path_for("replay_tag_frame_detail", project_id=project.id, tag=7), + params={"ts": "1693472000.000000", "node_id": "n1", "link_id": "l1", "marker": "icmp"}, + ) + assert response.status_code == status.HTTP_200_OK + assert response.json() == detail + + async def test_range_empty_timeline(self, app: FastAPI, client: AsyncClient, project: Project) -> None: + + link = UDPLink(project) + link._markers["icmp"] = { + "bpf": "icmp", + "tag": 9, + "enabled": False, + "color": None, + "highlight_duration": None, + "capture_node_id": "n1", + "direction": None, + "data_link_type": "DLT_EN10MB", + } + project._links = {link.id: link} + _write_pcap(f"{project.markers_directory}/n1_{link.id}_icmp.pcap", []) + + response = await client.get(app.url_path_for("replay_tag_range", project_id=project.id, tag=9)) + assert response.status_code == status.HTTP_200_OK + assert response.json() == { + "tag": 9, + "start": None, + "end": None, + "frame_count": 0, + "sources": [ + {"node_id": "n1", "link_id": link.id, "marker": "icmp", "data_link_type": "DLT_EN10MB", "count": 0} + ], + "frames": [], + } diff --git a/tests/api/routes/controller/test_version.py b/tests/api/routes/controller/test_version.py index a56e1ff26..19d1abe17 100644 --- a/tests/api/routes/controller/test_version.py +++ b/tests/api/routes/controller/test_version.py @@ -17,6 +17,7 @@ import pytest from fastapi import FastAPI, status from httpx import AsyncClient +from starlette.routing import Mount from gns3server.version import __version__ @@ -26,6 +27,14 @@ pytestmark = pytest.mark.asyncio class TestVersionRoutes: async def test_version_output(self, app: FastAPI, client: AsyncClient) -> None: + # controller_host is recorded on the mounted compute sub-application + # when the controller queries the compute capabilities. Set it + # directly so this test does not depend on another test having made + # that call first (it failed when run on its own). + for route in app.routes: + if isinstance(route, Mount) and route.name == "compute": + route.app.state.controller_host = "127.0.0.1" + response = await client.get(app.url_path_for("get_version")) assert response.status_code == status.HTTP_200_OK assert response.json() == {"controller_host": "127.0.0.1", "local": False, "version": __version__} diff --git a/tests/utils/test_drawings.py b/tests/utils/test_drawings.py new file mode 100644 index 000000000..63d6ef6f5 --- /dev/null +++ b/tests/utils/test_drawings.py @@ -0,0 +1,194 @@ +""" +Web UI compatible drawing SVG builder tests. +""" + +from xml.etree import ElementTree as ET + +import pytest + +from gns3server.utils.drawings import build_ellipse_svg, build_rectangle_svg, build_text_svg + + +class TestBuildTextSvg: + def test_single_line(self): + svg, width, height, line_height = build_text_svg("R1", font_size=13) + assert svg == ( + '' + 'R1' + ) + # 2 chars x 13 x 0.8 = 20.8 -> 21; one line of 13 x 1.9 = 24.7 -> 25 + assert width == 21 + assert height == 25 + assert line_height == 25 + + def test_bold_and_color(self): + svg, _, _, _ = build_text_svg("R1", font_size=13, color="#CC0000", bold=True) + assert 'fill="#CC0000"' in svg + assert 'font-weight="bold"' in svg + + def test_escapes_xml_characters(self): + svg, _, _, _ = build_text_svg("a mono_w + + def test_empty_line_kept(self): + # a blank line still occupies a line box + _, _, height, line_height = build_text_svg("a\n\nb", font_size=10) + assert height == 3 * line_height + + def test_matches_web_ui_serializer_shape(self): + # byte-for-byte structural parity with the Web UI's own text output + svg, width, height, _ = build_text_svg("demo", font_size=11) + assert svg.startswith(f'", "a&b", ""]) + def test_bad_font_family(self, font_family): + with pytest.raises(ValueError): + build_text_svg("R1", font_family=font_family) + + def test_font_family_with_spaces_accepted(self): + svg, _, _, _ = build_text_svg("R1", font_family="DejaVu Sans Mono") + assert 'font-family="DejaVu Sans Mono"' in svg + + +class TestBuildRectangleSvg: + def test_plain_fill(self): + svg, width, height = build_rectangle_svg(80, 50, fill="#4A90D9") + assert svg == ( + '' + '' + ) + assert (width, height) == (80, 50) + # no stroke requested -> no stroke attributes (Web UI serializer parity) + assert "stroke" not in svg + + def test_stroke_and_dashed(self): + svg, _, _ = build_rectangle_svg( + 80, 50, fill="none", fill_opacity=0.5, stroke="#333333", stroke_width=2, dashed=True + ) + assert 'fill="none"' in svg + assert 'fill-opacity="0.5"' in svg + assert 'stroke="#333333" stroke-width="2" stroke-dasharray="10,6"' in svg + + def test_custom_dasharray(self): + svg, _, _ = build_rectangle_svg(80, 50, stroke="#000000", dasharray="4,2") + assert 'stroke-dasharray="4,2"' in svg + + def test_corner_radius(self): + svg, _, _ = build_rectangle_svg(100, 60, rx=8) + assert 'rx="8" ry="8"' in svg + + def test_fill_none_renders_unfilled(self): + # the Web UI renders fill="none" as an outline-only box + svg, _, _ = build_rectangle_svg(80, 50, fill="none", stroke="#333333") + ET.fromstring(svg) + + @pytest.mark.parametrize("width,height", [(0, 50), (80, 0), (-1, 50), (2001, 50), ("80", 50)]) + def test_bad_dimensions(self, width, height): + with pytest.raises(ValueError): + build_rectangle_svg(width, height) + + def test_radius_too_large(self): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, rx=26) + + @pytest.mark.parametrize("fill", ["red", 123, "transparent"]) + def test_bad_fill(self, fill): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, fill=fill) + + @pytest.mark.parametrize("fill_opacity", [-0.1, 1.1, "0.5", True]) + def test_bad_fill_opacity(self, fill_opacity): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, fill_opacity=fill_opacity) + + def test_bad_stroke(self): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, stroke="blue") + + def test_bad_stroke_width(self): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, stroke="#333333", stroke_width=0) + + @pytest.mark.parametrize("dasharray", ["25, 25", "5, 25", "abc", "5;5", '5,5" onload="x']) + def test_bad_dasharray(self, dasharray): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, stroke="#333333", dasharray=dasharray) + + def test_dasharray_without_stroke(self): + with pytest.raises(ValueError): + build_rectangle_svg(80, 50, dasharray="5,5") + + +class TestBuildEllipseSvg: + def test_bounding_box_derivation(self): + svg, width, height = build_ellipse_svg(100, 60, fill="#5AA9DD", fill_opacity=0.8) + assert svg == ( + '' + '' + ) + assert (width, height) == (100, 60) + + def test_odd_dimensions_round_down(self): + svg, width, height = build_ellipse_svg(101, 61) + assert (width, height) == (100, 60) + assert 'cx="50" cy="30" rx="50" ry="30"' in svg + + def test_circle(self): + svg, width, height = build_ellipse_svg(80, 80, stroke="#333333") + assert width == height == 80 + assert 'rx="40" ry="40"' in svg + + def test_too_small(self): + with pytest.raises(ValueError): + build_ellipse_svg(1, 60) + + def test_stroke_and_dashed(self): + svg, _, _ = build_ellipse_svg(100, 60, fill="none", stroke="#E74C3C", stroke_width=2, dashed=True) + assert 'stroke-dasharray="10,6"' in svg + ET.fromstring(svg)