arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-28 11:18:11 +00:00
Code Issues Releases Wiki Activity

Prevent connection from Cloud to Cloud connection

Browse Source 
Ref #1429
This commit is contained in:
Julien Duponchelle 2016-08-23 23:33:19 +02:00
parent bfabf3ddc8
commit 6cad685a08
No known key found for this signature in database
GPG Key ID: CE8B29639E07F5E8
2 changed files with 57 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
gns3server/controller/link.py
View File

@ -19,7 +19,7 @@ import os
import re
import uuid
import asyncio
import aiohttp
import logging
log = logging.getLogger(__name__)
@ -48,6 +48,11 @@ class Link:
Add a node to the link
"""
for other_node in self._nodes:
if node.node_type in ["nat", "cloud"]:
if other_node["node"].node_type in ["nat", "cloud"]:
raise aiohttp.web.HTTPConflict(text="It's not allowed to connect a {} to a {}".format(other_node["node"].node_type, node.node_type))
if label is None:
label = {
"x": -10,

55
tests/controller/test_link.py
View File

@ -17,6 +17,7 @@
import os
import pytest
import aiohttp
import asyncio
from unittest.mock import MagicMock
@ -87,25 +88,61 @@ def test_add_node(async_run, project, compute):
link._project.controller.notification.emit.assert_called_with("link.created", link.__json__())
def test_update_nodes(async_run, project, compute):
def test_add_node(async_run, project, compute):
node1 = Node(project, compute, "node1", node_type="qemu")
project._nodes[node1.id] = node1
link = Link(project)
link._project.controller.notification.emit = MagicMock()
project.dump = AsyncioMagicMock()
async_run(link.add_node(node1, 0, 4))
label = {
'y': -42,
assert link._nodes == [
{
"node": node1,
"adapter_number": 0,
"port_number": 4,
'label': {
'y': -10,
'text': '0/4',
'x': -10,
'rotation': 0,
'style': 'font-size: 10; font-style: Verdana'
}
project.dump = AsyncioMagicMock()
link._project.controller.notification.emit = MagicMock()
async_run(link.update_nodes([{"node_id": node1.id, "label": label}]))
assert link._nodes[0]["label"]["y"] == -42
}
]
assert project.dump.called
link._project.controller.notification.emit.assert_called_with("link.updated", link.__json__())
assert not link._project.controller.notification.emit.called
# We call link.created only when both side are created
node2 = Node(project, compute, "node2", node_type="qemu")
async_run(link.add_node(node2, 0, 4))
link._project.controller.notification.emit.assert_called_with("link.created", link.__json__())
def test_add_node_cloud(async_run, project, compute):
node1 = Node(project, compute, "node1", node_type="qemu")
node2 = Node(project, compute, "node2", node_type="cloud")
link = Link(project)
link._project.controller.notification.emit = MagicMock()
async_run(link.add_node(node1, 0, 4))
async_run(link.add_node(node2, 0, 4))
def test_add_node_cloud_to_cloud(async_run, project, compute):
"""
Cloud to cloud connection is not allowed
"""
node1 = Node(project, compute, "node1", node_type="cloud")
node2 = Node(project, compute, "node2", node_type="cloud")
link = Link(project)
link._project.controller.notification.emit = MagicMock()
async_run(link.add_node(node1, 0, 4))
with pytest.raises(aiohttp.web.HTTPConflict):
async_run(link.add_node(node2, 0, 4))
def test_json(async_run, project, compute):