CODEWITHSUNDEEP
python-3.xtornadopytestpytest-asyncio
Conor
Conor

Reputation: 593

tornado_json unit testing async methods

I'm trying to set up a unit test for a tornado_json web app. I'm trying to test a post handler, but I'm failing miserably as the fetch method seems to return an _asyncio.Future object, which never seems to complete/have a result set. I've tried to post a summary of the code, at the moment I'm just returning ['test'] item. I've looked at https://github.com/tornadoweb/tornado/issues/1154, as well as the tornado documentation. It sounds like I need to self.stop, or self.wait() to complete the task, but I haven't worked out how to get this to work, or if that is the solution. Any help would be greatly appreciated.

@schema.validate( input_schema={ "type": "object", "properties": { "path": {"type": "string"} }, "required": ["path"] }, output_schema={ "type": "array", "items": { "properties": {"type": "string"} } } ) @coroutine def post(self): attributes = dict(self.body) path = attributes["path"] response = ["test"] return response

@gen_test def test_POST_method(self):
body = json.dumps({'path': 'bin'}) self.http_client.fetch(self.get_url('/api/listmyfiles'), method="POST", body=body ) response = self.wait() print(response.result()))

The error I get is: asyncio.base_futures.InvalidStateError: Result is not set.

Upvotes: 0

Views: 182

Answers (1)

Ben Darnell
Ben Darnell

Reputation: 22134

AsyncHTTPTestCase has a few different modes of operation that can't be mixed.

  1. @gen_test: Used with await self.http_client.fetch(self.get_url(...)):

    @gen_test
async def test_post_method(self):
    response = await self.http_client.fetch(self.get_url(...))

  2. self.stop/self.wait is an older interface that is mostly (but not completely) deprecated. AsyncHTTPClient will not be (easily) compatible with this interface in Tornado 6.0, so I won't show an example here.

  3. self.fetch is a shorthand method that combines the calls to http_client.fetch and self.get_url, and uses stop/wait under the covers (so it is not compatible with @gen_test):

    def test_post_method(self):
    response = self.fetch('/api/listmyfiles')

If the only asynchronous thing you're doing is HTTP fetches, you can use self.fetch. If you need to do anything else asynchronous, use gen_test and avoid the stop/wait/self.fetch methods.

Upvotes: 3

Related Questions