Twisted reactor 如何与基于试验的单元测试一起工作？答案

【问题标题】：How does Twisted reactor work with trial-based unit tests?Twisted reactor 如何与基于试验的单元测试一起工作？
【发布时间】：2020-11-22 08:16:20
【问题描述】：

我已经使用 Twisted 编写了一个 TCP/UDP 拦截代理，我想向它添加一些单元测试。我想设置一个回显协议，然后通过我的代理发送一些数据，然后检查返回的响应。

然而，即使对于使用套接字（不考虑我的拦截代理）连接回声器的简单测试，反应器似乎也不会似乎在setUp之后生成 -测试永远挂起。如果我向套接字添加超时，则会引发超时异常。我什至尝试与ncat 连接，以确保不是手动创建的套接字应该受到责备——回显器确实在监听，但我没有收到回显给ncat 客户端的回显数据。

我使用的测试代码如下

import pytest
import socket
from twisted.trial import unittest
from twisted.internet import reactor, protocol


class EchoTCP(protocol.Protocol):
    def dataReceived(self, data):
        self.transport.write(data)


class EchoTCPFactory(protocol.Factory):
    protocol = EchoTCP


class TestTCP(unittest.TestCase):
    """Twisted has its own unittest class
    https://twistedmatrix.com/documents/15.2.0/core/howto/trial.html
    """
    def setUp(self):
        self.iface = "127.0.0.1"
        self.data = b"Hello, World!"
        
        # Setup twised echoer
        self.port = reactor.listenTCP(
            8080,
            EchoTCPFactory(),
            interface=self.iface
        )

    def tearDown(self):
        self.port.stopListening()

    def test_echo(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((self.iface, self.port.getHost().port))
        
        sent = sock.send(self.data)
        data = sock.recv(1024)
        sock.close()

        assert data == self.data

要运行它，我使用以下命令

PYTHONPATH="${PWD}" trial --reactor=default mymodule

输出如下并保持这种状态，直到我终止进程

mymodule.test.test_network
  TestTCP
    test_echo ...

我似乎遗漏了有关反应堆工作原理的一些内容。我已经寻找过类似的示例，但无法正常工作。

我应该如何编写测试以获得预期的行为？

【问题讨论】：

标签： python unit-testing twisted trial

【解决方案1】：

结果我必须以Deffered 运行测试方法，使用inlineCallbacks，所以它们在反应器运行时被调用。为了测试这种行为，我使用了以下 sn-p


    from twisted.internet.defer import inlineCallbacks
    # [...]

    def check_reactor(self):
        # time.sleep(100)
        return reactor.running

    @inlineCallbacks
    def test_reactor(self):
        reactor_running = yield threads.deferToThread(self.check_reactor)
        assert reactor_running == True

...使测试成功完成

mymodule.test.test_network
  TestTCP
    test_reactor ...                                                       [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 0.007s

PASSED (successes=1)

如果我在回调方法中启用 sleep(100)，并在该时间跨度内连接ncat，我发送到监听端口的数据确实会回显

【讨论】：