2019-05-22 00:17:17 +02:00
|
|
|
import unittest
|
2019-05-22 23:40:24 +02:00
|
|
|
from unittest import mock
|
|
|
|
import socket
|
|
|
|
import struct
|
2019-05-22 00:17:17 +02:00
|
|
|
from asyncio import DatagramTransport
|
|
|
|
from aioupnp.protocols.multicast import MulticastProtocol
|
|
|
|
|
|
|
|
|
|
|
|
class TestMulticast(unittest.TestCase):
|
2019-05-22 23:40:24 +02:00
|
|
|
def test_multicast(self):
|
|
|
|
_ttl = None
|
|
|
|
mock_socket = mock.MagicMock(spec=socket.socket)
|
|
|
|
def getsockopt(*_):
|
|
|
|
return _ttl
|
|
|
|
|
|
|
|
def setsockopt(a, b, ttl: bytes):
|
|
|
|
nonlocal _ttl
|
|
|
|
_ttl, = struct.unpack('b', ttl)
|
2019-05-22 00:17:17 +02:00
|
|
|
|
2019-05-22 23:40:24 +02:00
|
|
|
mock_socket.getsockopt = getsockopt
|
|
|
|
mock_socket.setsockopt = setsockopt
|
2019-05-22 00:17:17 +02:00
|
|
|
|
|
|
|
protocol = MulticastProtocol('1.2.3.4', '1.2.3.4')
|
2019-05-22 23:40:24 +02:00
|
|
|
transport = DatagramTransport()
|
|
|
|
transport._extra = {'socket': mock_socket}
|
|
|
|
self.assertEqual(None, protocol.set_ttl(1))
|
|
|
|
self.assertEqual(0, protocol.get_ttl())
|
2019-05-22 00:17:17 +02:00
|
|
|
protocol.connection_made(transport)
|
2019-05-22 23:40:24 +02:00
|
|
|
self.assertEqual(None, protocol.set_ttl(1))
|
|
|
|
self.assertEqual(1, protocol.get_ttl())
|