1
0
mirror of https://github.com/GNS3/gns3-server synced 2024-12-25 16:28:11 +00:00

Add back loop

This commit is contained in:
grossmj 2022-08-27 23:41:52 +02:00
parent fc7d5be736
commit 8f296e54f4
26 changed files with 46 additions and 46 deletions

View File

@ -31,7 +31,7 @@ def nio():
@pytest.fixture @pytest.fixture
async def manager(): async def manager(loop):
m = MagicMock() m = MagicMock()
m.module_name = "builtins" m.module_name = "builtins"
@ -113,7 +113,7 @@ def test_json_without_ports(on_gns3vm, compute_project, manager):
} }
async def test_update_port_mappings(on_gns3vm, compute_project): async def test_update_port_mappings(loop, on_gns3vm, compute_project):
""" """
We don't allow an empty interface in the middle of port list We don't allow an empty interface in the middle of port list
""" """
@ -153,7 +153,7 @@ async def test_update_port_mappings(on_gns3vm, compute_project):
assert cloud.ports_mapping == ports1 assert cloud.ports_mapping == ports1
async def test_linux_ethernet_raw_add_nio(linux_platform, compute_project, nio): async def test_linux_ethernet_raw_add_nio(loop, linux_platform, compute_project, nio):
ports = [ ports = [
{ {
"interface": "eth0", "interface": "eth0",
@ -180,7 +180,7 @@ async def test_linux_ethernet_raw_add_nio(linux_platform, compute_project, nio):
]) ])
async def test_linux_ethernet_raw_add_nio_bridge(linux_platform, compute_project, nio): async def test_linux_ethernet_raw_add_nio_bridge(loop, linux_platform, compute_project, nio):
""" """
Bridge can't be connected directly to a cloud we use a tap in the middle Bridge can't be connected directly to a cloud we use a tap in the middle
""" """

View File

@ -24,7 +24,7 @@ from gns3server.compute.docker.docker_error import DockerError, DockerHttp404Err
@pytest.fixture @pytest.fixture
async def vm(): async def vm(loop):
vm = Docker() vm = Docker()
vm._connected = True vm._connected = True

View File

@ -34,7 +34,7 @@ from unittest.mock import patch, MagicMock, call
@pytest.fixture() @pytest.fixture()
async def manager(port_manager): async def manager(loop, port_manager):
m = Docker.instance() m = Docker.instance()
m.port_manager = port_manager m.port_manager = port_manager
@ -42,7 +42,7 @@ async def manager(port_manager):
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def vm(compute_project, manager): async def vm(loop, compute_project, manager):
vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest") vm = DockerVM("test", str(uuid.uuid4()), compute_project, manager, "ubuntu:latest")
vm._cid = "e90e34656842" vm._cid = "e90e34656842"

View File

@ -29,7 +29,7 @@ from tests.utils import asyncio_patch, AsyncioMagicMock
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = Dynamips.instance() m = Dynamips.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -26,7 +26,7 @@ from gns3server.config import Config
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = Dynamips.instance() m = Dynamips.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -37,7 +37,7 @@ if not sys.platform.startswith("win"):
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = IOU.instance() m = IOU.instance()
m.port_manager = port_manager m.port_manager = port_manager
@ -45,7 +45,7 @@ async def manager(port_manager):
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def vm(compute_project, manager, tmpdir, fake_iou_bin, iourc_file): async def vm(loop, compute_project, manager, tmpdir, fake_iou_bin, iourc_file):
vm = IOUVM("test", str(uuid.uuid4()), compute_project, manager, application_id=1) vm = IOUVM("test", str(uuid.uuid4()), compute_project, manager, application_id=1)
config = manager.config.get_section_config("IOU") config = manager.config.get_section_config("IOU")

View File

@ -66,7 +66,7 @@ def test_invalid_empty_file(tmpdir):
@pytest.mark.skipif(qemu_img() is None, reason="qemu-img is not available") @pytest.mark.skipif(qemu_img() is None, reason="qemu-img is not available")
async def test_rebase(tmpdir): async def test_rebase(loop, tmpdir):
shutil.copy("tests/resources/empty8G.qcow2", str(tmpdir / "empty16G.qcow2")) shutil.copy("tests/resources/empty8G.qcow2", str(tmpdir / "empty16G.qcow2"))
shutil.copy("tests/resources/linked.qcow2", str(tmpdir / "linked.qcow2")) shutil.copy("tests/resources/linked.qcow2", str(tmpdir / "linked.qcow2"))

