Gecko [ www.foryou-vip.com ]

Linux vps-61133.fhnet.fr 4.9.0-19-amd64 #1 SMP Debian 4.9.320-2 (2022-06-30) x86_64
Apache/2.4.25 (Debian)
Server IP : 93.113.207.21 & Your IP : 216.73.216.112
Domains : Cant Read [ /etc/named.conf ]
User : www-data

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

Rename

"""Tests for asyncio/threads.py"""

import asyncio
import unittest

from contextvars import ContextVar
from unittest import mock

def tearDownModule():
    asyncio.set_event_loop_policy(None)

class ToThreadTests(unittest.IsolatedAsyncioTestCase):
    async def test_to_thread(self):
        result = await asyncio.to_thread(sum, [40, 2])
        self.assertEqual(result, 42)

async def test_to_thread_exception(self):
        def raise_runtime():
            raise RuntimeError("test")

with self.assertRaisesRegex(RuntimeError, "test"):
            await asyncio.to_thread(raise_runtime)

async def test_to_thread_once(self):
        func = mock.Mock()

await asyncio.to_thread(func)
        func.assert_called_once()

async def test_to_thread_concurrent(self):
        func = mock.Mock()

futs = []
        for _ in range(10):
            fut = asyncio.to_thread(func)
            futs.append(fut)
        await asyncio.gather(*futs)

self.assertEqual(func.call_count, 10)

async def test_to_thread_args_kwargs(self):
        # Unlike run_in_executor(), to_thread() should directly accept kwargs.
        func = mock.Mock()

await asyncio.to_thread(func, 'test', something=True)

func.assert_called_once_with('test', something=True)

async def test_to_thread_contextvars(self):
        test_ctx = ContextVar('test_ctx')

def get_ctx():
            return test_ctx.get()

test_ctx.set('parrot')
        result = await asyncio.to_thread(get_ctx)

self.assertEqual(result, 'parrot')

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