def setUp(self):
        self.server = Mock(autospec=pybtracker.TrackerServer)
        self.server.logger = logger
        self.server.connids = {}
        self.server.activity = {}
        self.server.connid_valid_period = 100
        self.server.interval = 200
        self.server.torrents = {
            b'01234567890123456789': {
                b'11111111111111111111': ('1.1.1.1', 1111, 0, 0, 0, False),
                b'22222222222222222222': ('2.2.2.2', 2222, 0, 0, 0, False),
                b'33333333333333333333': ('3.3.3.3', 3333, 0, 0, 0, False),
            },
            b'ABCDEFGHIJKLMNOPQRST': {
                b'11111111111111111111': ('1.1.1.1', 1111, 0, 0, 0, False),
                b'44444444444444444444': ('4.4.4.4', 4444, 0, 0, 0, False),
            }
        }
        self.proto = pybtracker.server.UdpTrackerServerProto(self.server, allowed_torrents={})
        self.proto.transport = Mock(asyncio.DatagramTransport) 

def connection_made(self, transport: asyncio.BaseTransport) -> None:
        # we need to cast here because the signature in the base class dicates BaseTransport
        # and arguments can only be redefined contravariantly
        self.transport = cast(asyncio.DatagramTransport, transport) 

def __init__(self, app: ASGIFramework, loop: asyncio.AbstractEventLoop, config: Config) -> None:
        self.app = app
        self.config = config
        self.loop = loop
        self.protocol: "QuicProtocol"
        self.protocol_queue: asyncio.Queue = asyncio.Queue(10)
        self.transport: Optional[asyncio.DatagramTransport] = None

        self.loop.create_task(self._consume_events()) 

def connection_made(self, transport: asyncio.DatagramTransport) -> None:  # type: ignore
        # h3/Quic is an optional part of Hypercorn
        from ..protocol.quic import QuicProtocol  # noqa: F811

        self.transport = transport
        socket = self.transport.get_extra_info("socket")
        server = parse_socket_addr(socket.family, socket.getsockname())
        task_group = TaskGroup(self.loop)
        context = Context(task_group)
        self.protocol = QuicProtocol(
            self.app, self.config, cast(Any, context), server, self.protocol_send
        ) 

def test_dgram_not_implemented(self):
        transport = asyncio.DatagramTransport()

        self.assertRaises(NotImplementedError, transport.sendto, 'data')
        self.assertRaises(NotImplementedError, transport.abort) 

def __init__(self, receiver) -> None:
        self.__closed: asyncio.Future[bool] = asyncio.Future()
        self.id = next(protocol_id)
        self.local_candidate: Optional[Candidate] = None
        self.receiver = receiver
        self.transport: Optional[asyncio.DatagramTransport] = None
        self.transactions: Dict[bytes, stun.Transaction] = {} 

def test_dgram_not_implemented(self):
        transport = asyncio.DatagramTransport()

        self.assertRaises(NotImplementedError, transport.sendto, 'data')
        self.assertRaises(NotImplementedError, transport.abort) 

def __init__(self, loop: asyncio.AbstractEventLoop, peer_manager: 'PeerManager', node_id: bytes, udp_port: int,
                 internal_udp_port: int, peer_port: int, external_ip: str, rpc_timeout: float = constants.RPC_TIMEOUT,
                 split_buckets_under_index: int = constants.SPLIT_BUCKETS_UNDER_INDEX,
                 storage: typing.Optional['SQLiteStorage'] = None):
        self.loop = loop
        self.internal_udp_port = internal_udp_port
        self.protocol = KademliaProtocol(loop, peer_manager, node_id, external_ip, udp_port, peer_port, rpc_timeout,
                                         split_buckets_under_index)
        self.listening_port: asyncio.DatagramTransport = None
        self.joined = asyncio.Event(loop=self.loop)
        self._join_task: asyncio.Task = None
        self._refresh_task: asyncio.Task = None
        self._storage = storage 

def mock_network_loop(loop: asyncio.AbstractEventLoop,
                      dht_network: typing.Optional[typing.Dict[typing.Tuple[str, int], 'KademliaProtocol']] = None):
    dht_network: typing.Dict[typing.Tuple[str, int], 'KademliaProtocol'] = dht_network if dht_network is not None else {}

    async def create_datagram_endpoint(proto_lam: typing.Callable[[], 'KademliaProtocol'],
                                       from_addr: typing.Tuple[str, int]):
        def sendto(data, to_addr):
            rx = dht_network.get(to_addr)
            if rx and rx.external_ip:
                # print(f"{from_addr[0]}:{from_addr[1]} -{len(data)} bytes-> {rx.external_ip}:{rx.udp_port}")
                return rx.datagram_received(data, from_addr)

        protocol = proto_lam()
        transport = asyncio.DatagramTransport(extra={'socket': mock_sock})
        transport.is_closing = lambda: False
        transport.close = lambda: mock_sock.close()
        mock_sock.sendto = sendto
        transport.sendto = mock_sock.sendto
        protocol.connection_made(transport)
        dht_network[from_addr] = protocol
        return transport, protocol

    with mock.patch('socket.socket') as mock_socket:
        mock_sock = mock.Mock(spec=socket.socket)
        mock_sock.setsockopt = lambda *_: None
        mock_sock.bind = lambda *_: None
        mock_sock.setblocking = lambda *_: None
        mock_sock.getsockname = lambda: "0.0.0.0"
        mock_sock.getpeername = lambda: ""
        mock_sock.close = lambda: None
        mock_sock.type = socket.SOCK_DGRAM
        mock_sock.fileno = lambda: 7
        mock_socket.return_value = mock_sock
        loop.create_datagram_endpoint = create_datagram_endpoint
        yield 

def test_dgram_not_implemented(self):
        transport = asyncio.DatagramTransport()

        self.assertRaises(NotImplementedError, transport.sendto, 'data')
        self.assertRaises(NotImplementedError, transport.abort) 

def test_dgram_not_implemented(self):
        transport = asyncio.DatagramTransport()

        self.assertRaises(NotImplementedError, transport.sendto, 'data')
        self.assertRaises(NotImplementedError, transport.abort) 

def _test_send_msg(self, expected_failures, mock_asyncio):
        tid = 1000
        msg = b'x' * 20
        self.proto.sent_msgs[tid] = Mock(spec=asyncio.Event)
        self.proto.transport = Mock(spec=asyncio.DatagramTransport)
        self.proto.connect = Mock()
        wait_time = 0
        failures = 0
        async def wait_for_side_effect(task, timeout):
            nonlocal wait_time, failures
            if failures == expected_failures:
                return
            failures += 1
            wait_time += timeout
            raise asyncio.TimeoutError

        expected_wait_time = 0
        for i in range(min(expected_failures, self.client.max_retransmissions)):
            expected_wait_time += 15 * 2 ** i

        mock_asyncio.TimeoutError = asyncio.TimeoutError
        mock_asyncio.wait_for.side_effect = wait_for_side_effect
        if expected_failures > self.client.max_retransmissions:
            with self.assertRaises(TimeoutError):
                await self.proto.send_msg(msg, tid)
        else:
            await self.proto.send_msg(msg, tid)
        self.proto.transport.sendto(msg)
        self.assertEqual(wait_time, expected_wait_time)
        self.proto.connect.called_with()
        self.assertNotIn(tid, self.proto.sent_msgs)
        self.assertEqual(failures, min(expected_failures,
                                       self.client.max_retransmissions))