Ticket #454: broadcastudptestcasev2.patch

File broadcastudptestcasev2.patch, 2.1 KB (added by hawkowl, 3 years ago)

Unit Test - UDP Broadcast over loopback, v2

  • twisted/internet/test/test_udp.py

    diff --git twisted/internet/test/test_udp.py twisted/internet/test/test_udp.py
    index 4452e2c..f9640b3 100644
    from twisted.internet.protocol import DatagramProtocol 
    2626
    2727from twisted.internet.test.connectionmixins import (LogObserverMixin,
    2828                                                    findFreePort)
    29 from twisted.trial.unittest import SkipTest
     29from twisted.trial.unittest import TestCase, SkipTest
     30
     31
     32
     33class BroadcastDatagramProtocol(DatagramProtocol):
     34
     35        def __init__(self, broadcasthost, broadcastport):
     36            self.d = None
     37            self.sentmessage = None
     38            self.broadcasthost = broadcasthost
     39            self.broadcastport = broadcastport
     40
     41
     42        def setDeferred(self, d):
     43            self.d = d
     44
     45
     46        def startProtocol(self):
     47            self.sentmessage = "Twisted is a flying house"
     48            self.transport.socket.setsockopt(
     49                socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
     50            self.transport.write(
     51                self.sentmessage, addr=(self.broadcasthost, self.broadcastport))
     52
     53
     54        def datagramReceived(self, data, (host, port)):
     55            if self.d is not None:
     56                d, self.d = self.d, None
     57                d.callback((data, self.sentmessage))
    3058
    3159
    3260
    class UDPPortTestsMixin(object): 
    116144    Tests for L{IReactorUDP.listenUDP} and
    117145    L{IReactorSocket.adoptDatagramPort}.
    118146    """
     147    def test_UDPBroadcast(self):
     148
     149        def _cb(data):
     150
     151            self.assertEqual(data[0], data[1])
     152            return self.port.stopListening()
     153
     154        reactor = self.buildReactor()
     155        host, portNumber = findFreePort(type=socket.SOCK_DGRAM)
     156        senderprotocol = BroadcastDatagramProtocol(host, portNumber)
     157        port = self.getListeningPort(
     158            reactor, senderprotocol, port=portNumber, interface=host)
     159
     160        d = Deferred()
     161        d.addCallback(_cb)
     162
     163        senderprotocol.setDeferred(d)
     164        return d
     165
     166
    119167    def test_interface(self):
    120168        """
    121169        L{IReactorUDP.listenUDP} returns an object providing L{IListeningPort}.