mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-03 05:51:15 +02:00
186 files changed
+1173
-1206
No files matched your search
@@ -23,124 +23,124 @@ import asyncio
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from gns3server.controller.project import Project
|
||||
from gns3server.controller.hypervisor import Hypervisor, HypervisorError
|
||||
from gns3server.controller.compute import Compute, ComputeError
|
||||
from gns3server.version import __version__
|
||||
from tests.utils import asyncio_patch, AsyncioMagicMock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hypervisor():
|
||||
hypervisor = Hypervisor("my_hypervisor_id", protocol="https", host="example.com", port=84, controller=MagicMock())
|
||||
hypervisor._connected = True
|
||||
return hypervisor
|
||||
def compute():
|
||||
compute = Compute("my_compute_id", protocol="https", host="example.com", port=84, controller=MagicMock())
|
||||
compute._connected = True
|
||||
return compute
|
||||
|
||||
|
||||
def test_init(hypervisor):
|
||||
assert hypervisor.id == "my_hypervisor_id"
|
||||
def test_init(compute):
|
||||
assert compute.id == "my_compute_id"
|
||||
|
||||
|
||||
def test_hypervisor_local(hypervisor):
|
||||
def test_compute_local(compute):
|
||||
"""
|
||||
If the hypervisor is local but the hypervisor id is local
|
||||
If the compute is local but the compute id is local
|
||||
it's a configuration issue
|
||||
"""
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"local": False}):
|
||||
with pytest.raises(HypervisorError):
|
||||
s = Hypervisor("local", controller=MagicMock())
|
||||
with pytest.raises(ComputeError):
|
||||
s = Compute("local", controller=MagicMock())
|
||||
|
||||
with patch("gns3server.config.Config.get_section_config", return_value={"local": True}):
|
||||
s = Hypervisor("test", controller=MagicMock())
|
||||
s = Compute("test", controller=MagicMock())
|
||||
|
||||
|
||||
def test_hypervisor_httpQuery(hypervisor, async_run):
|
||||
def test_compute_httpQuery(compute, async_run):
|
||||
response = MagicMock()
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
response.status = 200
|
||||
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
mock.assert_called_with("POST", "https://example.com:84/v2/hypervisor/projects", data='{"a": "b"}', headers={'content-type': 'application/json'}, auth=None)
|
||||
assert hypervisor._auth is None
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
mock.assert_called_with("POST", "https://example.com:84/v2/compute/projects", data='{"a": "b"}', headers={'content-type': 'application/json'}, auth=None)
|
||||
assert compute._auth is None
|
||||
|
||||
|
||||
def test_hypervisor_httpQueryAuth(hypervisor, async_run):
|
||||
def test_compute_httpQueryAuth(compute, async_run):
|
||||
response = MagicMock()
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
response.status = 200
|
||||
|
||||
hypervisor.user = "root"
|
||||
hypervisor.password = "toor"
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
mock.assert_called_with("POST", "https://example.com:84/v2/hypervisor/projects", data='{"a": "b"}', headers={'content-type': 'application/json'}, auth=hypervisor._auth)
|
||||
assert hypervisor._auth.login == "root"
|
||||
assert hypervisor._auth.password == "toor"
|
||||
compute.user = "root"
|
||||
compute.password = "toor"
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
mock.assert_called_with("POST", "https://example.com:84/v2/compute/projects", data='{"a": "b"}', headers={'content-type': 'application/json'}, auth=compute._auth)
|
||||
assert compute._auth.login == "root"
|
||||
assert compute._auth.password == "toor"
|
||||
|
||||
|
||||
def test_hypervisor_httpQueryNotConnected(hypervisor, async_run):
|
||||
hypervisor._connected = False
|
||||
def test_compute_httpQueryNotConnected(compute, async_run):
|
||||
compute._connected = False
|
||||
response = AsyncioMagicMock()
|
||||
response.read = AsyncioMagicMock(return_value=json.dumps({"version": __version__}).encode())
|
||||
response.status = 200
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/hypervisor/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
mock.assert_any_call("POST", "https://example.com:84/v2/hypervisor/projects", data='{"a": "b"}', headers={'content-type': 'application/json'}, auth=None)
|
||||
assert hypervisor._connected
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/compute/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
mock.assert_any_call("POST", "https://example.com:84/v2/compute/projects", data='{"a": "b"}', headers={'content-type': 'application/json'}, auth=None)
|
||||
assert compute._connected
|
||||
|
||||
|
||||
def test_hypervisor_httpQueryNotConnectedInvalidVersion(hypervisor, async_run):
|
||||
hypervisor._connected = False
|
||||
def test_compute_httpQueryNotConnectedInvalidVersion(compute, async_run):
|
||||
compute._connected = False
|
||||
response = AsyncioMagicMock()
|
||||
response.read = AsyncioMagicMock(return_value=json.dumps({"version": "1.42.4"}).encode())
|
||||
response.status = 200
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
with pytest.raises(aiohttp.web.HTTPConflict):
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/hypervisor/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/compute/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
|
||||
|
||||
def test_hypervisor_httpQueryNotConnectedNonGNS3Server(hypervisor, async_run):
|
||||
hypervisor._connected = False
|
||||
def test_compute_httpQueryNotConnectedNonGNS3Server(compute, async_run):
|
||||
compute._connected = False
|
||||
response = AsyncioMagicMock()
|
||||
response.read = AsyncioMagicMock(return_value=b'Blocked by super antivirus')
|
||||
response.status = 200
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
with pytest.raises(aiohttp.web.HTTPConflict):
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/hypervisor/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/compute/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
|
||||
|
||||
def test_hypervisor_httpQueryNotConnectedNonGNS3Server2(hypervisor, async_run):
|
||||
hypervisor._connected = False
|
||||
def test_compute_httpQueryNotConnectedNonGNS3Server2(compute, async_run):
|
||||
compute._connected = False
|
||||
response = AsyncioMagicMock()
|
||||
response.read = AsyncioMagicMock(return_value=b'{}')
|
||||
response.status = 200
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
with pytest.raises(aiohttp.web.HTTPConflict):
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/hypervisor/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
mock.assert_any_call("GET", "https://example.com:84/v2/compute/version", headers={'content-type': 'application/json'}, data=None, auth=None)
|
||||
|
||||
|
||||
def test_hypervisor_httpQueryError(hypervisor, async_run):
|
||||
def test_compute_httpQueryError(compute, async_run):
|
||||
response = MagicMock()
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
response.status = 409
|
||||
|
||||
with pytest.raises(aiohttp.web.HTTPConflict):
|
||||
async_run(hypervisor.post("/projects", {"a": "b"}))
|
||||
async_run(compute.post("/projects", {"a": "b"}))
|
||||
|
||||
|
||||
def test_hypervisor_httpQuery_project(hypervisor, async_run):
|
||||
def test_compute_httpQuery_project(compute, async_run):
|
||||
response = MagicMock()
|
||||
with asyncio_patch("aiohttp.ClientSession.request", return_value=response) as mock:
|
||||
response.status = 200
|
||||
|
||||
project = Project()
|
||||
async_run(hypervisor.post("/projects", project))
|
||||
mock.assert_called_with("POST", "https://example.com:84/v2/hypervisor/projects", data=json.dumps(project.__json__()), headers={'content-type': 'application/json'}, auth=None)
|
||||
async_run(compute.post("/projects", project))
|
||||
mock.assert_called_with("POST", "https://example.com:84/v2/compute/projects", data=json.dumps(project.__json__()), headers={'content-type': 'application/json'}, auth=None)
|
||||
|
||||
|
||||
def test_connectNotification(hypervisor, async_run):
|
||||
def test_connectNotification(compute, async_run):
|
||||
ws_mock = AsyncioMagicMock()
|
||||
|
||||
call = 0
|
||||
@@ -159,20 +159,20 @@ def test_connectNotification(hypervisor, async_run):
|
||||
response.tp = aiohttp.MsgType.closed
|
||||
return response
|
||||
|
||||
hypervisor._controller = MagicMock()
|
||||
hypervisor._session = AsyncioMagicMock(return_value=ws_mock)
|
||||
hypervisor._session.ws_connect = AsyncioMagicMock(return_value=ws_mock)
|
||||
compute._controller = MagicMock()
|
||||
compute._session = AsyncioMagicMock(return_value=ws_mock)
|
||||
compute._session.ws_connect = AsyncioMagicMock(return_value=ws_mock)
|
||||
ws_mock.receive = receive
|
||||
async_run(hypervisor._connectNotification())
|
||||
async_run(compute._connectNotification())
|
||||
|
||||
hypervisor._controller.emit.assert_called_with('test', {'a': 1}, hypervisor_id=hypervisor.id, project_id='42')
|
||||
assert hypervisor._connected is False
|
||||
compute._controller.emit.assert_called_with('test', {'a': 1}, compute_id=compute.id, project_id='42')
|
||||
assert compute._connected is False
|
||||
|
||||
|
||||
def test_json(hypervisor):
|
||||
hypervisor.user = "test"
|
||||
assert hypervisor.__json__() == {
|
||||
"hypervisor_id": "my_hypervisor_id",
|
||||
def test_json(compute):
|
||||
compute.user = "test"
|
||||
assert compute.__json__() == {
|
||||
"compute_id": "my_compute_id",
|
||||
"protocol": "https",
|
||||
"host": "example.com",
|
||||
"port": 84,
|
||||
@@ -22,7 +22,7 @@ from unittest.mock import MagicMock
|
||||
|
||||
|
||||
from gns3server.controller import Controller
|
||||
from gns3server.controller.hypervisor import Hypervisor
|
||||
from gns3server.controller.compute import Compute
|
||||
from gns3server.controller.project import Project
|
||||
from gns3server.config import Config
|
||||
|
||||
@@ -34,22 +34,22 @@ def test_isEnabled(controller):
|
||||
assert controller.isEnabled()
|
||||
|
||||
|
||||
def test_addHypervisor(controller, async_run):
|
||||
async_run(controller.addHypervisor("test1"))
|
||||
assert len(controller.hypervisors) == 1
|
||||
async_run(controller.addHypervisor("test1"))
|
||||
assert len(controller.hypervisors) == 1
|
||||
async_run(controller.addHypervisor("test2"))
|
||||
assert len(controller.hypervisors) == 2
|
||||
def test_addCompute(controller, async_run):
|
||||
async_run(controller.addCompute("test1"))
|
||||
assert len(controller.computes) == 1
|
||||
async_run(controller.addCompute("test1"))
|
||||
assert len(controller.computes) == 1
|
||||
async_run(controller.addCompute("test2"))
|
||||
assert len(controller.computes) == 2
|
||||
|
||||
|
||||
def test_getHypervisor(controller, async_run):
|
||||
def test_getCompute(controller, async_run):
|
||||
|
||||
hypervisor = async_run(controller.addHypervisor("test1"))
|
||||
compute = async_run(controller.addCompute("test1"))
|
||||
|
||||
assert controller.getHypervisor("test1") == hypervisor
|
||||
assert controller.getCompute("test1") == compute
|
||||
with pytest.raises(aiohttp.web.HTTPNotFound):
|
||||
assert controller.getHypervisor("dsdssd")
|
||||
assert controller.getCompute("dsdssd")
|
||||
|
||||
|
||||
def test_addProject(controller, async_run):
|
||||
@@ -74,15 +74,15 @@ def test_removeProject(controller, async_run):
|
||||
assert len(controller.projects) == 0
|
||||
|
||||
|
||||
def test_addProject_with_hypervisor(controller, async_run):
|
||||
def test_addProject_with_compute(controller, async_run):
|
||||
uuid1 = str(uuid.uuid4())
|
||||
|
||||
hypervisor = Hypervisor("test1", controller=MagicMock())
|
||||
hypervisor.post = MagicMock()
|
||||
controller._hypervisors = {"test1": hypervisor}
|
||||
compute = Compute("test1", controller=MagicMock())
|
||||
compute.post = MagicMock()
|
||||
controller._computes = {"test1": compute}
|
||||
|
||||
project1 = async_run(controller.addProject(project_id=uuid1))
|
||||
hypervisor.post.assert_called_with("/projects", project1)
|
||||
compute.post.assert_called_with("/projects", project1)
|
||||
|
||||
|
||||
def test_getProject(controller, async_run):
|
||||
|
||||
@@ -20,7 +20,7 @@ from unittest.mock import MagicMock
|
||||
|
||||
from gns3server.controller.link import Link
|
||||
from gns3server.controller.vm import VM
|
||||
from gns3server.controller.hypervisor import Hypervisor
|
||||
from gns3server.controller.compute import Compute
|
||||
from gns3server.controller.project import Project
|
||||
|
||||
|
||||
@@ -30,12 +30,12 @@ def project():
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hypervisor():
|
||||
return Hypervisor("example.com", controller=MagicMock())
|
||||
def compute():
|
||||
return Compute("example.com", controller=MagicMock())
|
||||
|
||||
|
||||
def test_addVM(async_run, project, hypervisor):
|
||||
vm1 = VM(project, hypervisor)
|
||||
def test_addVM(async_run, project, compute):
|
||||
vm1 = VM(project, compute)
|
||||
|
||||
link = Link(project)
|
||||
async_run(link.addVM(vm1, 0, 4))
|
||||
@@ -48,9 +48,9 @@ def test_addVM(async_run, project, hypervisor):
|
||||
]
|
||||
|
||||
|
||||
def test_json(async_run, project, hypervisor):
|
||||
vm1 = VM(project, hypervisor)
|
||||
vm2 = VM(project, hypervisor)
|
||||
def test_json(async_run, project, compute):
|
||||
vm1 = VM(project, compute)
|
||||
vm2 = VM(project, compute)
|
||||
|
||||
link = Link(project)
|
||||
async_run(link.addVM(vm1, 0, 4))
|
||||
|
||||
@@ -39,31 +39,31 @@ def test_json(tmpdir):
|
||||
|
||||
|
||||
def test_addVM(async_run):
|
||||
hypervisor = MagicMock()
|
||||
compute = MagicMock()
|
||||
project = Project()
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
hypervisor.post = AsyncioMagicMock(return_value=response)
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
vm = async_run(project.addVM(hypervisor, None, name="test", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
vm = async_run(project.addVM(compute, None, name="test", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
|
||||
hypervisor.post.assert_called_with('/projects/{}/vpcs/vms'.format(project.id),
|
||||
data={'vm_id': vm.id,
|
||||
'console_type': 'telnet',
|
||||
'startup_config': 'test.cfg',
|
||||
'name': 'test'})
|
||||
compute.post.assert_called_with('/projects/{}/vpcs/vms'.format(project.id),
|
||||
data={'vm_id': vm.id,
|
||||
'console_type': 'telnet',
|
||||
'startup_config': 'test.cfg',
|
||||
'name': 'test'})
|
||||
|
||||
|
||||
def test_getVM(async_run):
|
||||
hypervisor = MagicMock()
|
||||
compute = MagicMock()
|
||||
project = Project()
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
hypervisor.post = AsyncioMagicMock(return_value=response)
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
vm = async_run(project.addVM(hypervisor, None, name="test", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
vm = async_run(project.addVM(compute, None, name="test", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
assert project.getVM(vm.id) == vm
|
||||
|
||||
with pytest.raises(aiohttp.web_exceptions.HTTPNotFound):
|
||||
@@ -71,15 +71,15 @@ def test_getVM(async_run):
|
||||
|
||||
|
||||
def test_addLink(async_run):
|
||||
hypervisor = MagicMock()
|
||||
compute = MagicMock()
|
||||
project = Project()
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
hypervisor.post = AsyncioMagicMock(return_value=response)
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
vm1 = async_run(project.addVM(hypervisor, None, name="test1", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
vm2 = async_run(project.addVM(hypervisor, None, name="test2", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
vm1 = async_run(project.addVM(compute, None, name="test1", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
vm2 = async_run(project.addVM(compute, None, name="test2", vm_type="vpcs", properties={"startup_config": "test.cfg"}))
|
||||
link = async_run(project.addLink())
|
||||
async_run(link.addVM(vm1, 3, 1))
|
||||
async_run(link.addVM(vm2, 4, 2))
|
||||
@@ -87,12 +87,12 @@ def test_addLink(async_run):
|
||||
|
||||
|
||||
def test_getLink(async_run):
|
||||
hypervisor = MagicMock()
|
||||
compute = MagicMock()
|
||||
project = Project()
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
hypervisor.post = AsyncioMagicMock(return_value=response)
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
link = async_run(project.addLink())
|
||||
assert project.getLink(link.id) == link
|
||||
|
||||
@@ -21,7 +21,7 @@ import aiohttp
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from gns3server.controller.project import Project
|
||||
from gns3server.controller.hypervisor import Hypervisor
|
||||
from gns3server.controller.compute import Compute
|
||||
from gns3server.controller.udp_link import UDPLink
|
||||
from gns3server.controller.vm import VM
|
||||
|
||||
@@ -32,18 +32,18 @@ def project():
|
||||
|
||||
|
||||
def test_create(async_run, project):
|
||||
hypervisor1 = MagicMock()
|
||||
hypervisor2 = MagicMock()
|
||||
compute1 = MagicMock()
|
||||
compute2 = MagicMock()
|
||||
|
||||
vm1 = VM(project, hypervisor1, vm_type="vpcs")
|
||||
vm2 = VM(project, hypervisor2, vm_type="vpcs")
|
||||
vm1 = VM(project, compute1, vm_type="vpcs")
|
||||
vm2 = VM(project, compute2, vm_type="vpcs")
|
||||
|
||||
link = UDPLink(project)
|
||||
async_run(link.addVM(vm1, 0, 4))
|
||||
async_run(link.addVM(vm2, 3, 1))
|
||||
|
||||
@asyncio.coroutine
|
||||
def hypervisor1_callback(path, data={}):
|
||||
def compute1_callback(path, data={}):
|
||||
"""
|
||||
Fake server
|
||||
"""
|
||||
@@ -53,7 +53,7 @@ def test_create(async_run, project):
|
||||
return response
|
||||
|
||||
@asyncio.coroutine
|
||||
def hypervisor2_callback(path, data={}):
|
||||
def compute2_callback(path, data={}):
|
||||
"""
|
||||
Fake server
|
||||
"""
|
||||
@@ -62,32 +62,32 @@ def test_create(async_run, project):
|
||||
response.json = {"udp_port": 2048}
|
||||
return response
|
||||
|
||||
hypervisor1.post.side_effect = hypervisor1_callback
|
||||
hypervisor1.host = "example.com"
|
||||
hypervisor2.post.side_effect = hypervisor2_callback
|
||||
hypervisor2.host = "example.org"
|
||||
compute1.post.side_effect = compute1_callback
|
||||
compute1.host = "example.com"
|
||||
compute2.post.side_effect = compute2_callback
|
||||
compute2.host = "example.org"
|
||||
async_run(link.create())
|
||||
|
||||
hypervisor1.post.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/0/ports/4/nio".format(project.id, vm1.id), data={
|
||||
compute1.post.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/0/ports/4/nio".format(project.id, vm1.id), data={
|
||||
"lport": 1024,
|
||||
"rhost": hypervisor2.host,
|
||||
"rhost": compute2.host,
|
||||
"rport": 2048,
|
||||
"type": "nio_udp"
|
||||
})
|
||||
hypervisor2.post.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/3/ports/1/nio".format(project.id, vm2.id), data={
|
||||
compute2.post.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/3/ports/1/nio".format(project.id, vm2.id), data={
|
||||
"lport": 2048,
|
||||
"rhost": hypervisor1.host,
|
||||
"rhost": compute1.host,
|
||||
"rport": 1024,
|
||||
"type": "nio_udp"
|
||||
})
|
||||
|
||||
|
||||
def test_delete(async_run, project):
|
||||
hypervisor1 = MagicMock()
|
||||
hypervisor2 = MagicMock()
|
||||
compute1 = MagicMock()
|
||||
compute2 = MagicMock()
|
||||
|
||||
vm1 = VM(project, hypervisor1, vm_type="vpcs")
|
||||
vm2 = VM(project, hypervisor2, vm_type="vpcs")
|
||||
vm1 = VM(project, compute1, vm_type="vpcs")
|
||||
vm2 = VM(project, compute2, vm_type="vpcs")
|
||||
|
||||
link = UDPLink(project)
|
||||
async_run(link.addVM(vm1, 0, 4))
|
||||
@@ -95,5 +95,5 @@ def test_delete(async_run, project):
|
||||
|
||||
async_run(link.delete())
|
||||
|
||||
hypervisor1.delete.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/0/ports/4/nio".format(project.id, vm1.id))
|
||||
hypervisor2.delete.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/3/ports/1/nio".format(project.id, vm2.id))
|
||||
compute1.delete.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/0/ports/4/nio".format(project.id, vm1.id))
|
||||
compute2.delete.assert_any_call("/projects/{}/vpcs/vms/{}/adapters/3/ports/1/nio".format(project.id, vm2.id))
|
||||
+26
-26
@@ -28,16 +28,16 @@ from gns3server.controller.project import Project
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def hypervisor():
|
||||
def compute():
|
||||
s = AsyncioMagicMock()
|
||||
s.id = "http://test.com:42"
|
||||
return s
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def vm(hypervisor):
|
||||
def vm(compute):
|
||||
project = Project(str(uuid.uuid4()))
|
||||
vm = VM(project, hypervisor,
|
||||
vm = VM(project, compute,
|
||||
name="demo",
|
||||
vm_id=str(uuid.uuid4()),
|
||||
vm_type="vpcs",
|
||||
@@ -46,9 +46,9 @@ def vm(hypervisor):
|
||||
return vm
|
||||
|
||||
|
||||
def test_json(vm, hypervisor):
|
||||
def test_json(vm, compute):
|
||||
assert vm.__json__() == {
|
||||
"hypervisor_id": hypervisor.id,
|
||||
"compute_id": compute.id,
|
||||
"project_id": vm.project.id,
|
||||
"vm_id": vm.id,
|
||||
"vm_type": vm.vm_type,
|
||||
@@ -59,19 +59,19 @@ def test_json(vm, hypervisor):
|
||||
}
|
||||
|
||||
|
||||
def test_init_without_uuid(project, hypervisor):
|
||||
vm = VM(project, hypervisor,
|
||||
def test_init_without_uuid(project, compute):
|
||||
vm = VM(project, compute,
|
||||
vm_type="vpcs",
|
||||
console_type="vnc")
|
||||
assert vm.id is not None
|
||||
|
||||
|
||||
def test_create(vm, hypervisor, project, async_run):
|
||||
def test_create(vm, compute, project, async_run):
|
||||
vm._console = 2048
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048}
|
||||
hypervisor.post = AsyncioMagicMock(return_value=response)
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
async_run(vm.create())
|
||||
data = {
|
||||
@@ -81,43 +81,43 @@ def test_create(vm, hypervisor, project, async_run):
|
||||
"startup_script": "echo test",
|
||||
"name": "demo"
|
||||
}
|
||||
hypervisor.post.assert_called_with("/projects/{}/vpcs/vms".format(vm.project.id), data=data)
|
||||
compute.post.assert_called_with("/projects/{}/vpcs/vms".format(vm.project.id), data=data)
|
||||
assert vm._console == 2048
|
||||
assert vm._properties == {"startup_script": "echo test"}
|
||||
|
||||
|
||||
def test_start(vm, hypervisor, project, async_run):
|
||||
def test_start(vm, compute, project, async_run):
|
||||
|
||||
hypervisor.post = AsyncioMagicMock()
|
||||
compute.post = AsyncioMagicMock()
|
||||
|
||||
async_run(vm.start())
|
||||
hypervisor.post.assert_called_with("/projects/{}/vpcs/vms/{}/start".format(vm.project.id, vm.id))
|
||||
compute.post.assert_called_with("/projects/{}/vpcs/vms/{}/start".format(vm.project.id, vm.id))
|
||||
|
||||
|
||||
def test_stop(vm, hypervisor, project, async_run):
|
||||
def test_stop(vm, compute, project, async_run):
|
||||
|
||||
hypervisor.post = AsyncioMagicMock()
|
||||
compute.post = AsyncioMagicMock()
|
||||
|
||||
async_run(vm.stop())
|
||||
hypervisor.post.assert_called_with("/projects/{}/vpcs/vms/{}/stop".format(vm.project.id, vm.id))
|
||||
compute.post.assert_called_with("/projects/{}/vpcs/vms/{}/stop".format(vm.project.id, vm.id))
|
||||
|
||||
|
||||
def test_suspend(vm, hypervisor, project, async_run):
|
||||
def test_suspend(vm, compute, project, async_run):
|
||||
|
||||
hypervisor.post = AsyncioMagicMock()
|
||||
compute.post = AsyncioMagicMock()
|
||||
|
||||
async_run(vm.suspend())
|
||||
hypervisor.post.assert_called_with("/projects/{}/vpcs/vms/{}/suspend".format(vm.project.id, vm.id))
|
||||
compute.post.assert_called_with("/projects/{}/vpcs/vms/{}/suspend".format(vm.project.id, vm.id))
|
||||
|
||||
|
||||
def test_create_without_console(vm, hypervisor, project, async_run):
|
||||
def test_create_without_console(vm, compute, project, async_run):
|
||||
"""
|
||||
None properties should be send. Because it can mean the emulator doesn"t support it
|
||||
"""
|
||||
|
||||
response = MagicMock()
|
||||
response.json = {"console": 2048, "test_value": "success"}
|
||||
hypervisor.post = AsyncioMagicMock(return_value=response)
|
||||
compute.post = AsyncioMagicMock(return_value=response)
|
||||
|
||||
async_run(vm.create())
|
||||
data = {
|
||||
@@ -126,16 +126,16 @@ def test_create_without_console(vm, hypervisor, project, async_run):
|
||||
"startup_script": "echo test",
|
||||
"name": "demo"
|
||||
}
|
||||
hypervisor.post.assert_called_with("/projects/{}/vpcs/vms".format(vm.project.id), data=data)
|
||||
compute.post.assert_called_with("/projects/{}/vpcs/vms".format(vm.project.id), data=data)
|
||||
assert vm._console == 2048
|
||||
assert vm._properties == {"test_value": "success", "startup_script": "echo test"}
|
||||
|
||||
|
||||
def test_post(vm, hypervisor, async_run):
|
||||
def test_post(vm, compute, async_run):
|
||||
async_run(vm.post("/test", {"a": "b"}))
|
||||
hypervisor.post.assert_called_with("/projects/{}/vpcs/vms/{}/test".format(vm.project.id, vm.id), data={"a": "b"})
|
||||
compute.post.assert_called_with("/projects/{}/vpcs/vms/{}/test".format(vm.project.id, vm.id), data={"a": "b"})
|
||||
|
||||
|
||||
def test_delete(vm, hypervisor, async_run):
|
||||
def test_delete(vm, compute, async_run):
|
||||
async_run(vm.delete("/test"))
|
||||
hypervisor.delete.assert_called_with("/projects/{}/vpcs/vms/{}/test".format(vm.project.id, vm.id))
|
||||
compute.delete.assert_called_with("/projects/{}/vpcs/vms/{}/test".format(vm.project.id, vm.id))
|
||||
Reference in new issue
Block a user