Python code coverage for Lib/test/test_asyncio/test_transports.py

#	count	content
1	n/a	"""Tests for transports.py."""
2	n/a
3	n/a	import unittest
4	n/a	from unittest import mock
5	n/a
6	n/a	import asyncio
7	n/a	from asyncio import transports
8	n/a
9	n/a
10	n/a	class TransportTests(unittest.TestCase):
11	n/a
12	n/a	def test_ctor_extra_is_none(self):
13	n/a	transport = asyncio.Transport()
14	n/a	self.assertEqual(transport._extra, {})
15	n/a
16	n/a	def test_get_extra_info(self):
17	n/a	transport = asyncio.Transport({'extra': 'info'})
18	n/a	self.assertEqual('info', transport.get_extra_info('extra'))
19	n/a	self.assertIsNone(transport.get_extra_info('unknown'))
20	n/a
21	n/a	default = object()
22	n/a	self.assertIs(default, transport.get_extra_info('unknown', default))
23	n/a
24	n/a	def test_writelines(self):
25	n/a	transport = asyncio.Transport()
26	n/a	transport.write = mock.Mock()
27	n/a
28	n/a	transport.writelines([b'line1',
29	n/a	bytearray(b'line2'),
30	n/a	memoryview(b'line3')])
31	n/a	self.assertEqual(1, transport.write.call_count)
32	n/a	transport.write.assert_called_with(b'line1line2line3')
33	n/a
34	n/a	def test_not_implemented(self):
35	n/a	transport = asyncio.Transport()
36	n/a
37	n/a	self.assertRaises(NotImplementedError,
38	n/a	transport.set_write_buffer_limits)
39	n/a	self.assertRaises(NotImplementedError, transport.get_write_buffer_size)
40	n/a	self.assertRaises(NotImplementedError, transport.write, 'data')
41	n/a	self.assertRaises(NotImplementedError, transport.write_eof)
42	n/a	self.assertRaises(NotImplementedError, transport.can_write_eof)
43	n/a	self.assertRaises(NotImplementedError, transport.pause_reading)
44	n/a	self.assertRaises(NotImplementedError, transport.resume_reading)
45	n/a	self.assertRaises(NotImplementedError, transport.close)
46	n/a	self.assertRaises(NotImplementedError, transport.abort)
47	n/a
48	n/a	def test_dgram_not_implemented(self):
49	n/a	transport = asyncio.DatagramTransport()
50	n/a
51	n/a	self.assertRaises(NotImplementedError, transport.sendto, 'data')
52	n/a	self.assertRaises(NotImplementedError, transport.abort)
53	n/a
54	n/a	def test_subprocess_transport_not_implemented(self):
55	n/a	transport = asyncio.SubprocessTransport()
56	n/a
57	n/a	self.assertRaises(NotImplementedError, transport.get_pid)
58	n/a	self.assertRaises(NotImplementedError, transport.get_returncode)
59	n/a	self.assertRaises(NotImplementedError, transport.get_pipe_transport, 1)
60	n/a	self.assertRaises(NotImplementedError, transport.send_signal, 1)
61	n/a	self.assertRaises(NotImplementedError, transport.terminate)
62	n/a	self.assertRaises(NotImplementedError, transport.kill)
63	n/a
64	n/a	def test_flowcontrol_mixin_set_write_limits(self):
65	n/a
66	n/a	class MyTransport(transports._FlowControlMixin,
67	n/a	transports.Transport):
68	n/a
69	n/a	def get_write_buffer_size(self):
70	n/a	return 512
71	n/a
72	n/a	loop = mock.Mock()
73	n/a	transport = MyTransport(loop=loop)
74	n/a	transport._protocol = mock.Mock()
75	n/a
76	n/a	self.assertFalse(transport._protocol_paused)
77	n/a
78	n/a	with self.assertRaisesRegex(ValueError, 'high.*must be >= low'):
79	n/a	transport.set_write_buffer_limits(high=0, low=1)
80	n/a
81	n/a	transport.set_write_buffer_limits(high=1024, low=128)
82	n/a	self.assertFalse(transport._protocol_paused)
83	n/a	self.assertEqual(transport.get_write_buffer_limits(), (128, 1024))
84	n/a
85	n/a	transport.set_write_buffer_limits(high=256, low=128)
86	n/a	self.assertTrue(transport._protocol_paused)
87	n/a	self.assertEqual(transport.get_write_buffer_limits(), (128, 256))
88	n/a
89	n/a
90	n/a	if __name__ == '__main__':
91	n/a	unittest.main()