不能在我的扭曲的網絡資源測試reactor.callInThread

Question

我有一個扭曲的Web資源至極處理傳入的請求，併發送芹菜任務在隊列中進行處理，這芹菜呼叫使用reactor.callInThread

的處理事情是，我想這個單元測試代碼，但因爲在reactor.callInThread代碼永遠不會執行

from twisted.internet.defer import succeed 
from twisted.web import server 
from twisted.web.resource import Resource 
from twisted.web.test.test_web import DummyRequest 
from twisted.internet.defer import inlineCallbacks 
from twisted.trial.unittest import TestCase 
from twisted.internet import reactor 

import pytest 

class BadBeConnectedResultType(Exception): 
    pass 

class SmartDummyRequest(DummyRequest): 
    def __init__(self, method, url, args=None, headers=None): 
     DummyRequest.__init__(self, url.split('/')) 
     self.method = method 
     self.headers.update(headers or {}) 

     # set args 
     args = args or {} 
     for k, v in args.items(): 
      self.addArg(k, v) 


    def value(self): 
     return "".join(self.written) 


class DummySite(server.Site): 
    def get(self, url, args=None, headers=None): 
     return self._request("GET", url, args, headers) 


    def post(self, url, args=None, headers=None): 
     return self._request("POST", url, args, headers) 


    def _request(self, method, url, args, headers): 
     request = SmartDummyRequest(method, url, args, headers) 
     resource = self.getResourceFor(request) 
     result = resource.render(request) 
     return self._resolveResult(request, result) 


    def _resolveResult(self, request, result): 
     if isinstance(result, str): 
      request.write(result) 
      request.finish() 
      return succeed(request) 
     elif result is server.NOT_DONE_YET: 
      if request.finished: 
       return succeed(request) 
      else: 
       return request.notifyFinish().addCallback(lambda _: request) 
     else: 
      raise ValueError("Unexpected return value: %r" % (result,)) 

class TwistedBCTestCase(TestCase): 
    def setUp(self): 
     self.web = DummySite(RootResource()) 

    @inlineCallbacks 
    def test_bc_resource(self): 
     with pytest.raises(BadBeConnectedResultType): 
      response = yield self.web.post("CLARO", args={ 
       'ResultType':2, 
       'ResultData':1, 
       'idDeliver': 1 
      }) 

class RootResource(Resource): 

    def render_GET(self, request): 
     return '' 

    def getChild(self, operator_id, request): 
     if operator_id: 
      return BCResource(operator_id) 

     return self 

class BCResource(Resource): 

    def __init__(self, operator_id): 
     Resource.__init__(self) 
     self.operator_id = operator_id 

    def send_to_queue(self, request): 
     #This is never executed in my test 
     try: 
      result_type = cgi.escape(request.args['ResultType'][0]) 
      if result_type != '1': 
       raise BadBeConnectedResultType(result_type) 

      raw_result_data = cgi.escape(request.args['ResultData'][0]) 
      unique_id = cgi.escape(request.args['IdDeliver'][0]) 


      task = ProcessBCNotificationTask.delay(unique_id, raw_result_data) 

     except BadBeConnectedResultType as e: 
      logger.exception(e) 


    def render_POST(self, request): 
     reactor.callInThread(self.send_to_queue, request) 
     return '' 

爲了記錄我使用http://findingscience.com/python/twisted/2012/02/20/testing-twisted-web-resources.html描述的方法來測試Web類似下面的簡單的測試用例失敗資源

$ py.test twisted_test.py --tb=short 
====================================================================================== test session starts ====================================================================================== 
platform linux2 -- Python 2.7.3 -- pytest-2.2.4 
collected 1 items 

twisted_test.py F 

=========================================================================================== FAILURES ============================================================================================ 
______________________________________________________________________________ TwistedBCTestCase.test_bc_resource _______________________________________________________________________________ 
env/lib/python2.7/site-packages/twisted/internet/defer.py:1045: in _inlineCallbacks 
>     result = g.send(result) 
twisted_test.py:69: in test_bc_resource 
>    'idDeliver': 1 
E Failed: DID NOT RAISE 
=================================================================================== 1 failed in 0.30 seconds ==================================================================================== 

Answer 1

reactor在測試過程中沒有運行，所以沒有線程池。您可以通過首先製作dummy threads.deferToThread call來測試reactor.callInThread。