How to do unit test on Autobahn applications using Twisted Trial?

Question

Assuming that you use only Autobahn connection (not raw WebSocket).

How can we test WITHOUT network, our RPC methods and events?
As it is Twisted, I think that the most appropriate tool would be Twisted Trial.

But I can't figure out how should I write those tests, without writing a lot of boilerplate code and re-using internal implementation of Autobahn (and even I'm not sure that I would be able to do it this way).

How would you do it?

Acetanilide

Answer 1

It's an attempt to answer my own question.

The problem

Then, to unit test RPC methods and events, we need to assume that Autobahn is well tested and we don't have to test it, the solution becomes simple:

The solution

Mock all.

Context

In my application, I have two types of components (read ApplicationSession): StandardComponent and DatabaseComponent (which inherits from StandardComponent).

The biggest problem during a unit testing is that we have a lot of dependencies, like a database connection, a Redis connection, etc...

Examples

What I do in my tests is to patch all those objects, by subclassing unittest.TestCase:

class APITestCase(unittest.TestCase):

    def _patchObject(self, module_name, **kwargs):
        patcher = patch(module_name, **kwargs)
        mock = patcher.start()

        self.patches.append(patcher)
        return mock

    def setUp(self):
        logging.disable(logging.CRITICAL)
        self.patches = []
        self.session = self._patchObject('components.ApplicationAPI')
        self.database = self._patchObject('txpostgres.txpostgres.Connection')

    def tearDown(self):
        for patcher in self.patches:
            patcher.stop()

I inject a mocked session and a mocked database in my test case.

Then, testing becomes very very simple.

Whenever, I call an RPC method which needs to call the database or get results from the database, I patch it: self.mocked_auth_user.return_value = (1, "abc", "something", "admin")

And in my test method:

def test_authenticate_success(self):
    self.mocked_auth_user.return_value = (1, "abc", "paris", "admin")

    def _doTest(auth_result):
        attempted_auth_result = {
            "secret": "abc",
            "role": "admin",
            "authid": "1",
            "salt": "paris",
            "iterations": 1000,
            "keylen": 32
        }

        self.assertEqual(auth_result, attempted_auth_result)
        self.mocked_auth_user.assert_called_with(self.api.database, "raito")

    return self.api.authenticate("test", "raito", {}).addCallback(_doTest)

You can do some more advanced and interesting tests to see if your method is fail-proof:

def test_authenticate_authid_not_found(self):
    def _raiseException(db, user):
        return defer.fail(Exception("User {} not found!".format(user)))

    self.mocked_auth_user.side_effect = _raiseException
    return self.failUnlessFailure(self.api.authenticate("test", "raito", {}), AuthenticationError)

The same goes for events, you just need to call them and test if they publish an event or not (self.session.publish.assert_called_with(...))

It becomes magic!

Anyway, it solves the unit testing problem, but the integration is yet. I'm working on it, but the problem will likely be solved using some virtualization tech (Docker) or something like that.

Acetanilide