How do I write tests for Cyclone in the style of Tornado?

Question

I have been googling and asking on IRC to no avail. Cyclone is supposed to be a Tornado-like protocol for Twisted. But, there are no tests in the Cyclone repository and no-one has written up how to convert tornado.testing.AsyncHTTPTestCase tests to exercise code written against Cyclone.

1. How do I start a server to test the web interface?
2. Where is the self.fetch()?
3. Where is the documentation in Cyclone to describe how to convert an existing Tornado app?

Answer 1

Unfortunately, there's nothing like tornado.testing.AsyncHTTPTestCase in cyclone at the moment. Your best bet would be to use Twisted Trial to write unit tests. One (slightly kludgy) approach would be explicitly call self.listener = reactor.listenTCP(<someport>, YourCycloneApplication()) in the setUp method inside your test case and call self.listener.stopListening() in the tearDown method.

Then, inside your test methods, you could use cyclone.httpclient.fetch to fetch the pages. This is far from ideal. But as of now, this is the only way to go.

Answer 2

Here is what we are currently using to test our cylcone handler like we did with tornado:

from twisted.trial.unittest import TestCase
from twisted.internet import defer, reactor
from cyclone import httpclient

# copied from tornado
_next_port = 10000
def get_unused_port():
    """Returns a (hopefully) unused port number."""
global _next_port
    port = _next_port
    _next_port = _next_port + 1
    return port

class TxTestCase(TestCase):

    def get_http_port(self):
        """Returns the port used by the HTTPServer.

        A new port is chosen for each test.
        """
        if self.__port is None:
            self.__port = get_unused_port()
        return self.__port

    def setUp(self, *args, **kwargs):
        self.__port = None
        self._app = self.get_app()
        self._listener = None
        if self._app:
            self._listener = reactor.listenTCP(self.get_http_port(), self._app)
        return TestCase.setUp(self, *args, **kwargs)

    def get_app(self):
        return None

    def tearDown(self):
        if self._listener:
            self._listener.stopListening()

    @defer.inlineCallbacks
    def fetch(self, url, *args, **kwargs):
        response = yield httpclient.fetch('http://localhost:%s%s'%(self.get_http_port(), url), *args, **kwargs)
        defer.returnValue(response)

This way, you get the fetch method back ;)

And there are no more needs to use trial.

Here is an usage example:

from twisted.internet import defer

class Test(TxTestCase):

    def get_app(self):
        return MyApplication()

    @defer.inlineCallbacks
    def some_test_method(self):
        res = yield self.fetch('/path/to/resource')
        self.assertEquals(200, res.code)

Hope that will help you.