Gecko [ www.ctm-euromeuble.fr ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
__init__.py	283 B	-rw-r--r--	2025-06-04 09:23
__main__.py	58 B	-rw-r--r--	2025-06-04 09:23
echo.py	148 B	-rw-r--r--	2025-06-04 09:23
echo2.py	123 B	-rw-r--r--	2025-06-04 09:23
echo3.py	276 B	-rw-r--r--	2025-06-04 09:23
functional.py	7.3 KB	-rw-r--r--	2025-06-04 09:23
test_base_events.py	79.07 KB	-rw-r--r--	2025-06-04 09:23
test_buffered_proto.py	2.28 KB	-rw-r--r--	2025-06-04 09:23
test_context.py	1.04 KB	-rw-r--r--	2025-06-04 09:23
test_events.py	104.04 KB	-rw-r--r--	2025-06-04 09:23
test_futures.py	29.24 KB	-rw-r--r--	2025-06-04 09:23
test_futures2.py	1.58 KB	-rw-r--r--	2025-06-04 09:23
test_locks.py	29.46 KB	-rw-r--r--	2025-06-04 09:23
test_pep492.py	5.65 KB	-rw-r--r--	2025-06-04 09:23
test_proactor_events.py	37.03 KB	-rw-r--r--	2025-06-04 09:23
test_protocols.py	2.24 KB	-rw-r--r--	2025-06-04 09:23
test_queues.py	18.6 KB	-rw-r--r--	2025-06-04 09:23
test_runners.py	5.18 KB	-rw-r--r--	2025-06-04 09:23
test_selector_events.py	47.96 KB	-rw-r--r--	2025-06-04 09:23
test_sendfile.py	20.95 KB	-rw-r--r--	2025-06-04 09:23
test_server.py	3.8 KB	-rw-r--r--	2025-06-04 09:23
test_sock_lowlevel.py	18.07 KB	-rw-r--r--	2025-06-04 09:23
test_sslproto.py	25.77 KB	-rw-r--r--	2025-06-04 09:23
test_streams.py	36.51 KB	-rw-r--r--	2025-06-04 09:23
test_subprocess.py	26.26 KB	-rw-r--r--	2025-06-04 09:23
test_tasks.py	111.8 KB	-rw-r--r--	2025-06-04 09:23
test_threads.py	1.59 KB	-rw-r--r--	2025-06-04 09:23
test_transports.py	3.73 KB	-rw-r--r--	2025-06-04 09:23
test_unix_events.py	61.11 KB	-rw-r--r--	2025-06-04 09:23
test_waitfor.py	8.47 KB	-rw-r--r--	2025-06-04 09:23
test_windows_events.py	10.69 KB	-rw-r--r--	2025-06-04 09:23
test_windows_utils.py	4.14 KB	-rw-r--r--	2025-06-04 09:23
utils.py	16.94 KB	-rw-r--r--	2025-06-04 09:23

Rename

import asyncio
import unittest

from test.test_asyncio import functional as func_tests

def tearDownModule():
    asyncio.set_event_loop_policy(None)

class ReceiveStuffProto(asyncio.BufferedProtocol):
    def __init__(self, cb, con_lost_fut):
        self.cb = cb
        self.con_lost_fut = con_lost_fut

def get_buffer(self, sizehint):
        self.buffer = bytearray(100)
        return self.buffer

def buffer_updated(self, nbytes):
        self.cb(self.buffer[:nbytes])

def connection_lost(self, exc):
        if exc is None:
            self.con_lost_fut.set_result(None)
        else:
            self.con_lost_fut.set_exception(exc)

class BaseTestBufferedProtocol(func_tests.FunctionalTestCaseMixin):

def new_loop(self):
        raise NotImplementedError

def test_buffered_proto_create_connection(self):

NOISE = b'12345678+' * 1024

async def client(addr):
            data = b''

def on_buf(buf):
                nonlocal data
                data += buf
                if data == NOISE:
                    tr.write(b'1')

conn_lost_fut = self.loop.create_future()

tr, pr = await self.loop.create_connection(
                lambda: ReceiveStuffProto(on_buf, conn_lost_fut), *addr)

await conn_lost_fut

async def on_server_client(reader, writer):
            writer.write(NOISE)
            await reader.readexactly(1)
            writer.close()
            await writer.wait_closed()

srv = self.loop.run_until_complete(
            asyncio.start_server(
                on_server_client, '127.0.0.1', 0))

addr = srv.sockets[0].getsockname()
        self.loop.run_until_complete(
            asyncio.wait_for(client(addr), 5))

srv.close()
        self.loop.run_until_complete(srv.wait_closed())

class BufferedProtocolSelectorTests(BaseTestBufferedProtocol,
                                    unittest.TestCase):

def new_loop(self):
        return asyncio.SelectorEventLoop()

@unittest.skipUnless(hasattr(asyncio, 'ProactorEventLoop'), 'Windows only')
class BufferedProtocolProactorTests(BaseTestBufferedProtocol,
                                    unittest.TestCase):

def new_loop(self):
        return asyncio.ProactorEventLoop()

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