2

我有一个扭曲的 Web 资源，它处理传入的请求并发送要在队列中处理的 celery 任务，这个 celery 调用是使用 reactor.callInThread 处理的

问题是我想对这段代码进行单元测试，但是像下面这样的简单 TestCase 失败了，因为 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中描述的方法来测试网络资源

$ 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 ====================================================================================
4

1 回答 1

1

reactor在测试期间未运行，因此没有线程池。您可以先拨打虚拟电话reactor.callInThread进行测试。threads.deferToThread

于 2012-08-31T18:04:28.043 回答