Как запустить процесс блокировки в фоновом потоке в python

Question

У меня есть тестовый скрипт, который нуждается в тестировании всего остального api. В то же время у меня также есть сервер xmpp, который генерирует сообщения.

Мне нужно запустить экземпляр xmpp внутри моего модульного теста для получения этих сообщений. Но проблема в том, что xmpp является процессом блокировки.

---> self.process (блок = True)

В результате модульного тестирования пробуксовки.

Есть ли способ запустить этот xmpp в фоновом потоке и продолжать получать сообщения и запускать модульный тест в основном потоке. Если да, могу ли я получить фрагмент кода, который я мог бы реализовать.

Заранее спасибо.

Answer 1

1. Одним из решений является запуск сервера в фоновом режиме в вашей setUp() рутина - т.е. os.system('myserver &'), а затем убить его, когда тест закончился, в tearDown()

2. Если вы хотите прямой контроль над сервером, используйте fork() и выполните примерно тот же шаблон, что и # 1.

Пример:

import os, sys, subprocess, time, unittest 

def server(): 
    try: 
     for _ in xrange(5, 0, -1): 
      print 'ding' 
      time.sleep(1) 
    except KeyboardInterrupt: 
     pass 


class TestClient(unittest.TestCase): 
    def setUp(self): 
     self.server_pid = None 
     pid = os.fork() 
     if not pid:    # child 
      return server() 
     # parent 
     self.server_pid = pid 

    def test1(self): 
     print 'test server, PID',self.server_pid 
     time.sleep(2) 

    def tearDown(self): 
     if not self.server_pid: 
      return 
     import signal 
     os.kill(self.server_pid, signal.SIGINT) 

Запуск с:

python -m unittest ptest 

Выходные:

тестовый сервер, PID-16490 Ding Ding .test сервер, ПИД Без

------------------------------------------------- --------------------- Ран 1 тест в 2.003s

OK