使用 Twisted 进行测试驱动开发¶
编写好的代码很难，或者至少可能很难。一个主要挑战是确保在添加新功能时代码保持正确。
单元测试 是一种现代的轻量级测试方法，在许多编程语言中被广泛使用。依赖于单元测试的开发通常被称为测试驱动开发 (TDD )。大多数 Twisted 代码都是使用 TDD 进行测试的。
要深入了解 Python 中的单元测试，您应该阅读 unittest -- Unit testing framework chapter 的 Python 库参考。网上和书籍中有很多相关信息。
Python 单元测试的入门示例¶
本文档主要介绍 Trial，Twisted 的单元测试框架。Trial 基于 Python 的单元测试框架。虽然我们不打算提供关于一般 Python 单元测试的全面指南，但在扩展到涵盖需要 Trial 特殊功能的网络代码之前，考虑一个简单的非网络示例将有所帮助。如果您已经熟悉 Python 中的单元测试，请直接跳到关于 测试 Twisted 代码 的部分。
注意
在接下来的内容中，我们将对一些简单的类进行一系列改进。为了使示例和源代码链接完整，并允许您在每个阶段对中间结果运行 Trial，我在文件名中添加了 _N（其中 N 是连续的整数）以将它们分开。这只是一个轻微的视觉干扰，可以忽略。
创建 API 并编写测试¶
我们将创建一个用于算术计算的库。首先，创建一个包含名为 calculus 的目录的项目结构，该目录包含一个空的 __init__.py 文件。
然后将以下简单的类定义 API 放入 calculus/base_1.py 中
# -*- test-case-name: calculus.test.test_base_1 -*-
class Calculation:
def add(self, a, b):
pass
def subtract(self, a, b):
pass
def multiply(self, a, b):
pass
def divide(self, a, b):
pass
（现在忽略 test-case-name 注释。您将在 下面 看到为什么它有用。）
我们已经编写了接口，但没有编写代码。现在我们将编写一组测试。在开发的这个阶段，我们预计所有测试都会失败。别担心，这是重点的一部分。一旦我们拥有一个正常运行的测试框架，并且编写了一些不错的测试（并且失败了！），我们就会开始实际开发我们的计算 API。对于许多使用 TDD 的人来说，这是首选的工作方式——先编写测试，确保它们失败，然后进行开发。其他人没有那么严格，在进行开发后编写测试。
在 calculus 下面创建一个名为 test 的目录，其中包含一个空的 __init__.py 文件。在 calculus/test/test_base_1.py 中，放入以下内容
from calculus.base_1 import Calculation
from twisted.trial import unittest
class CalculationTestCase(unittest.TestCase):
def test_add(self):
calc = Calculation()
result = calc.add(3, 8)
self.assertEqual(result, 11)
def test_subtract(self):
calc = Calculation()
result = calc.subtract(7, 3)
self.assertEqual(result, 4)
def test_multiply(self):
calc = Calculation()
result = calc.multiply(12, 5)
self.assertEqual(result, 60)
def test_divide(self):
calc = Calculation()
result = calc.divide(12, 5)
self.assertEqual(result, 2)
您现在应该拥有以下 4 个文件
calculus/__init__.py
calculus/base_1.py
calculus/test/__init__.py
calculus/test/test_base_1.py
要运行测试，您必须确保能够加载它们。确保您位于包含 calculus 文件夹的目录中，如果您运行 ls 或 dir，您应该看到该文件夹。您可以通过运行 python -c import calculus 来测试是否可以导入 calculus 包。如果它报告错误（“没有名为 calculus 的模块”），请仔细检查您是否位于正确的目录中。
当您位于包含 calculus 目录的目录中时，从命令行运行 python -m twisted.trial calculus.test.test_base_1。
您应该看到以下输出（尽管您的文件可能不在 /tmp 中）
$ python -m twisted.trial calculus.test.test_base_1
calculus.test.test_base_1
CalculationTestCase
test_add ... [FAIL]
test_divide ... [FAIL]
test_multiply ... [FAIL]
test_subtract ... [FAIL]
===============================================================================
[FAIL]
Traceback (most recent call last):
File "/tmp/calculus/test/test_base_1.py", line 8, in test_add
self.assertEqual(result, 11)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 11
calculus.test.test_base_1.CalculationTestCase.test_add
===============================================================================
[FAIL]
Traceback (most recent call last):
File "/tmp/calculus/test/test_base_1.py", line 23, in test_divide
self.assertEqual(result, 2)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 2
calculus.test.test_base_1.CalculationTestCase.test_divide
===============================================================================
[FAIL]
Traceback (most recent call last):
File "/tmp/calculus/test/test_base_1.py", line 18, in test_multiply
self.assertEqual(result, 60)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 60
calculus.test.test_base_1.CalculationTestCase.test_multiply
===============================================================================
[FAIL]
Traceback (most recent call last):
File "/tmp/calculus/test/test_base_1.py", line 13, in test_subtract
self.assertEqual(result, 4)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 4
calculus.test.test_base_1.CalculationTestCase.test_subtract
-------------------------------------------------------------------------------
Ran 4 tests in 0.042s
FAILED (failures=4)
如何解释此输出？您将获得一个包含各个测试的列表，每个测试后面跟着其结果。默认情况下，失败的测试将在最后打印，但这可以通过 -e（或 --rterrors）选项更改。
此输出中一个非常有用的东西是失败测试的完全限定名称。它出现在输出中每个 =-delimited 区域的底部。这使您可以复制并粘贴它以仅运行您感兴趣的单个测试。在我们的示例中，您可以从 shell 中运行 python -m twisted.trial calculus.test.test_base_1.CalculationTestCase.test_subtract。
请注意，trial 可以使用不同的报告器来修改其输出。运行 python -m twisted.trial --help-reporters 以查看报告器列表。
Trial 可以通过多种方式运行测试
python -m twisted.trial calculus：运行 calculus 包的所有测试。python -m twisted.trial calculus.test：使用 Python 的import符号运行。python -m twisted.trial calculus.test.test_base_1：如上所述，用于特定的测试模块。您可以通过放置您的类名甚至方法名来遵循该逻辑，以仅运行那些特定的测试。python -m twisted.trial --testmodule=calculus/base_1.py：使用base_1.py的第一行中的test-case-name注释查找测试。python -m twisted.trial calculus/test：运行测试目录中的所有测试（不推荐）。python -m twisted.trial calculus/test/test_base_1.py: 运行特定测试文件（不推荐）。
强烈建议使用前 3 个版本使用完全限定名：它们更可靠，并且允许您轻松地在测试运行中更具选择性。
您会注意到，Trial 在运行测试的目录中创建了一个名为 _trial_temp 的目录。该目录中有一个名为 test.log 的文件，其中包含测试的日志输出（使用 log.msg 或 log.err 函数创建）。如果您在测试中添加了日志记录，请检查此文件。
使测试通过¶
现在我们已经建立了一个可用的测试框架，并且我们的测试失败了（如预期），我们可以尝试实现正确的 API。我们将在上面 base_1 模块的新版本 calculus/base_2.py 中执行此操作。
# -*- test-case-name: calculus.test.test_base_2 -*-
class Calculation:
def add(self, a, b):
return a + b
def subtract(self, a, b):
return a - b
def multiply(self, a, b):
return a * b
def divide(self, a, b):
return a // b
我们还将在 calculus/test_base_2.py 中创建一个 test_base_1 的新版本，该版本导入并测试此新实现。
from calculus.base_2 import Calculation
from twisted.trial import unittest
class CalculationTestCase(unittest.TestCase):
def test_add(self):
calc = Calculation()
result = calc.add(3, 8)
self.assertEqual(result, 11)
def test_subtract(self):
calc = Calculation()
result = calc.subtract(7, 3)
self.assertEqual(result, 4)
def test_multiply(self):
calc = Calculation()
result = calc.multiply(12, 5)
self.assertEqual(result, 60)
def test_divide(self):
calc = Calculation()
result = calc.divide(12, 5)
self.assertEqual(result, 2)
是 test_base_1 的副本，但导入已更改。再次运行 Trial，如上所述，您的测试现在应该通过。
$ python -m twisted.trial calculus.test.test_base_2
Running 4 tests.
calculus.test.test_base
CalculationTestCase
test_add ... [OK]
test_divide ... [OK]
test_multiply ... [OK]
test_subtract ... [OK]
-------------------------------------------------------------------------------
Ran 4 tests in 0.067s
PASSED (successes=4)
分解公共测试逻辑¶
您会注意到，我们的测试文件包含冗余代码。让我们摆脱它。Python 的单元测试框架允许您的测试类定义一个名为 setUp 的方法，该方法在类中的每个测试方法之前调用。这允许您向 self 添加属性，这些属性可以在测试方法中使用。我们还将添加一个参数化测试方法来进一步简化代码。
请注意，测试类还可以提供 setUp 的对应项，名为 tearDown ，它将在每个测试（无论成功与否）之后调用。 tearDown 主要用于测试后清理目的。我们将在稍后使用 tearDown 。
创建 calculus/test/test_base_2b.py 如下
from calculus.base_2 import Calculation
from twisted.trial import unittest
class CalculationTestCase(unittest.TestCase):
def setUp(self):
self.calc = Calculation()
def _test(self, operation, a, b, expected):
result = operation(a, b)
self.assertEqual(result, expected)
def test_add(self):
self._test(self.calc.add, 3, 8, 11)
def test_subtract(self):
self._test(self.calc.subtract, 7, 3, 4)
def test_multiply(self):
self._test(self.calc.multiply, 6, 9, 54)
def test_divide(self):
self._test(self.calc.divide, 12, 5, 2)
干净多了，不是吗？
现在我们将添加一些额外的错误测试。仅测试 API 的成功使用通常是不够的，尤其是在您期望其他人使用您的代码的情况下。让我们确保 Calculation 类在有人尝试使用无法转换为整数的参数调用其方法时引发异常。
我们到达 calculus/test/test_base_3.py
from calculus.base_3 import Calculation
from twisted.trial import unittest
class CalculationTestCase(unittest.TestCase):
def setUp(self):
self.calc = Calculation()
def _test(self, operation, a, b, expected):
result = operation(a, b)
self.assertEqual(result, expected)
def _test_error(self, operation):
self.assertRaises(TypeError, operation, "foo", 2)
self.assertRaises(TypeError, operation, "bar", "egg")
self.assertRaises(TypeError, operation, [3], [8, 2])
self.assertRaises(TypeError, operation, {"e": 3}, {"r": "t"})
def test_add(self):
self._test(self.calc.add, 3, 8, 11)
def test_subtract(self):
self._test(self.calc.subtract, 7, 3, 4)
def test_multiply(self):
self._test(self.calc.multiply, 6, 9, 54)
def test_divide(self):
self._test(self.calc.divide, 12, 5, 2)
def test_errorAdd(self):
self._test_error(self.calc.add)
def test_errorSubtract(self):
self._test_error(self.calc.subtract)
def test_errorMultiply(self):
self._test_error(self.calc.multiply)
def test_errorDivide(self):
self._test_error(self.calc.divide)
我们添加了四个新测试和一个通用函数 _test_error 。此函数使用 assertRaises 方法，该方法接受一个异常类、一个要运行的函数及其参数，并检查在参数上调用函数是否确实引发了给定的异常。
如果您运行上面的代码，您会发现并非所有测试都失败。在 Python 中，添加和乘以不同类型甚至不同类型的对象通常是有效的，因此 add 和 multiply 测试中的代码不会引发异常，因此这些测试失败。因此，让我们在 API 类中添加显式类型转换。这将我们带到 calculus/base_3.py
# -*- test-case-name: calculus.test.test_base_3 -*-
class Calculation:
def _make_ints(self, *args):
try:
return [int(arg) for arg in args]
except ValueError:
raise TypeError("Couldn't coerce arguments to integers: {}".format(*args))
def add(self, a, b):
a, b = self._make_ints(a, b)
return a + b
def subtract(self, a, b):
a, b = self._make_ints(a, b)
return a - b
def multiply(self, a, b):
a, b = self._make_ints(a, b)
return a * b
def divide(self, a, b):
a, b = self._make_ints(a, b)
return a // b
这里，_make_ints 辅助函数尝试将列表转换为等效整数列表，并在转换出错时引发 TypeError 。
注意
如果传递了错误类型的对象（例如列表），int 转换也会引发 TypeError 。我们只是让该异常通过，因为 TypeError 已经是我们希望在出现问题时出现的异常。
Twisted 特定的测试¶
到目前为止，我们一直在进行相当标准的 Python 单元测试。只需进行一些美观上的更改（最重要的是，直接导入 unittest 而不是使用 Twisted 的 unittest 版本），我们可以使用 Python 的标准库单元测试框架运行上面的测试。
在这里，我们将假设您对 Twisted 的网络 I/O、计时和 Deferred API 有基本了解。如果您还没有阅读它们，您应该阅读有关 编写服务器 、 编写客户端 和 Deferreds 的文档。
现在我们将进入本教程的真正目的，并利用 Trial 来测试 Twisted 代码。
测试协议¶
现在我们将创建一个自定义协议，以从类似 telnet 的会话中调用我们的类。我们将远程调用带有参数的命令并读回响应。目标是在不创建套接字的情况下测试我们的网络代码。
创建和测试服务器¶
首先，我们将编写测试，然后解释它们的作用。远程测试代码的第一个版本是
from calculus.remote_1 import RemoteCalculationFactory
from twisted.test import proto_helpers
from twisted.trial import unittest
class RemoteCalculationTestCase(unittest.TestCase):
def setUp(self):
factory = RemoteCalculationFactory()
self.proto = factory.buildProtocol(("127.0.0.1", 0))
self.tr = proto_helpers.StringTransport()
self.proto.makeConnection(self.tr)
def _test(self, operation, a, b, expected):
self.proto.dataReceived(f"{operation} {a} {b}\r\n".encode())
self.assertEqual(int(self.tr.value()), expected)
def test_add(self):
return self._test("add", 7, 6, 13)
def test_subtract(self):
return self._test("subtract", 82, 78, 4)
def test_multiply(self):
return self._test("multiply", 2, 8, 16)
def test_divide(self):
return self._test("divide", 14, 3, 4)
为了完全理解此客户端，熟悉 Twisted 中使用的 Factory/Protocol/Transport 模式非常有帮助。
我们首先创建一个协议工厂对象。请注意，我们还没有看到 RemoteCalculationFactory 类。它位于下面的 calculus/remote_1.py 中。我们调用 buildProtocol 以要求工厂为我们构建一个知道如何与我们的服务器通信的协议对象。然后，我们创建一个伪造的网络传输，一个 twisted.test.proto_helpers.StringTransport 类实例（请注意，测试包通常不是 Twisted 公共 API 的一部分；``twisted.test.proto_helpers`` 是一个例外）。这个伪造的传输是通信的关键。它用于模拟没有网络的网络连接。传递给 buildProtocol 的地址和端口通常由工厂用来选择立即拒绝远程连接；由于我们使用的是伪造的传输，我们可以选择工厂可以接受的任何值。在这种情况下，工厂只是忽略地址，因此我们不需要选择任何特定内容。
在测试 Twisted 代码时，不使用真实网络连接来测试协议既简单又推荐。尽管 Twisted 中有许多使用网络的测试，但大多数好的测试都没有。单元测试和网络的问题在于网络不可靠。我们无法知道它们是否会一直表现出合理的行为。这会导致由于网络变化而导致的间歇性测试失败。现在，我们正在尝试测试我们的 Twisted 代码，而不是网络可靠性。通过设置和使用伪造的传输，我们可以编写 100% 可靠的测试。我们还可以以确定性方式测试网络故障，这是完整测试套件的另一个重要部分。
理解此客户端代码的最后一个关键是 _test 方法。对 dataReceived 的调用模拟了数据到达网络传输。但它从哪里到达？它被传递给协议实例的 lineReceived 方法（位于下面的 calculus/remote_1.py 中）。因此，客户端本质上是在欺骗服务器，让它认为它已经通过网络收到了操作和参数。服务器（再次参见下文）将工作交给它的 CalculationProxy 对象，该对象又将工作交给它的 Calculation 实例。结果通过 sendLine 写回（到伪造的字符串传输对象中），然后立即可供客户端使用，客户端使用 tr.value() 获取它并检查它是否具有预期的值。因此，在上面的两行 _test 方法中，幕后发生了很多事情。
最后 ，让我们看看此协议的实现。将以下内容放入 calculus/remote_1.py
# -*- test-case-name: calculus.test.test_remote_1 -*-
from calculus.base_3 import Calculation
from twisted.internet import protocol
from twisted.protocols import basic
class CalculationProxy:
def __init__(self):
self.calc = Calculation()
for m in ["add", "subtract", "multiply", "divide"]:
setattr(self, f"remote_{m}", getattr(self.calc, m))
class RemoteCalculationProtocol(basic.LineReceiver):
def __init__(self):
self.proxy = CalculationProxy()
def lineReceived(self, line):
op, a, b = line.decode("utf-8").split()
a = int(a)
b = int(b)
op = getattr(self.proxy, f"remote_{op}")
result = op(a, b)
self.sendLine(str(result).encode("utf-8"))
class RemoteCalculationFactory(protocol.Factory):
protocol = RemoteCalculationProtocol
def main():
import sys
from twisted.internet import reactor
from twisted.python import log
log.startLogging(sys.stdout)
reactor.listenTCP(0, RemoteCalculationFactory())
reactor.run()
if __name__ == "__main__":
main()
如前所述，此服务器创建了一个从 basic.LineReceiver 继承的协议，然后创建一个使用它作为协议的工厂。唯一的技巧是 CalculationProxy 对象，它通过 remote_* 方法调用 Calculation 方法。这种模式在 Twisted 中经常使用，因为它非常明确地说明了您要使哪些方法可访问。
如果您运行此测试（python -m twisted.trial calculus.test.test_remote_1 ），一切应该都很好。您还可以运行一个服务器来使用 telnet 客户端对其进行测试。为此，请调用 python calculus/remote_1.py 。您应该有以下输出
2008-04-25 10:53:27+0200 [-] Log opened.
2008-04-25 10:53:27+0200 [-] __main__.RemoteCalculationFactory starting on 46194
2008-04-25 10:53:27+0200 [-] Starting factory <__main__.RemoteCalculationFactory instance at 0x846a0cc>
46194 被随机端口替换。然后，您可以在其上调用 telnet
$ telnet localhost 46194
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
add 4123 9423
13546
它有效！
创建和测试客户端¶
当然，我们现在构建的东西并没有特别有用：我们现在将为我们的服务器构建一个客户端，以便能够在 Python 程序中使用它。它将服务于我们的下一个目的。
创建 calculus/test/test_client_1.py
from calculus.client_1 import RemoteCalculationClient
from twisted.test import proto_helpers
from twisted.trial import unittest
class ClientCalculationTestCase(unittest.TestCase):
def setUp(self):
self.tr = proto_helpers.StringTransport()
self.proto = RemoteCalculationClient()
self.proto.makeConnection(self.tr)
def _test(self, operation, a, b, expected):
d = getattr(self.proto, operation)(a, b)
self.assertEqual(self.tr.value(), f"{operation} {a} {b}\r\n".encode())
self.tr.clear()
d.addCallback(self.assertEqual, expected)
self.proto.dataReceived(
"{}\r\n".format(
expected,
).encode("utf-8")
)
return d
def test_add(self):
return self._test("add", 7, 6, 13)
def test_subtract(self):
return self._test("subtract", 82, 78, 4)
def test_multiply(self):
return self._test("multiply", 2, 8, 16)
def test_divide(self):
return self._test("divide", 14, 3, 4)
它与服务器测试用例非常对称。唯一棘手的部分是我们没有使用客户端工厂。我们很懒，在客户端部分它也不是很有用，所以我们直接实例化协议。
顺便说一下，我们在这里引入了一个非常重要的概念：测试现在返回一个 Deferred 对象，断言是在回调中完成的。当测试返回一个 Deferred 时，反应器将运行，直到 Deferred 触发并运行其回调。这里要做的重要事情是 **不要忘记返回 Deferred** 。如果你这样做，即使没有断言，你的测试也会通过。这也是为什么先让测试失败很重要的原因：如果你的测试通过了，而你知道它们不应该通过，那么你的测试中就存在问题。
我们现在将添加远程客户端类以生成 calculus/client_1.py
# -*- test-case-name: calculus.test.test_client_1 -*-
from twisted.internet import defer
from twisted.protocols import basic
class RemoteCalculationClient(basic.LineReceiver):
def __init__(self):
self.results = []
def lineReceived(self, line):
d = self.results.pop(0)
d.callback(int(line))
def _sendOperation(self, op, a, b):
d = defer.Deferred()
self.results.append(d)
line = f"{op} {a} {b}".encode()
self.sendLine(line)
return d
def add(self, a, b):
return self._sendOperation("add", a, b)
def subtract(self, a, b):
return self._sendOperation("subtract", a, b)
def multiply(self, a, b):
return self._sendOperation("multiply", a, b)
def divide(self, a, b):
return self._sendOperation("divide", a, b)
更多良好实践¶
测试调度¶
在测试涉及时间流逝的代码时，例如等待测试中两小时的超时发生，并不现实。Twisted 为此提供了一个解决方案，即 Clock 类，它允许模拟时间流逝。
例如，我们将测试客户端请求超时的代码：由于我们的客户端使用 TCP，它可能会长时间挂起（防火墙、连接问题等）。因此，通常我们需要在客户端侧实现超时。基本上，我们只是发送一个请求，没有收到响应，并期望在一定时间后触发超时错误。
from calculus.client_2 import ClientTimeoutError, RemoteCalculationClient
from twisted.internet import task
from twisted.test import proto_helpers
from twisted.trial import unittest
class ClientCalculationTestCase(unittest.TestCase):
def setUp(self):
self.tr = proto_helpers.StringTransportWithDisconnection()
self.clock = task.Clock()
self.proto = RemoteCalculationClient()
self.tr.protocol = self.proto
self.proto.callLater = self.clock.callLater
self.proto.makeConnection(self.tr)
def _test(self, operation, a, b, expected):
d = getattr(self.proto, operation)(a, b)
self.assertEqual(self.tr.value(), f"{operation} {a} {b}\r\n".encode())
self.tr.clear()
d.addCallback(self.assertEqual, expected)
self.proto.dataReceived(f"{expected}\r\n".encode())
return d
def test_add(self):
return self._test("add", 7, 6, 13)
def test_subtract(self):
return self._test("subtract", 82, 78, 4)
def test_multiply(self):
return self._test("multiply", 2, 8, 16)
def test_divide(self):
return self._test("divide", 14, 3, 4)
def test_timeout(self):
d = self.proto.add(9, 4)
self.assertEqual(self.tr.value(), b"add 9 4\r\n")
self.clock.advance(self.proto.timeOut)
return self.assertFailure(d, ClientTimeoutError)
这里发生了什么？我们像往常一样实例化我们的协议，唯一的技巧是创建时钟，并将 proto.callLater 赋值给 clock.callLater 。因此，协议中的每个 callLater 调用都将在 clock.advance() 返回之前完成。
在新测试（test_timeout）中，我们调用 clock.advance ，它模拟时间前进（逻辑上类似于 time.sleep 调用）。我们只需要验证我们的 Deferred 是否收到了超时错误。
让我们在代码中实现它。
# -*- test-case-name: calculus.test.test_client_2 -*-
from twisted.internet import defer, reactor
from twisted.protocols import basic
class ClientTimeoutError(Exception):
pass
class RemoteCalculationClient(basic.LineReceiver):
callLater = reactor.callLater
timeOut = 60
def __init__(self):
self.results = []
def lineReceived(self, line):
d, callID = self.results.pop(0)
callID.cancel()
d.callback(int(line))
def _cancel(self, d):
d.errback(ClientTimeoutError())
def _sendOperation(self, op, a, b):
d = defer.Deferred()
callID = self.callLater(self.timeOut, self._cancel, d)
self.results.append((d, callID))
line = f"{op} {a} {b}".encode()
self.sendLine(line)
return d
def add(self, a, b):
return self._sendOperation("add", a, b)
def subtract(self, a, b):
return self._sendOperation("subtract", a, b)
def multiply(self, a, b):
return self._sendOperation("multiply", a, b)
def divide(self, a, b):
return self._sendOperation("divide", a, b)
如果一切顺利完成，请务必记住取消 callLater 返回的 DelayedCall 。
测试后清理¶
本章主要面向希望在测试中创建套接字或进程的人员。如果还不明显，你必须尝试避免使用它们，因为最终会导致很多问题，其中之一是间歇性故障。间歇性故障是自动化测试的祸根。
为了真正测试这一点，我们将使用我们的协议启动一个服务器。
from calculus.client_2 import RemoteCalculationClient
from calculus.remote_1 import RemoteCalculationFactory
from twisted.internet import protocol, reactor
from twisted.trial import unittest
class RemoteRunCalculationTestCase(unittest.TestCase):
def setUp(self):
factory = RemoteCalculationFactory()
self.port = reactor.listenTCP(0, factory, interface="127.0.0.1")
self.client = None
def tearDown(self):
if self.client is not None:
self.client.transport.loseConnection()
return self.port.stopListening()
def _test(self, op, a, b, expected):
creator = protocol.ClientCreator(reactor, RemoteCalculationClient)
def cb(client):
self.client = client
return getattr(self.client, op)(a, b).addCallback(
self.assertEqual, expected
)
return creator.connectTCP("127.0.0.1", self.port.getHost().port).addCallback(cb)
def test_add(self):
return self._test("add", 5, 9, 14)
def test_subtract(self):
return self._test("subtract", 47, 13, 34)
def test_multiply(self):
return self._test("multiply", 7, 3, 21)
def test_divide(self):
return self._test("divide", 84, 10, 8)
如果你删除了 stopListening 调用，最新版本的 Trial 会大声失败，这很好。
此外，你应该知道 tearDown 将在任何情况下都会被调用，无论成功还是失败。因此，不要期望在测试方法中创建的每个对象都存在，因为你的测试可能在中途失败了。
Trial 还具有 addCleanup 方法，它使这种清理变得容易，并消除了对 tearDown 的需求。例如，你可以通过以下方式删除 _test 中的代码
def setUp(self):
factory = RemoteCalculationFactory()
self.port = reactor.listenTCP(0, factory, interface="127.0.0.1")
self.addCleanup(self.port.stopListening)
def _test(self, op, a, b, expected):
creator = protocol.ClientCreator(reactor, RemoteCalculationClient)
def cb(client):
self.addCleanup(self.client.transport.loseConnection)
return getattr(client, op)(a, b).addCallback(self.assertEqual, expected)
return creator.connectTCP('127.0.0.1', self.port.getHost().port).addCallback(cb)
这消除了对 tearDown 方法的需求，并且你无需检查 self.client 的值：你只在创建客户端时调用 addCleanup。
处理记录的错误¶
目前，如果你向我们的服务器发送无效命令或无效参数，它会记录异常并关闭连接。这是一种完全有效的行为，但为了本教程的目的，我们希望在用户发送无效运算符时向用户返回错误，并在服务器端记录任何错误。因此，我们需要一个像这样的测试
def test_invalidParameters(self):
self.proto.dataReceived('add foo bar\r\n')
self.assertEqual(self.tr.value(), "error\r\n")
# -*- test-case-name: calculus.test.test_remote_1 -*-
from calculus.base_3 import Calculation
from twisted.internet import protocol
from twisted.protocols import basic
from twisted.python import log
class CalculationProxy:
def __init__(self):
self.calc = Calculation()
for m in ["add", "subtract", "multiply", "divide"]:
setattr(self, f"remote_{m}", getattr(self.calc, m))
class RemoteCalculationProtocol(basic.LineReceiver):
def __init__(self):
self.proxy = CalculationProxy()
def lineReceived(self, line):
op, a, b = line.decode("utf-8").split()
op = getattr(
self.proxy,
"remote_{}".format(
op,
),
)
try:
result = op(a, b)
except TypeError:
log.err()
self.sendLine(b"error")
else:
self.sendLine(str(result).encode("utf-8"))
class RemoteCalculationFactory(protocol.Factory):
protocol = RemoteCalculationProtocol
def main():
import sys
from twisted.internet import reactor
from twisted.python import log
log.startLogging(sys.stdout)
reactor.listenTCP(0, RemoteCalculationFactory())
reactor.run()
if __name__ == "__main__":
main()
如果你尝试类似的操作，它将无法正常工作。以下是您应该得到的输出
$ python -m twisted.trial calculus.test.test_remote_3.RemoteCalculationTestCase.test_invalidParameters
calculus.test.test_remote_3
RemoteCalculationTestCase
test_invalidParameters ... [ERROR]
===============================================================================
[ERROR]: calculus.test.test_remote_3.RemoteCalculationTestCase.test_invalidParameters
Traceback (most recent call last):
File "/tmp/calculus/remote_2.py", line 27, in lineReceived
result = op(a, b)
File "/tmp/calculus/base_3.py", line 11, in add
a, b = self._make_ints(a, b)
File "/tmp/calculus/base_3.py", line 8, in _make_ints
raise TypeError
exceptions.TypeError:
-------------------------------------------------------------------------------
Ran 1 tests in 0.004s
FAILED (errors=1)
起初，你可能会认为存在问题，因为你捕获了此异常。但实际上，Trial 不会让你在没有控制的情况下这样做：你必须预期记录的错误并清理它们。为此，你必须使用 flushLoggedErrors 方法。你用你期望的异常调用它，它将返回自测试开始以来记录的异常列表。通常，你希望检查此列表是否具有预期的长度，或者可能每个异常是否具有预期的消息。我们在测试中执行了前者
from calculus.remote_2 import RemoteCalculationFactory
from twisted.test import proto_helpers
from twisted.trial import unittest
class RemoteCalculationTestCase(unittest.TestCase):
def setUp(self):
factory = RemoteCalculationFactory()
self.proto = factory.buildProtocol(("127.0.0.1", 0))
self.tr = proto_helpers.StringTransport()
self.proto.makeConnection(self.tr)
def _test(self, operation, a, b, expected):
self.proto.dataReceived(f"{operation} {a} {b}\r\n".encode())
self.assertEqual(int(self.tr.value()), expected)
def test_add(self):
return self._test("add", 7, 6, 13)
def test_subtract(self):
return self._test("subtract", 82, 78, 4)
def test_multiply(self):
return self._test("multiply", 2, 8, 16)
def test_divide(self):
return self._test("divide", 14, 3, 4)
def test_invalidParameters(self):
self.proto.dataReceived(b"add foo bar\r\n")
self.assertEqual(self.tr.value(), b"error\r\n")
errors = self.flushLoggedErrors(TypeError)
self.assertEqual(len(errors), 1)
解决错误¶
在超时开发过程中遗留了一个错误（可能存在多个错误，但这并不重要），涉及在超时时重用协议：连接不会断开，因此你可能会永远超时。通常，用户会来找你说“我的网络很糟糕，我遇到了一个奇怪的问题。似乎你可以通过在 YYY 处执行 XXX 来解决它。”
实际上，这个错误可以通过多种方式纠正。但是，如果你在没有添加测试的情况下纠正它，有一天你会遇到一个大问题：回归。因此，第一步是添加一个失败的测试。
from calculus.client_3 import ClientTimeoutError, RemoteCalculationClient
from twisted.internet import task
from twisted.test import proto_helpers
from twisted.trial import unittest
class ClientCalculationTestCase(unittest.TestCase):
def setUp(self):
self.tr = proto_helpers.StringTransportWithDisconnection()
self.clock = task.Clock()
self.proto = RemoteCalculationClient()
self.tr.protocol = self.proto
self.proto.callLater = self.clock.callLater
self.proto.makeConnection(self.tr)
def _test(self, operation, a, b, expected):
d = getattr(self.proto, operation)(a, b)
self.assertEqual(self.tr.value(), f"{operation} {a} {b}\r\n".encode())
self.tr.clear()
d.addCallback(self.assertEqual, expected)
self.proto.dataReceived(f"{expected}\r\n".encode())
return d
def test_add(self):
return self._test("add", 7, 6, 13)
def test_subtract(self):
return self._test("subtract", 82, 78, 4)
def test_multiply(self):
return self._test("multiply", 2, 8, 16)
def test_divide(self):
return self._test("divide", 14, 3, 4)
def test_timeout(self):
d = self.proto.add(9, 4)
self.assertEqual(self.tr.value(), b"add 9 4\r\n")
self.clock.advance(self.proto.timeOut)
return self.assertFailure(d, ClientTimeoutError)
def test_timeoutConnectionLost(self):
called = []
def lost(arg):
called.append(True)
self.proto.connectionLost = lost
d = self.proto.add(9, 4)
self.assertEqual(self.tr.value(), b"add 9 4\r\n")
self.clock.advance(self.proto.timeOut)
def check(ignore):
self.assertEqual(called, [True])
return self.assertFailure(d, ClientTimeoutError).addCallback(check)
我们在这里做了什么？
我们切换到 StringTransportWithDisconnection。此传输管理
loseConnection并将其转发到其协议。我们通过
protocol属性将协议分配给传输。我们检查超时后我们的连接是否已关闭。
为此，我们随后使用 TimeoutMixin 类，它几乎完成了我们想要做的一切。最棒的是，它几乎没有改变我们的类。
# -*- test-case-name: calculus.test.test_client -*-
from twisted.internet import defer
from twisted.protocols import basic, policies
class ClientTimeoutError(Exception):
pass
class RemoteCalculationClient(basic.LineReceiver, policies.TimeoutMixin):
def __init__(self):
self.results = []
self._timeOut = 60
def lineReceived(self, line):
self.setTimeout(None)
d = self.results.pop(0)
d.callback(int(line))
def timeoutConnection(self):
for d in self.results:
d.errback(ClientTimeoutError())
self.transport.loseConnection()
def _sendOperation(self, op, a, b):
d = defer.Deferred()
self.results.append(d)
line = f"{op} {a} {b}".encode()
self.sendLine(line)
self.setTimeout(self._timeOut)
return d
def add(self, a, b):
return self._sendOperation("add", a, b)
def subtract(self, a, b):
return self._sendOperation("subtract", a, b)
def multiply(self, a, b):
return self._sendOperation("multiply", a, b)
def divide(self, a, b):
return self._sendOperation("divide", a, b)
在没有反应器的情况下测试 Deferreds¶
上面我们了解了如何从测试方法中返回 Deferreds，以便对它们的结果或仅在它们触发后发生的副作用进行断言。这可能很有用，但我们在这个示例中实际上不需要此功能。因为我们小心地使用了 Clock ，所以我们不需要在测试中运行全局反应器。与其返回一个附加了回调的 Deferred，该回调执行必要的断言，不如使用测试助手 successResultOf （以及相应的错误情况助手 failureResultOf ），以提取其结果并直接对其进行断言。与返回 Deferred 相比，这避免了忘记返回 Deferred 的问题，改进了断言失败时报告的堆栈跟踪，并避免了使用全局反应器的复杂性（例如，这可能需要清理）。
from calculus.client_3 import ClientTimeoutError, RemoteCalculationClient
from twisted.internet import task
from twisted.test import proto_helpers
from twisted.trial import unittest
class ClientCalculationTestCase(unittest.TestCase):
def setUp(self):
self.tr = proto_helpers.StringTransportWithDisconnection()
self.clock = task.Clock()
self.proto = RemoteCalculationClient()
self.tr.protocol = self.proto
self.proto.callLater = self.clock.callLater
self.proto.makeConnection(self.tr)
def _test(self, operation, a, b, expected):
d = getattr(self.proto, operation)(a, b)
self.assertEqual(self.tr.value(), f"{operation} {a} {b}\r\n".encode())
self.tr.clear()
self.proto.dataReceived(f"{expected}\r\n".encode())
self.assertEqual(expected, self.successResultOf(d))
def test_add(self):
self._test("add", 7, 6, 13)
def test_subtract(self):
self._test("subtract", 82, 78, 4)
def test_multiply(self):
self._test("multiply", 2, 8, 16)
def test_divide(self):
self._test("divide", 14, 3, 4)
def test_timeout(self):
d = self.proto.add(9, 4)
self.assertEqual(self.tr.value(), b"add 9 4\r\n")
self.clock.advance(self.proto.timeOut)
self.failureResultOf(d).trap(ClientTimeoutError)
def test_timeoutConnectionLost(self):
called = []
def lost(arg):
called.append(True)
self.proto.connectionLost = lost
d = self.proto.add(9, 4)
self.assertEqual(self.tr.value(), b"add 9 4\r\n")
self.clock.advance(self.proto.timeOut)
def check(ignore):
self.assertEqual(called, [True])
self.failureResultOf(d).trap(ClientTimeoutError)
self.assertEqual(called, [True])
此版本的代码执行相同的断言，但不再从任何测试方法中返回任何 Deferreds。与其在回调中对 Deferred 的结果进行断言，不如在它知道 Deferred 应该有一个结果时（在 _test 方法以及 test_timeout 和 test_timeoutConnectionLost 中）进行断言。知道 Deferred 应该何时具有测试的可能性是 successResultOf 在单元测试中变得有用的原因，但阻止它适用于非测试目的。
successResultOf 如果传递给它的 Deferred 没有结果，或者有失败结果，则会引发异常（测试失败）。类似地，failureResultOf 如果传递给它的 Deferred 没有结果，或者有成功结果，则会引发异常（同样导致测试失败）。还有一个用于测试最终情况的辅助方法，assertNoResult ，它只在传递给它的 Deferred 有 结果（成功或失败）时引发异常（测试失败）。
进入调试器¶
在编写和运行测试的过程中，使用调试器通常很有帮助。这在跟踪代码中麻烦错误的来源时尤其有用。Python 的标准库包含一个调试器，以 pdb 模块的形式提供。使用 pdb 运行测试很简单，只需使用 --debug 选项调用 Twisted，这将在您的测试套件执行开始时启动 pdb。
Trial 还提供了一个 --debugger 选项，可以使用另一个调试器来运行您的测试套件。要指定除 pdb 之外的调试器，请传入提供与 pdb 相同接口的对象的完全限定名称。大多数第三方调试器倾向于实现类似于 pdb 的接口，或者至少提供一个这样做的包装器对象。例如，使用额外参数 --debug --debugger pudb 调用 Trial 将打开 PuDB 调试器，前提是它已正确安装。
代码覆盖率¶
代码覆盖率是软件测试的一个方面，它显示了您的测试覆盖了程序代码的多少。有不同类型的度量：路径覆盖率、条件覆盖率、语句覆盖率……这里我们只考虑语句覆盖率，即一行代码是否被执行。
Trial 具有一个选项来生成测试的语句覆盖率。此选项为 –coverage。它在 _trial_temp 中创建一个覆盖率目录，其中包含每个测试期间使用的模块的 .cover 文件。对我们来说有趣的是 calculus.base.cover 和 calculus.remote.cover。每行以一个计数器开头，显示该行在测试期间执行的次数，或者如果该行未被覆盖，则显示标记 ‘>>>>>>’。如果您已经完成了本教程的所有步骤，那么您应该拥有完整的覆盖率：）。
同样，这只是一个有用的提示，但这并不意味着您的代码是完美的：您的测试应该考虑所有可能的输入和输出，以获得完整的覆盖率（条件、路径等）。
结论¶
那么您从本文档中学到了什么？
如何使用 Trial 命令行工具来运行您的测试
如何使用字符串传输来测试单个客户端和服务器，而无需创建套接字
如果您确实想要创建套接字，如何干净地执行此操作，以避免产生不良副作用
以及一些您不可或缺的小技巧。
如果您对其中某个主题仍然感到困惑，请向我们提供您的反馈！您可以提交工单来改进本文档 - 了解如何在 Twisted 网站上贡献。