我们从Python开源项目中，提取了以下18个代码示例，用于说明如何使用asyncio.DatagramProtocol()。
def test_empty(self): f = mock.Mock() p = asyncio.Protocol() self.assertIsNone(p.connection_made(f)) self.assertIsNone(p.connection_lost(f)) self.assertIsNone(p.data_received(f)) self.assertIsNone(p.eof_received()) dp = asyncio.DatagramProtocol() self.assertIsNone(dp.connection_made(f)) self.assertIsNone(dp.connection_lost(f)) self.assertIsNone(dp.error_received(f)) self.assertIsNone(dp.datagram_received(f, f)) sp = asyncio.SubprocessProtocol() self.assertIsNone(sp.connection_made(f)) self.assertIsNone(sp.connection_lost(f)) self.assertIsNone(sp.pipe_data_received(1, f)) self.assertIsNone(sp.pipe_connection_lost(1, f)) self.assertIsNone(sp.process_exited())
def setUp(self): self.loop = self.new_test_loop() self.protocol = test_utils.make_test_protocol(asyncio.DatagramProtocol) self.sock = mock.Mock(spec_set=socket.socket) self.sock.fileno.return_value = 7
def test_create_datagram_endpoint_connect_err(self): self.loop.sock_connect = mock.Mock() self.loop.sock_connect.side_effect = OSError coro = self.loop.create_datagram_endpoint( asyncio.DatagramProtocol, remote_addr=('127.0.0.1', 0)) self.assertRaises( OSError, self.loop.run_until_complete, coro)
def test_create_datagram_endpoint_socket_err(self, m_socket): m_socket.getaddrinfo = socket.getaddrinfo m_socket.socket.side_effect = OSError coro = self.loop.create_datagram_endpoint( asyncio.DatagramProtocol, family=socket.AF_INET) self.assertRaises( OSError, self.loop.run_until_complete, coro) coro = self.loop.create_datagram_endpoint( asyncio.DatagramProtocol, local_addr=('127.0.0.1', 0)) self.assertRaises( OSError, self.loop.run_until_complete, coro)
def test_create_datagram_endpoint_no_matching_family(self): coro = self.loop.create_datagram_endpoint( asyncio.DatagramProtocol, remote_addr=('127.0.0.1', 0), local_addr=('::1', 0)) self.assertRaises( ValueError, self.loop.run_until_complete, coro)
def test_create_datagram_endpoint_noaddr_nofamily(self): coro = self.loop.create_datagram_endpoint( asyncio.DatagramProtocol) self.assertRaises(ValueError, self.loop.run_until_complete, coro)
