How does Twisted reactor work with trial-based unit tests?

Question

I have written a TCP/UDP intercepting proxy using Twisted and I want to add some unit tests to it. I want to setup an echo protocol, then send some data through my proxy, then check the returned response.

However, it seems like even for a simple test using a socket (let aside my intercepting proxy) to connect to the echoer, the reactor desn't seem to be spawned after setUp - the test hangs forever. If I add a timeout to the socket then a timeout exception is raised. I even tried to connect with ncat to make sure is not the manually created socket to blame - the echoer is listening indeed but I receive no echoed data back to the ncat client.

The test code I use is the following

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

To run it I use the following command

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

The output is the following and stays like this until I kill the process

mymodule.test.test_network
  TestTCP
    test_echo ...

It seems like I'm missing something regarding how the reactor works. I've looked for similar examples but couldn't get it working.

How should I write the test to get the expected behavior?

Answer 1

It turned out I must run the test methods as Deffered, using inlineCallbacks, so they are called when the reactor is running. To test this behavior I've used the following snippet

    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

...which makes the test successfully complete

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

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

PASSED (successes=1)

If I enable the sleep(100) in the calledback method, and connect with ncat in that timespan, the data that I send to the listening port is indeed echoed back