NgTech-LTD/gns3-server
2
0 Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-11-03 22:11:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch '3.1' into feat/mcp-text-drawing-tool

Browse Source
This commit is contained in:
Jeremy Grossmann authored and GitHub committed 2026-10-10 06:49:24 +02:00
2 parents 4de6758e2f 15dd9f7d18
commit 2bf35c31e4
32 files changed
+1548 -119
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/api/routes/controller/test_appliances.py
+1 -1
View File
@@ -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
)
tests/api/routes/controller/test_links.py
+41 -2
View File
@@ -81,6 +81,45 @@ class TestLinkRoutes:
assert len(project.links) == 1
assert list(project.links.values())[0].filters == filters
async def test_create_links_batch(
self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
) -> None:
node1, node2 = nodes
valid = {
"nodes": [
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
}
same_node = {
"nodes": [
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
{"node_id": node1.id, "adapter_number": 0, "port_number": 3},
]
}
unknown_node = {
"nodes": [
{"node_id": str(uuid.uuid4()), "adapter_number": 0, "port_number": 3},
{"node_id": node2.id, "adapter_number": 2, "port_number": 4},
]
}
with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
response = await client.post(
app.url_path_for("create_links", project_id=project.id), json=[same_node, unknown_node, valid]
)
assert response.status_code == status.HTTP_200_OK
results = response.json()
assert [r["status_code"] for r in results] == [409, 404, 201]
assert results[0]["link"] is None
assert results[0]["error"]["message"]
assert results[1]["error"]["message"]
assert results[2]["error"] is None
assert results[2]["link"]["link_id"] in project.links
assert len(project.links) == 1
async def test_create_link_failure(
self, app: FastAPI, client: AsyncClient, compute: Compute, project: Project
) -> None:
@@ -609,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:
"""
tests/api/routes/controller/test_markers.py
+68 -3
View File
@@ -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
tests/api/routes/controller/test_nodes.py
+98 -1
View File
@@ -34,7 +34,7 @@ from pydantic import SecretStr
from gns3server.api.routes.controller.nodes import vnc_console, ws_console
from gns3server.config import Config
from gns3server.controller.compute import Compute
from gns3server.controller.controller_error import ComputeError
from gns3server.controller.controller_error import ComputeError, ControllerError
from gns3server.controller.node import Node
from gns3server.controller.project import Project
from gns3server.services import auth_service
@@ -73,6 +73,103 @@ class TestNodeRoutes:
assert response.json()["name"] == "test"
assert "name" not in response.json()["properties"]
async def test_create_node_taken_name_is_renamed(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
payload = {"name": "R1", "node_type": "vpcs", "compute_id": "example.com"}
first = await client.post(app.url_path_for("create_node", project_id=project.id), json=payload)
second = await client.post(app.url_path_for("create_node", project_id=project.id), json=payload)
assert first.json()["name"] == "R1"
assert second.status_code == status.HTTP_201_CREATED
assert second.json()["name"] == "R2"
async def test_create_node_strict_names(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
payload = {"name": "R1", "node_type": "vpcs", "compute_id": "example.com", "strict_names": True}
first = await client.post(app.url_path_for("create_node", project_id=project.id), json=payload)
second = await client.post(app.url_path_for("create_node", project_id=project.id), json=payload)
assert first.status_code == status.HTTP_201_CREATED
assert second.status_code == status.HTTP_409_CONFLICT
assert "R1" in second.json()["message"]
assert len(project.nodes) == 1
async def test_create_nodes_batch(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
nodes = [{"name": "R{0}", "node_type": "vpcs", "compute_id": "example.com"} for _ in range(3)]
response = await client.post(app.url_path_for("create_nodes", project_id=project.id), json=nodes)
assert response.status_code == status.HTTP_200_OK
results = response.json()
assert [r["status_code"] for r in results] == [201, 201, 201]
assert [r["node"]["name"] for r in results] == ["R1", "R2", "R3"]
assert all(r["error"] is None for r in results)
assert len(project.nodes) == 3
async def test_create_nodes_batch_item_failures_keep_order(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute
) -> None:
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
nodes = [
{"name": "R1", "node_type": "vpcs", "compute_id": "example.com", "strict_names": True},
{"name": "R1", "node_type": "vpcs", "compute_id": "example.com", "strict_names": True},
{"name": "R1", "node_type": "vpcs", "compute_id": "unknown"},
{"name": "R1", "node_type": "vpcs", "compute_id": "example.com"},
]
response = await client.post(app.url_path_for("create_nodes", project_id=project.id), json=nodes)
assert response.status_code == status.HTTP_200_OK
results = response.json()
assert [r["status_code"] for r in results] == [201, 409, 404, 201]
assert results[0]["node"]["name"] == "R1"
assert results[1]["node"] is None
assert "R1" in results[1]["error"]["message"]
assert results[2]["error"]["message"]
assert results[3]["node"]["name"] == "R2"
assert len(project.nodes) == 2
async def test_create_nodes_batch_failed_node_releases_name(
self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute
) -> None:
compute.post = AsyncioMagicMock(side_effect=ControllerError("compute failure"))
nodes = [{"name": "R1", "node_type": "vpcs", "compute_id": "example.com", "strict_names": True}]
response = await client.post(app.url_path_for("create_nodes", project_id=project.id), json=nodes)
assert response.json()[0]["status_code"] == status.HTTP_409_CONFLICT
assert len(project.nodes) == 0
response_ok = MagicMock()
response_ok.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response_ok)
response = await client.post(app.url_path_for("create_nodes", project_id=project.id), json=nodes)
assert response.json()[0]["status_code"] == status.HTTP_201_CREATED
assert response.json()[0]["node"]["name"] == "R1"
async def test_list_node(self, app: FastAPI, client: AsyncClient, project: Project, compute: Compute) -> None:
response = MagicMock()
tests/api/routes/controller/test_response_models.py
+464
View File
@@ -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 <http://www.gnu.org/licenses/>.
"""
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": [],
}
tests/controller/test_project.py
+40
View File
@@ -1118,6 +1118,46 @@ async def test_node_name(project):
assert node.name == "R3"
@pytest.mark.asyncio
async def test_node_name_strict(project):
compute = MagicMock()
compute.id = "local"
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node = await project.add_node(compute, "R1", None, node_type="vpcs", strict_names=True)
assert node.name == "R1"
with pytest.raises(ControllerError):
await project.add_node(compute, "R1", None, node_type="vpcs", strict_names=True)
assert len(project.nodes) == 1
node = await project.add_node(compute, "R1", None, node_type="vpcs", strict_names=False)
assert node.name == "R2"
node = await project.add_node(compute, "T{0}", None, node_type="vpcs", strict_names=True)
assert node.name == "T1"
node = await project.add_node(compute, "T{0}", None, node_type="vpcs", strict_names=True)
assert node.name == "T2"
@pytest.mark.asyncio
async def test_node_name_released_when_creation_fails(project):
compute = MagicMock()
compute.id = "local"
compute.post = AsyncioMagicMock(side_effect=ControllerError("failure"))
with pytest.raises(ControllerError):
await project.add_node(compute, "R1", None, node_type="vpcs")
response = MagicMock()
response.json = {"console": 2048}
compute.post = AsyncioMagicMock(return_value=response)
node = await project.add_node(compute, "R1", None, node_type="vpcs", strict_names=True)
assert node.name == "R1"
@pytest.mark.asyncio
async def test_duplicate_node(project):