Как я могу запустить Thrift-сервер или TCP-сервер в качестве приспособления Pytest?

Question

платформа linux2, python 2.7.12-final-0

Мои интеграционные тесты проверяют связь с внешним сервером, но я хотел бы изменить это, чтобы использовать тестовое устройство.

Я несколько дней пытался запустить TCP-сервер библиотеки Thrift в качестве фиксатора области сеанса в pytest. Я пытался запустить сервер в качестве фонового потока, чтобы мои тесты не блокировались.

@pytest.fixture(scope="session", autouse=True)
def thrift_server():
    config = get_config()
    nprocesses = 4

    try:
        nprocesses = config['num_processes']
    except:
        pass

    args = (Handler, config['db_credentials'], config['server_port'])
    kwargs = ({'env': config['env'], 'processpool': True, 'num_processes': nprocesses,
        'handler_config': config, 'logfile': 'tests/test_server_log.txt'})

    tserver = ThriftServer()
    tserver.add_path(config['thrift_path'])
    tserver.set_service("search")

    try:
        thread = threading.Thread(target=tserver.runserver, args=args, kwargs=kwargs)
        thread.daemon = True
        thread.start()
        yield tserver
    except:
        print("BOOHOO")
    print("TEARDOWN: test server")

Для каждого из моих тестов я получаю сообщение об ошибке из кода библиотеки pytest:

ERROR at setup of Tests.test_item_is_found 
venv/lib/python2.7/site-packages/_pytest/runner.py:226: in from_call
    result = func()
venv/lib/python2.7/site-packages/_pytest/runner.py:198: in <lambda>
    lambda: ihook(item=item, **kwds), when=when, reraise=reraise
venv/lib/python2.7/site-packages/pluggy/hooks.py:289: in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
venv/lib/python2.7/site-packages/pluggy/manager.py:87: in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
venv/lib/python2.7/site-packages/pluggy/manager.py:81: in <lambda>
    firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
venv/lib/python2.7/site-packages/_pytest/runner.py:116: in pytest_runtest_setup
    item.session._setupstate.prepare(item)
venv/lib/python2.7/site-packages/_pytest/runner.py:362: in prepare
    col.setup()
venv/lib/python2.7/site-packages/_pytest/unittest.py:119: in setup
    self._request._fillfixtures()
venv/lib/python2.7/site-packages/_pytest/fixtures.py:469: in _fillfixtures
    item.funcargs[argname] = self.getfixturevalue(argname)
venv/lib/python2.7/site-packages/_pytest/fixtures.py:479: in getfixturevalue
    return self._get_active_fixturedef(argname).cached_result[0]
venv/lib/python2.7/site-packages/_pytest/fixtures.py:502: in _get_active_fixturedef
    self._compute_fixture_value(fixturedef)
venv/lib/python2.7/site-packages/_pytest/fixtures.py:587: in _compute_fixture_value
    fixturedef.execute(request=subrequest)
venv/lib/python2.7/site-packages/_pytest/fixtures.py:894: in execute
    return hook.pytest_fixture_setup(fixturedef=self, request=request)
venv/lib/python2.7/site-packages/pluggy/hooks.py:289: in __call__
    return self._hookexec(self, self.get_hookimpls(), kwargs)
venv/lib/python2.7/site-packages/pluggy/manager.py:87: in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
venv/lib/python2.7/site-packages/pluggy/manager.py:81: in <lambda>
    firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
venv/lib/python2.7/site-packages/_pytest/fixtures.py:936: in pytest_fixture_setup
    result = call_fixture_func(fixturefunc, request, kwargs)
venv/lib/python2.7/site-packages/_pytest/fixtures.py:791: in call_fixture_func
    res = next(it)
E   StopIteration

Это проблема, для которой я не нашел отличной документации или поддержки, поэтому я должен задать вопрос .. Как я могу настроить TCP-сервер в качестве приспособления pytest, которое могут использовать мои тесты?

Как следствие, это ситуация, когда я должен бросить на себя или есть хороший плагин для поддержки этого сценария использования Pytest? Опять же, мои поиски мало что дали по этому поводу.

Answer 1

Мне удалось настроить сервер Thrift в качестве устройства Pytest, используя следующую настройку:

import pytest
import threading
from pythrift import ThriftServer
from myHandler import Handler


@pytest.fixture(scope="session", autouse=True)
def thrift_server():
     config = get_config()
     nprocesses = 4

     try:
         nprocesses = config['num_processes']
     except:
         pass

     args = (Handler, config['db_credentials'], config['server_port'])
     kwargs = ({'env': config['env'], 'processpool': True, 'num_processes': nprocesses,
         'handler_config': config, 'logfile': 'tests/test_server_log.txt'})

     tserver = ThriftServer()
     tserver.add_path(config['thrift_path'])
     tserver.set_service("search")

     try:
         thread = threading.Thread(target=tserver.run_server, args=args, kwargs=kwargs)
         thread.daemon = True
         thread.start()
         yield tserver
     except Exception as e:
         print(e)
     print("TEARDOWN: test server")