ยปCore Development>Code coverage>Lib/test/test_asyncio/test_transports.py

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

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