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 time
import threading
import unittest

from test.support import socket_helper
from test.test_asyncio import utils as test_utils
from test.test_asyncio import functional as func_tests

def tearDownModule():
    asyncio.set_event_loop_policy(None)

class BaseStartServer(func_tests.FunctionalTestCaseMixin):

def new_loop(self):
        raise NotImplementedError

def test_start_server_1(self):
        HELLO_MSG = b'1' * 1024 * 5 + b'\n'

def client(sock, addr):
            for i in range(10):
                time.sleep(0.2)
                if srv.is_serving():
                    break
            else:
                raise RuntimeError

sock.settimeout(2)
            sock.connect(addr)
            sock.send(HELLO_MSG)
            sock.recv_all(1)
            sock.close()

async def serve(reader, writer):
            await reader.readline()
            main_task.cancel()
            writer.write(b'1')
            writer.close()
            await writer.wait_closed()

async def main(srv):
            async with srv:
                await srv.serve_forever()

srv = self.loop.run_until_complete(asyncio.start_server(
            serve, socket_helper.HOSTv4, 0, start_serving=False))

self.assertFalse(srv.is_serving())

main_task = self.loop.create_task(main(srv))

addr = srv.sockets[0].getsockname()
        with self.assertRaises(asyncio.CancelledError):
            with self.tcp_client(lambda sock: client(sock, addr)):
                self.loop.run_until_complete(main_task)

self.assertEqual(srv.sockets, ())

self.assertIsNone(srv._sockets)
        self.assertIsNone(srv._waiters)
        self.assertFalse(srv.is_serving())

with self.assertRaisesRegex(RuntimeError, r'is closed'):
            self.loop.run_until_complete(srv.serve_forever())

class SelectorStartServerTests(BaseStartServer, unittest.TestCase):

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

@socket_helper.skip_unless_bind_unix_socket
    def test_start_unix_server_1(self):
        HELLO_MSG = b'1' * 1024 * 5 + b'\n'
        started = threading.Event()

def client(sock, addr):
            sock.settimeout(2)
            started.wait(5)
            sock.connect(addr)
            sock.send(HELLO_MSG)
            sock.recv_all(1)
            sock.close()

async def serve(reader, writer):
            await reader.readline()
            main_task.cancel()
            writer.write(b'1')
            writer.close()
            await writer.wait_closed()

async def main(srv):
            async with srv:
                self.assertFalse(srv.is_serving())
                await srv.start_serving()
                self.assertTrue(srv.is_serving())
                started.set()
                await srv.serve_forever()

with test_utils.unix_socket_path() as addr:
            srv = self.loop.run_until_complete(asyncio.start_unix_server(
                serve, addr, start_serving=False))

main_task = self.loop.create_task(main(srv))

with self.assertRaises(asyncio.CancelledError):
                with self.unix_client(lambda sock: client(sock, addr)):
                    self.loop.run_until_complete(main_task)

self.assertEqual(srv.sockets, ())

self.assertIsNone(srv._sockets)
            self.assertIsNone(srv._waiters)
            self.assertFalse(srv.is_serving())

with self.assertRaisesRegex(RuntimeError, r'is closed'):
                self.loop.run_until_complete(srv.serve_forever())

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

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

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