View File

@ -34,7 +34,7 @@ from gns3server.compute.notification_manager import NotificationManager
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = Qemu.instance() m = Qemu.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -29,7 +29,7 @@ from gns3server.compute.nios.nio_udp import NIOUDP
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def manager(port_manager): async def manager(loop, port_manager):
m = VPCS.instance() m = VPCS.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -29,7 +29,7 @@ from gns3server.utils import force_unix_path
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def vpcs(port_manager): async def vpcs(loop, port_manager):
VPCS._instance = None VPCS._instance = None
vpcs = VPCS.instance() vpcs = VPCS.instance()
@ -38,7 +38,7 @@ async def vpcs(port_manager):
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def qemu(port_manager): async def qemu(loop, port_manager):
Qemu._instance = None Qemu._instance = None
Qemu._init_config_disk = MagicMock() # do not create the config.img image Qemu._init_config_disk = MagicMock() # do not create the config.img image

View File

@ -33,7 +33,7 @@ from gns3server.config import Config
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def manager(port_manager): async def manager(loop, port_manager):
m = VPCS.instance() m = VPCS.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -28,7 +28,7 @@ from gns3server.compute.notification_manager import NotificationManager
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = TraceNG.instance() m = TraceNG.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -29,7 +29,7 @@ from tests.utils import asyncio_patch
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = VirtualBox.instance() m = VirtualBox.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -25,7 +25,7 @@ from gns3server.compute.virtualbox import VirtualBox
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = VirtualBox.instance() m = VirtualBox.instance()
m.port_manager = port_manager m.port_manager = port_manager

View File

@ -22,7 +22,7 @@ from gns3server.compute.vmware import VMware
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = VMware.instance() m = VMware.instance()
m.port_manager = port_manager m.port_manager = port_manager
@ -30,7 +30,7 @@ async def manager(port_manager):
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def vm(compute_project, manager, tmpdir): async def vm(loop, compute_project, manager, tmpdir):
fake_vmx = str(tmpdir / "test.vmx") fake_vmx = str(tmpdir / "test.vmx")
open(fake_vmx, "w+").close() open(fake_vmx, "w+").close()

View File

@ -31,7 +31,7 @@ from gns3server.compute.notification_manager import NotificationManager
@pytest.fixture @pytest.fixture
async def manager(port_manager): async def manager(loop, port_manager):
m = VPCS.instance() m = VPCS.instance()
m.port_manager = port_manager m.port_manager = port_manager
@ -39,7 +39,7 @@ async def manager(port_manager):
@pytest.fixture(scope="function") @pytest.fixture(scope="function")
async def vm(compute_project, manager, tmpdir, ubridge_path): async def vm(loop, compute_project, manager, tmpdir, ubridge_path):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager) vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager)
vm._vpcs_version = parse_version("0.9") vm._vpcs_version = parse_version("0.9")

View File

@ -29,9 +29,9 @@ if sys.platform.startswith("win"):
"""Return an event loop and destroy it at the end of test""" """Return an event loop and destroy it at the end of test"""
loop = asyncio.ProactorEventLoop() loop = asyncio.ProactorEventLoop()
asyncio.set_event_loop(loop) # Replace main loop to avoid conflict between tests asyncio.set_loop(loop) # Replace main loop to avoid conflict between tests
yield loop yield loop
asyncio.set_event_loop(None) asyncio.set_loop(None)
@pytest.fixture(scope='function') @pytest.fixture(scope='function')
@ -72,7 +72,7 @@ def compute(controller):
@pytest.fixture @pytest.fixture
async def project(tmpdir, controller): async def project(loop, tmpdir, controller):
return await controller.add_project(name="Test") return await controller.add_project(name="Test")

View File

