Ticket #454: broadcastudptestcasev3.patch

File broadcastudptestcasev3.patch, 1.8 KB (added by hawkowl, 3 years ago)

Unit Test - UDP Broadcast over loopback, v3

  • twisted/internet/test/test_udp.py

    diff --git twisted/internet/test/test_udp.py twisted/internet/test/test_udp.py
    index 4452e2c..6554ddd 100644
    class UDPPortTestsMixin(object): 
    116116    Tests for L{IReactorUDP.listenUDP} and
    117117    L{IReactorSocket.adoptDatagramPort}.
    118118    """
     119    def test_UDPBroadcast(self):
     120        """
     121        Broadcasting to 127.0.0.255 works.
     122        """
     123        class BroadcastDatagramProtocol(DatagramProtocol):
     124
     125            def __init__(self, broadcasthost, broadcastport, d):
     126                self.d = d
     127                self.sentmessage = "Twisted is a flying house"
     128                self.broadcasthost = broadcasthost
     129                self.broadcastport = broadcastport
     130
     131
     132            def startProtocol(self):
     133                self.transport.socket.setsockopt(
     134                    socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
     135                self.transport.write(self.sentmessage, addr=(self.broadcasthost, self.broadcastport))
     136
     137
     138            def datagramReceived(self, data, (host, port)):
     139                self.d.callback((data, self.sentmessage))
     140
     141        reactor = self.buildReactor()
     142
     143        d = Deferred()
     144
     145        portNumber = findFreePort(type=socket.SOCK_DGRAM)[1]
     146        host = "127.0.0.255"
     147        senderprotocol = BroadcastDatagramProtocol(host, portNumber, d)
     148        port = self.getListeningPort(
     149            reactor, senderprotocol, port=portNumber, interface=host)
     150
     151        d.addCallback(lambda data: self.assertEqual(data[0], data[1]))
     152        d.addErrback(err)
     153        d.addCallback(lambda ignored: port.stopListening())
     154        d.addCallback(lambda ignored: reactor.stop())
     155
     156        self.runReactor(reactor)
     157
     158
    119159    def test_interface(self):
    120160        """
    121161        L{IReactorUDP.listenUDP} returns an object providing L{IListeningPort}.