File Manager

003 File Manager

Current Path: /usr/local/lib/python3.8/test/test_asyncio

usr / local / lib / python3.8 / test / test_asyncio /

📁 ..
📄 __init__.py(244 B)
📄 __main__.py(58 B)
📁 __pycache__
📄 echo.py(148 B)
📄 echo2.py(123 B)
📄 echo3.py(276 B)
📄 functional.py(7.47 KB)
📄 test_asyncio_waitfor.py(1.45 KB)
📄 test_base_events.py(78.14 KB)
📄 test_buffered_proto.py(2.28 KB)
📄 test_context.py(1020 B)
📄 test_events.py(99.68 KB)
📄 test_futures.py(26.42 KB)
📄 test_futures2.py(694 B)
📄 test_locks.py(33.67 KB)
📄 test_pep492.py(6.04 KB)
📄 test_proactor_events.py(35.04 KB)
📄 test_protocols.py(2 KB)
📄 test_queues.py(21.16 KB)
📄 test_runners.py(5.06 KB)
📄 test_selector_events.py(47.17 KB)
📄 test_sendfile.py(19.67 KB)
📄 test_server.py(3.93 KB)
📄 test_sock_lowlevel.py(11.84 KB)
📄 test_sslproto.py(25.78 KB)
📄 test_streams.py(36.21 KB)
📄 test_subprocess.py(25.33 KB)
📄 test_tasks.py(107.17 KB)
📄 test_transports.py(3.53 KB)
📄 test_unix_events.py(66.23 KB)
📄 test_windows_events.py(8.92 KB)
📄 test_windows_utils.py(4.07 KB)
📄 utils.py(16.83 KB)

Editing: test_transports.py

"""Tests for transports.py."""

import unittest
from unittest import mock

import asyncio
from asyncio import transports

class TransportTests(unittest.TestCase):

def test_ctor_extra_is_none(self):
        transport = asyncio.Transport()
        self.assertEqual(transport._extra, {})

def test_get_extra_info(self):
        transport = asyncio.Transport({'extra': 'info'})
        self.assertEqual('info', transport.get_extra_info('extra'))
        self.assertIsNone(transport.get_extra_info('unknown'))

default = object()
        self.assertIs(default, transport.get_extra_info('unknown', default))

def test_writelines(self):
        writer = mock.Mock()

class MyTransport(asyncio.Transport):
            def write(self, data):
                writer(data)

transport = MyTransport()

transport.writelines([b'line1',
                              bytearray(b'line2'),
                              memoryview(b'line3')])
        self.assertEqual(1, writer.call_count)
        writer.assert_called_with(b'line1line2line3')

def test_not_implemented(self):
        transport = asyncio.Transport()

self.assertRaises(NotImplementedError,
                          transport.set_write_buffer_limits)
        self.assertRaises(NotImplementedError, transport.get_write_buffer_size)
        self.assertRaises(NotImplementedError, transport.write, 'data')
        self.assertRaises(NotImplementedError, transport.write_eof)
        self.assertRaises(NotImplementedError, transport.can_write_eof)
        self.assertRaises(NotImplementedError, transport.pause_reading)
        self.assertRaises(NotImplementedError, transport.resume_reading)
        self.assertRaises(NotImplementedError, transport.is_reading)
        self.assertRaises(NotImplementedError, transport.close)
        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_subprocess_transport_not_implemented(self):
        transport = asyncio.SubprocessTransport()

self.assertRaises(NotImplementedError, transport.get_pid)
        self.assertRaises(NotImplementedError, transport.get_returncode)
        self.assertRaises(NotImplementedError, transport.get_pipe_transport, 1)
        self.assertRaises(NotImplementedError, transport.send_signal, 1)
        self.assertRaises(NotImplementedError, transport.terminate)
        self.assertRaises(NotImplementedError, transport.kill)

def test_flowcontrol_mixin_set_write_limits(self):

class MyTransport(transports._FlowControlMixin,
                          transports.Transport):

def get_write_buffer_size(self):
                return 512

loop = mock.Mock()
        transport = MyTransport(loop=loop)
        transport._protocol = mock.Mock()

self.assertFalse(transport._protocol_paused)

with self.assertRaisesRegex(ValueError, 'high.*must be >= low'):
            transport.set_write_buffer_limits(high=0, low=1)

transport.set_write_buffer_limits(high=1024, low=128)
        self.assertFalse(transport._protocol_paused)
        self.assertEqual(transport.get_write_buffer_limits(), (128, 1024))

transport.set_write_buffer_limits(high=256, low=128)
        self.assertTrue(transport._protocol_paused)
        self.assertEqual(transport.get_write_buffer_limits(), (128, 256))

if __name__ == '__main__':
    unittest.main()

003 File Manager

Editing: test_transports.py

Upload File

Create Folder