@ -55,7 +55,7 @@ async def test_start(gns3vm, controller):
assert gns3vm.password == "world" assert gns3vm.password == "world"
async def test_start_invalid_vm(gns3vm, controller): async def test_start_invalid_vm(loop, gns3vm, controller):
await controller.add_compute("r1", await controller.add_compute("r1",
name="R1", name="R1",

View File

@ -25,7 +25,7 @@ from gns3server.controller.gns3vm.virtualbox_gns3_vm import VirtualBoxGNS3VM
@pytest.fixture @pytest.fixture
async def gns3vm(controller): async def gns3vm(loop, controller):
vm = VirtualBoxGNS3VM(controller) vm = VirtualBoxGNS3VM(controller)
vm.vmname = "GNS3 VM" vm.vmname = "GNS3 VM"

View File

@ -21,7 +21,7 @@ from gns3server.controller.gns3vm.vmware_gns3_vm import VMwareGNS3VM
@pytest.fixture @pytest.fixture
async def gns3vm(controller): async def gns3vm(loop, controller):
vm = VMwareGNS3VM(controller) vm = VMwareGNS3VM(controller)
vm.vmname = "GNS3 VM" vm.vmname = "GNS3 VM"

View File

@ -33,7 +33,7 @@ from gns3server.utils.asyncio import aiozipstream
@pytest.fixture @pytest.fixture
async def project(controller): async def project(loop, controller):
p = Project(controller=controller, name="test") p = Project(controller=controller, name="test")
p.dump = MagicMock() p.dump = MagicMock()

View File

@ -47,7 +47,7 @@ def dummy_gns3vm(controller, dummy_engine):
return vm return vm
async def test_list(controller): async def test_list(loop, controller):
vm = GNS3VM(controller) vm = GNS3VM(controller)
with asyncio_patch("gns3server.controller.gns3vm.vmware_gns3_vm.VMwareGNS3VM.list", return_value=[{"vmname": "test", "vmx_path": "test"}]): with asyncio_patch("gns3server.controller.gns3vm.vmware_gns3_vm.VMwareGNS3VM.list", return_value=[{"vmname": "test", "vmx_path": "test"}]):
@ -60,14 +60,14 @@ async def test_list(controller):
await vm.list("hyperv") await vm.list("hyperv")
async def test_json(controller): async def test_json(loop, controller):
vm = GNS3VM(controller) vm = GNS3VM(controller)
assert vm.__json__() == vm._settings assert vm.__json__() == vm._settings
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not working well on Windows") @pytest.mark.skipif(sys.platform.startswith("win"), reason="Not working well on Windows")
async def test_update_settings(controller): async def test_update_settings(loop, controller):
vm = GNS3VM(controller) vm = GNS3VM(controller)
vm.settings = { vm.settings = {

View File

@ -166,7 +166,7 @@ async def test_import_with_images(tmpdir, controller):
assert os.path.exists(path), path assert os.path.exists(path), path
async def test_import_iou_linux_no_vm(linux_platform, tmpdir, controller): async def test_import_iou_linux_no_vm(loop, linux_platform, tmpdir, controller):
""" """
On non linux host IOU should be local if we don't have a GNS3 VM On non linux host IOU should be local if we don't have a GNS3 VM
""" """
@ -210,7 +210,7 @@ async def test_import_iou_linux_no_vm(linux_platform, tmpdir, controller):
assert topo["topology"]["nodes"][0]["compute_id"] == "local" assert topo["topology"]["nodes"][0]["compute_id"] == "local"
async def test_import_iou_linux_with_vm(linux_platform, tmpdir, controller): async def test_import_iou_linux_with_vm(loop, linux_platform, tmpdir, controller):
""" """
On non linux host IOU should be vm if we have a GNS3 VM configured On non linux host IOU should be vm if we have a GNS3 VM configured
""" """
@ -255,7 +255,7 @@ async def test_import_iou_linux_with_vm(linux_platform, tmpdir, controller):
assert topo["topology"]["nodes"][0]["compute_id"] == "vm" assert topo["topology"]["nodes"][0]["compute_id"] == "vm"
async def test_import_nat_non_linux(windows_platform, tmpdir, controller): async def test_import_nat_non_linux(loop, windows_platform, tmpdir, controller):
""" """
On non linux host NAT should be moved to the GNS3 VM On non linux host NAT should be moved to the GNS3 VM
""" """
@ -300,7 +300,7 @@ async def test_import_nat_non_linux(windows_platform, tmpdir, controller):
assert topo["topology"]["nodes"][0]["compute_id"] == "vm" assert topo["topology"]["nodes"][0]["compute_id"] == "vm"
async def test_import_iou_non_linux(windows_platform, tmpdir, controller): async def test_import_iou_non_linux(loop, windows_platform, tmpdir, controller):
""" """
On non linux host IOU should be moved to the GNS3 VM On non linux host IOU should be moved to the GNS3 VM
""" """
@ -356,7 +356,7 @@ async def test_import_iou_non_linux(windows_platform, tmpdir, controller):
mock.assert_called_with(controller._computes["vm"], project_id, project.path, os.path.join('project-files', 'iou', topo["topology"]["nodes"][0]['node_id'])) mock.assert_called_with(controller._computes["vm"], project_id, project.path, os.path.join('project-files', 'iou', topo["topology"]["nodes"][0]['node_id']))
async def test_import_node_id(linux_platform, tmpdir, controller): async def test_import_node_id(loop, linux_platform, tmpdir, controller):
""" """
When importing a node, node_id should change When importing a node, node_id should change
""" """
@ -449,7 +449,7 @@ async def test_import_node_id(linux_platform, tmpdir, controller):
assert os.path.exists(os.path.join(project.path, "project-files", "iou", topo["topology"]["nodes"][0]["node_id"], "startup.cfg")) assert os.path.exists(os.path.join(project.path, "project-files", "iou", topo["topology"]["nodes"][0]["node_id"], "startup.cfg"))
async def test_import_keep_compute_id(windows_platform, tmpdir, controller): async def test_import_keep_compute_id(loop, windows_platform, tmpdir, controller):
""" """
On linux host IOU should be moved to the GNS3 VM On linux host IOU should be moved to the GNS3 VM
""" """

View File

@ -53,7 +53,7 @@ def write_config(tmpdir, settings):
return path return path
def test_get_section_config(tmpdir): def test_get_section_config(loop, tmpdir):
config = load_config(tmpdir, { config = load_config(tmpdir, {
"Server": { "Server": {
@ -63,7 +63,7 @@ def test_get_section_config(tmpdir):
assert dict(config.get_section_config("Server")) == {"host": "127.0.0.1"} assert dict(config.get_section_config("Server")) == {"host": "127.0.0.1"}
def test_set_section_config(tmpdir): def test_set_section_config(loop, tmpdir):
config = load_config(tmpdir, { config = load_config(tmpdir, {
"Server": { "Server": {
@ -77,7 +77,7 @@ def test_set_section_config(tmpdir):
assert dict(config.get_section_config("Server")) == {"host": "192.168.1.1", "local": "true"} assert dict(config.get_section_config("Server")) == {"host": "192.168.1.1", "local": "true"}
def test_set(tmpdir): def test_set(loop, tmpdir):
config = load_config(tmpdir, { config = load_config(tmpdir, {
"Server": { "Server": {
@ -90,7 +90,7 @@ def test_set(tmpdir):
assert dict(config.get_section_config("Server")) == {"host": "192.168.1.1"} assert dict(config.get_section_config("Server")) == {"host": "192.168.1.1"}
def test_reload(tmpdir): def test_reload(loop, tmpdir):
config = load_config(tmpdir, { config = load_config(tmpdir, {
"Server": { "Server": {

View File

@ -44,7 +44,7 @@ async def test_exception_wait_run_in_executor():
@pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows") @pytest.mark.skipif(sys.platform.startswith("win"), reason="Not supported on Windows")
async def test_subprocess_check_output(tmpdir): async def test_subprocess_check_output(loop, tmpdir):
path = str(tmpdir / "test") path = str(tmpdir / "test")
result = await subprocess_check_output("echo", "-n", path) result = await subprocess_check_output("echo", "-n", path)

View File

@ -39,7 +39,7 @@ def response():
# assert response.status == 200 # assert response.status == 200
async def test_response_file_not_found(tmpdir, response): async def test_response_file_not_found(loop, tmpdir, response):
filename = str(tmpdir / 'hello-not-found') filename = str(tmpdir / 'hello-not-found')
with pytest.raises(HTTPNotFound): with pytest.raises(HTTPNotFound):