Source

udt4twisted / tests / test_udt_reactor.py

Full commit
import socket as s
from twisted.trial import unittest
from twisted.internet.protocol import DatagramProtocol
from twisted.internet.protocol import Protocol, Factory
from twisted.internet import reactor
from twisted.internet import udp
import udt4
from udt4twisted import udt


class TEcho(Protocol):

    def dataReceived(self, data):
        self.transport.write(data)

class EchoFactory(Factory):

    def buildProtocol(self, addr):
        return TEcho()

class Echo(DatagramProtocol):

    def datagramReceived(self, data, (host, port)):
        log.msg("received %r from %s:%d" % (data, host, port))
        self.transport.write(data, (host, port))

class TestUDTReactor(unittest.TestCase):

    def test_reactor_init(self):
        self.assertIsInstance(reactor._poller, udt4.UDTepoll)
        self.assertTrue(len(reactor._udtsockets) == 0)

    def test_reactor_add_system_socket(self):
        srv = reactor.listenUDP(0, Echo())
        tsrv = reactor.listenTCP(0,  EchoFactory())
        self.assertIn(srv.fileno(), reactor._selectables)
        self.assertIn(tsrv.fileno(), reactor._selectables)
        srv.stopListening()
        tsrv.stopListening()


    def test_reactor_add_udt_socket(self):
        srv = reactor.listenUDT(0, EchoFactory())
        self.assertIn(srv.socket.UDTSOCKET.UDTSOCKET, reactor._selectables)
        srv.stopListening()