Ticket #454: udpbroadcastunittest.patch

File udpbroadcastunittest.patch, 1.8 KB (added by hawkowl, 4 years ago)

Unit Test - UDP Broadcast over loopback

  • twisted/test/test_udp.py

    diff --git twisted/test/test_udp.py twisted/test/test_udp.py
    index 21d145c..b089e1b 100644
    from twisted.python.compat import intToBytes 
    1414from twisted.internet.defer import Deferred, gatherResults, maybeDeferred
    1515from twisted.internet import protocol, reactor, error, defer, interfaces, udp
    1616from twisted.python import runtime
    17 
     17import socket
     18import time
    1819
    1920class Mixin:
    2021
    class BadClient(protocol.DatagramProtocol): 
    113114        raise BadClientError("Application code is very buggy!")
    114115
    115116
     117class BroadcastDatagramProtocol(protocol.DatagramProtocol):
     118
     119    d = None
     120    sentmessage = None
     121    broadcasthost = "127.0.0.255"
     122    broadcastport = 6666
     123
     124    def setDeferred(self, d):
     125        self.d = d
     126
     127    def startProtocol(self):
     128        self.sentmessage = str(time.time())
     129        self.transport.socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
     130        self.sendDatagram(self.sentmessage, (self.broadcasthost, self.broadcastport))
     131   
     132    def sendDatagram(self, data, address):
     133        self.transport.write(data, addr=address)
     134
     135    def datagramReceived(self, data, (host, port)):
     136        if self.d is not None:
     137            d, self.d = self.d, None
     138            d.callback((data, self.sentmessage))
     139
     140class UDPBroadcastTestCase(unittest.TestCase):
     141
     142    def test_UDPBroadcast(self):
     143
     144        port = None
     145
     146        def cbverify(data):
     147
     148            self.assertEqual(data[0],data[1])
     149            self.port.stopListening()
     150           
     151        senderprotocol = BroadcastDatagramProtocol()
     152        self.port = reactor.listenUDP(senderprotocol.broadcastport, senderprotocol)
     153
     154        d = Deferred()
     155        d.addCallback(cbverify)
     156
     157        senderprotocol.setDeferred(d)
    116158
    117159class UDPTestCase(unittest.TestCase):
    118160