diff --git a/tests/api/base.py b/tests/api/base.py index 638d53cc..01d460d1 100644 --- a/tests/api/base.py +++ b/tests/api/base.py @@ -21,19 +21,9 @@ import json import re import asyncio -import socket -import pytest -from aiohttp import web import aiohttp import os -from gns3server.web.route import Route -# TODO: get rid of * -from gns3server.handlers import * -from gns3server.modules import MODULES -from gns3server.modules.port_manager import PortManager -from gns3server.modules.project_manager import ProjectManager - class Query: @@ -125,53 +115,3 @@ class Query: def _example_file_path(self, method, path): path = re.sub('[^a-z0-9]', '', path) return "docs/api/examples/{}_{}.txt".format(method.lower(), path) - - -def _get_unused_port(): - """ Return an unused port on localhost. In rare occasion it can return - an already used port (race condition)""" - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.bind(('localhost', 0)) - addr, port = s.getsockname() - s.close() - return port - - -@pytest.fixture(scope="session") -def loop(request): - """Return an event loop and destroy it at the end of test""" - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) # Replace main loop to avoid conflict between tests - - def tear_down(): - loop.close() - asyncio.set_event_loop(None) - request.addfinalizer(tear_down) - return loop - - -@pytest.fixture(scope="session") -def server(request, loop, port_manager): - port = _get_unused_port() - host = "localhost" - app = web.Application() - for method, route, handler in Route.get_routes(): - app.router.add_route(method, route, handler) - for module in MODULES: - instance = module.instance() - instance.port_manager = port_manager - srv = loop.create_server(app.make_handler(), host, port) - srv = loop.run_until_complete(srv) - - def tear_down(): - for module in MODULES: - loop.run_until_complete(module.destroy()) - srv.close() - srv.wait_closed() - request.addfinalizer(tear_down) - return Query(loop, host=host, port=port) - - -@pytest.fixture(scope="module") -def project(): - return ProjectManager.instance().create_project(uuid="a1e920ca-338a-4e9f-b363-aa607b09dd80") diff --git a/tests/api/test_project.py b/tests/api/test_project.py index b7ae8b55..7c7d1c6b 100644 --- a/tests/api/test_project.py +++ b/tests/api/test_project.py @@ -20,8 +20,7 @@ This test suite check /project endpoint """ -from tests.utils import asyncio_patch, port_manager -from tests.api.base import server, loop +from tests.utils import asyncio_patch from gns3server.version import __version__ diff --git a/tests/api/test_version.py b/tests/api/test_version.py index 5b479006..76e7db72 100644 --- a/tests/api/test_version.py +++ b/tests/api/test_version.py @@ -20,8 +20,7 @@ This test suite check /version endpoint It's also used for unittest the HTTP implementation. """ -from tests.utils import asyncio_patch, port_manager -from tests.api.base import server, loop +from tests.utils import asyncio_patch from gns3server.version import __version__ diff --git a/tests/api/test_virtualbox.py b/tests/api/test_virtualbox.py index 6ce767aa..820bf34a 100644 --- a/tests/api/test_virtualbox.py +++ b/tests/api/test_virtualbox.py @@ -15,8 +15,7 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . -from tests.api.base import server, loop, project -from tests.utils import asyncio_patch, port_manager +from tests.utils import asyncio_patch from gns3server.modules.virtualbox.virtualbox_vm import VirtualBoxVM diff --git a/tests/api/test_vpcs.py b/tests/api/test_vpcs.py index 029e4db1..5fed9aa9 100644 --- a/tests/api/test_vpcs.py +++ b/tests/api/test_vpcs.py @@ -17,8 +17,7 @@ import pytest import os -from tests.api.base import server, loop, project -from tests.utils import asyncio_patch, free_console_port, port_manager +from tests.utils import asyncio_patch from unittest.mock import patch, Mock from gns3server.modules.vpcs.vpcs_vm import VPCSVM diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..572e3f00 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2015 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 . + +import pytest +import socket +import asyncio +from aiohttp import web +from gns3server.web.route import Route +# TODO: get rid of * +from gns3server.handlers import * +from gns3server.modules import MODULES +from gns3server.modules.port_manager import PortManager +from gns3server.modules.project_manager import ProjectManager +from tests.api.base import Query + + +@pytest.fixture(scope="session") +def loop(request): + """Return an event loop and destroy it at the end of test""" + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) # Replace main loop to avoid conflict between tests + + def tear_down(): + loop.close() + asyncio.set_event_loop(None) + request.addfinalizer(tear_down) + return loop + + +def _get_unused_port(): + """ Return an unused port on localhost. In rare occasion it can return + an already used port (race condition)""" + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(('localhost', 0)) + addr, port = s.getsockname() + s.close() + return port + + +@pytest.fixture(scope="session") +def server(request, loop, port_manager): + port = _get_unused_port() + host = "localhost" + app = web.Application() + for method, route, handler in Route.get_routes(): + app.router.add_route(method, route, handler) + for module in MODULES: + instance = module.instance() + instance.port_manager = port_manager + srv = loop.create_server(app.make_handler(), host, port) + srv = loop.run_until_complete(srv) + + def tear_down(): + for module in MODULES: + loop.run_until_complete(module.destroy()) + srv.close() + srv.wait_closed() + request.addfinalizer(tear_down) + return Query(loop, host=host, port=port) + + +@pytest.fixture(scope="module") +def project(): + return ProjectManager.instance().create_project(uuid="a1e920ca-338a-4e9f-b363-aa607b09dd80") + + +@pytest.fixture(scope="session") +def port_manager(): + return PortManager("127.0.0.1", False) + + +@pytest.fixture(scope="function") +def free_console_port(request, port_manager): + # In case of already use ports we will raise an exception + port = port_manager.get_free_console_port() + # We release the port immediately in order to allow + # the test do whatever the test want + port_manager.release_console_port(port) + return port diff --git a/tests/modules/vpcs/test_vpcs_vm.py b/tests/modules/vpcs/test_vpcs_vm.py index cc10cb7f..9887c97b 100644 --- a/tests/modules/vpcs/test_vpcs_vm.py +++ b/tests/modules/vpcs/test_vpcs_vm.py @@ -18,10 +18,9 @@ import pytest import asyncio import os -from tests.utils import asyncio_patch, port_manager, free_console_port +from tests.utils import asyncio_patch + -# TODO: Move loop to util -from tests.api.base import loop, project from asyncio.subprocess import Process from unittest.mock import patch, MagicMock from gns3server.modules.vpcs.vpcs_vm import VPCSVM diff --git a/tests/utils.py b/tests/utils.py index 55069f5e..bc8dc5dc 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -16,10 +16,7 @@ # along with this program. If not, see . import asyncio -import pytest from unittest.mock import patch -from gns3server.modules.project_manager import ProjectManager -from gns3server.modules.port_manager import PortManager class _asyncio_patch: @@ -57,18 +54,3 @@ class _asyncio_patch: def asyncio_patch(function, *args, **kwargs): return _asyncio_patch(function, *args, **kwargs) - - -@pytest.fixture(scope="session") -def port_manager(): - return PortManager("127.0.0.1", False) - - -@pytest.fixture(scope="function") -def free_console_port(request, port_manager): - # In case of already use ports we will raise an exception - port = port_manager.get_free_console_port() - # We release the port immediately in order to allow - # the test do whatever the test want - port_manager.release_console_port(port) - return port