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 unittest import mock
from test.test_asyncio import utils as test_utils

def tearDownModule():
    asyncio.set_event_loop_policy(None)

class TestPolicy(asyncio.AbstractEventLoopPolicy):

def __init__(self, loop_factory):
        self.loop_factory = loop_factory
        self.loop = None

def get_event_loop(self):
        # shouldn't ever be called by asyncio.run()
        raise RuntimeError

def new_event_loop(self):
        return self.loop_factory()

def set_event_loop(self, loop):
        if loop is not None:
            # we want to check if the loop is closed
            # in BaseTest.tearDown
            self.loop = loop

class BaseTest(unittest.TestCase):

def new_loop(self):
        loop = asyncio.BaseEventLoop()
        loop._process_events = mock.Mock()
        loop._selector = mock.Mock()
        loop._selector.select.return_value = ()
        loop.shutdown_ag_run = False

async def shutdown_asyncgens():
            loop.shutdown_ag_run = True
        loop.shutdown_asyncgens = shutdown_asyncgens

return loop

def setUp(self):
        super().setUp()

policy = TestPolicy(self.new_loop)
        asyncio.set_event_loop_policy(policy)

def tearDown(self):
        policy = asyncio.get_event_loop_policy()
        if policy.loop is not None:
            self.assertTrue(policy.loop.is_closed())
            self.assertTrue(policy.loop.shutdown_ag_run)

asyncio.set_event_loop_policy(None)
        super().tearDown()

class RunTests(BaseTest):

def test_asyncio_run_return(self):
        async def main():
            await asyncio.sleep(0)
            return 42

self.assertEqual(asyncio.run(main()), 42)

def test_asyncio_run_raises(self):
        async def main():
            await asyncio.sleep(0)
            raise ValueError('spam')

with self.assertRaisesRegex(ValueError, 'spam'):
            asyncio.run(main())

def test_asyncio_run_only_coro(self):
        for o in {1, lambda: None}:
            with self.subTest(obj=o), \
                    self.assertRaisesRegex(ValueError,
                                           'a coroutine was expected'):
                asyncio.run(o)

def test_asyncio_run_debug(self):
        async def main(expected):
            loop = asyncio.get_event_loop()
            self.assertIs(loop.get_debug(), expected)

asyncio.run(main(False))
        asyncio.run(main(True), debug=True)
        with mock.patch('asyncio.coroutines._is_debug_mode', lambda: True):
            asyncio.run(main(True))
            asyncio.run(main(False), debug=False)

def test_asyncio_run_from_running_loop(self):
        async def main():
            coro = main()
            try:
                asyncio.run(coro)
            finally:
                coro.close()  # Suppress ResourceWarning

with self.assertRaisesRegex(RuntimeError,
                                    'cannot be called from a running'):
            asyncio.run(main())

def test_asyncio_run_cancels_hanging_tasks(self):
        lo_task = None

async def leftover():
            await asyncio.sleep(0.1)

async def main():
            nonlocal lo_task
            lo_task = asyncio.create_task(leftover())
            return 123

self.assertEqual(asyncio.run(main()), 123)
        self.assertTrue(lo_task.done())

def test_asyncio_run_reports_hanging_tasks_errors(self):
        lo_task = None
        call_exc_handler_mock = mock.Mock()

async def leftover():
            try:
                await asyncio.sleep(0.1)
            except asyncio.CancelledError:
                1 / 0

async def main():
            loop = asyncio.get_running_loop()
            loop.call_exception_handler = call_exc_handler_mock

nonlocal lo_task
            lo_task = asyncio.create_task(leftover())
            return 123

self.assertEqual(asyncio.run(main()), 123)
        self.assertTrue(lo_task.done())

call_exc_handler_mock.assert_called_with({
            'message': test_utils.MockPattern(r'asyncio.run.*shutdown'),
            'task': lo_task,
            'exception': test_utils.MockInstanceOf(ZeroDivisionError)
        })

def test_asyncio_run_closes_gens_after_hanging_tasks_errors(self):
        spinner = None
        lazyboy = None

class FancyExit(Exception):
            pass

async def fidget():
            while True:
                yield 1
                await asyncio.sleep(1)

async def spin():
            nonlocal spinner
            spinner = fidget()
            try:
                async for the_meaning_of_life in spinner:  # NoQA
                    pass
            except asyncio.CancelledError:
                1 / 0

async def main():
            loop = asyncio.get_running_loop()
            loop.call_exception_handler = mock.Mock()

nonlocal lazyboy
            lazyboy = asyncio.create_task(spin())
            raise FancyExit

with self.assertRaises(FancyExit):
            asyncio.run(main())

self.assertTrue(lazyboy.done())

self.assertIsNone(spinner.ag_frame)
        self.assertFalse(spinner.ag_running)

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