如何使用 Twisted Trial 对 Autobahn 应用程序进行单元测试？

Posted 2023-04-18

【中文标题】如何使用 Twisted Trial 对 Autobahn 应用程序进行单元测试？

【英文标题】：How to do unit test on Autobahn applications using Twisted Trial?

【发布时间】：2015-05-25 19:58:48

【问题描述】：

假设您只使用高速公路连接（不是原始 WebSocket）。

我们如何测试没有网络、我们的 RPC 方法和事件？ 因为它是 Twisted，我认为最合适的工具是 Twisted Trial。

但是我不知道我应该如何编写这些测试，而不编写大量样板代码并重新使用 Autobahn 的内部实现（甚至我不确定我是否能够这样做)。

你会怎么做？

【参考方案1】：

这是试图回答我自己的问题。

问题

那么，要对 RPC 方法和事件进行单元测试，我们需要假设 Autobahn 已经过良好测试并且我们不必对其进行测试，解决方案就变得简单了：

解决方案

全部模拟。

上下文

在我的应用程序中，我有两种类型的组件（读取 ApplicationSession）：StandardComponent 和 DatabaseComponent（继承自 StandardComponent）。

单元测试最大的问题是我们有很多依赖，比如一个数据库连接，一个Redis连接等等……

示例

我在测试中所做的是通过子类化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()

我在我的测试用例中注入了一个模拟会话和一个模拟数据库。

然后，测试变得非常非常简单。

每当我调用需要调用数据库或从数据库获取结果的 RPC 方法时，我都会对其进行修补： self.mocked_auth_user.return_value = (1, "abc", "something", "admin")

在我的测试方法中：

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)

你可以做一些更高级和更有趣的测试来看看你的方法是否是防故障的：

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)

事件也是如此，您只需要调用它们并测试它们是否发布事件（self.session.publish.assert_called_with(...)）

它变成了魔法！

不管怎样，它解决了单元测试问题，但集成尚未完成。我正在努力，但问题可能会使用一些虚拟化技术（Docker）或类似的东西来解决。