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

Move fixtures to conftest

This commit is contained in:
Julien Duponchelle 2015-01-21 22:32:33 +01:00
parent f3e07d5ad9
commit 97cefa23fb
8 changed files with 99 additions and 89 deletions

View File

@ -21,19 +21,9 @@
import json import json
import re import re
import asyncio import asyncio
import socket
import pytest
from aiohttp import web
import aiohttp import aiohttp
import os 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: class Query:
@ -125,53 +115,3 @@ class Query:
def _example_file_path(self, method, path): def _example_file_path(self, method, path):
path = re.sub('[^a-z0-9]', '', path) path = re.sub('[^a-z0-9]', '', path)
return "docs/api/examples/{}_{}.txt".format(method.lower(), 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")

View File

@ -20,8 +20,7 @@ This test suite check /project endpoint
""" """
from tests.utils import asyncio_patch, port_manager from tests.utils import asyncio_patch
from tests.api.base import server, loop
from gns3server.version import __version__ from gns3server.version import __version__

View File

@ -20,8 +20,7 @@ This test suite check /version endpoint
It's also used for unittest the HTTP implementation. It's also used for unittest the HTTP implementation.
""" """
from tests.utils import asyncio_patch, port_manager from tests.utils import asyncio_patch
from tests.api.base import server, loop
from gns3server.version import __version__ from gns3server.version import __version__

View File

@ -15,8 +15,7 @@
# You should have received a copy of the GNU General Public License # You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
from tests.api.base import server, loop, project from tests.utils import asyncio_patch
from tests.utils import asyncio_patch, port_manager
from gns3server.modules.virtualbox.virtualbox_vm import VirtualBoxVM from gns3server.modules.virtualbox.virtualbox_vm import VirtualBoxVM

View File

@ -17,8 +17,7 @@
import pytest import pytest
import os import os
from tests.api.base import server, loop, project from tests.utils import asyncio_patch
from tests.utils import asyncio_patch, free_console_port, port_manager
from unittest.mock import patch, Mock from unittest.mock import patch, Mock
from gns3server.modules.vpcs.vpcs_vm import VPCSVM from gns3server.modules.vpcs.vpcs_vm import VPCSVM

93
tests/conftest.py Normal file
View File

@ -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 <http://www.gnu.org/licenses/>.
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

View File

@ -18,10 +18,9 @@
import pytest import pytest
import asyncio import asyncio
import os 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 asyncio.subprocess import Process
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
from gns3server.modules.vpcs.vpcs_vm import VPCSVM from gns3server.modules.vpcs.vpcs_vm import VPCSVM

View File

@ -16,10 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio import asyncio
import pytest
from unittest.mock import patch from unittest.mock import patch
from gns3server.modules.project_manager import ProjectManager
from gns3server.modules.port_manager import PortManager
class _asyncio_patch: class _asyncio_patch:
@ -57,18 +54,3 @@ class _asyncio_patch:
def asyncio_patch(function, *args, **kwargs): def asyncio_patch(function, *args, **kwargs):
return _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