Source

udt4twisted / tests / test_top_api.py

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

class TEcho(Protocol):

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

class EchoFactory(Factory):

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

class CEchoFactory(ClientFactory):

    pr = []
    def buildProtocol(self, addr):
        p = TEcho()
        self.pr.append(p)
        return p


class Echo(DatagramProtocol):

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

class TestTopLevelAPI(unittest.TestCase):
    def setUp(self):
        pass
    def test_listening(self):
        udtServer = reactor.listenUDT(9999, EchoFactory())
        udtServer.stopListening()

    def test_udp_handling(self):
        p = reactor.listenUDP(5000, Echo())
        p.stopListening()

    def test_udt_conversation(self):
        #FIXME learn how to test transport protocols
        srv = reactor.listenUDT(6000, EchoFactory())
        echo = CEchoFactory()
        client = reactor.connectUDT("127.0.0.1", 6000, echo)
        srv.stopListening()
        client.disconnect()