Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

gh-93357: Port test_pep492 to IsolatedAsyncioTestCase #98877

Closed
wants to merge 2 commits into from
Closed

gh-93357: Port test_pep492 to IsolatedAsyncioTestCase #98877

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8a0d65f
Port test_pep492 to IsolatedAsyncioTestCase
arhadthedev Oct 6, 2022
cc42fe8
Merge branch 'main' into utilsTestCase-pep492
arhadthedev Mar 22, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 17 additions & 20 deletions Lib/test/test_asyncio/test_pep492.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
from unittest import mock

import asyncio
from test.test_asyncio import utils as test_utils


def tearDownModule():
Expand All @@ -29,20 +28,18 @@ def __await__(self):
yield


class BaseTest(test_utils.TestCase):
class BaseTest(unittest.IsolatedAsyncioTestCase):

def setUp(self):
super().setUp()
self.loop = asyncio.BaseEventLoop()
self.loop._process_events = mock.Mock()
self.loop._selector = mock.Mock()
self.loop._selector.select.return_value = ()
self.set_event_loop(self.loop)
async def asyncSetUp(self):
loop = asyncio.get_running_loop()
loop._process_events = mock.Mock()
loop._selector = mock.Mock()
loop._selector.select.return_value = ()


class LockTests(BaseTest):

def test_context_manager_async_with(self):
async def test_context_manager_async_with(self):
primitives = [
asyncio.Lock(),
asyncio.Condition(),
Expand All @@ -61,10 +58,10 @@ async def test(lock):
self.assertFalse(lock.locked())

for primitive in primitives:
self.loop.run_until_complete(test(primitive))
await test(primitive)
self.assertFalse(primitive.locked())

def test_context_manager_with_await(self):
async def test_context_manager_with_await(self):
primitives = [
asyncio.Lock(),
asyncio.Condition(),
Expand All @@ -83,13 +80,13 @@ async def test(lock):
pass

for primitive in primitives:
self.loop.run_until_complete(test(primitive))
await test(primitive)
self.assertFalse(primitive.locked())


class StreamReaderTests(BaseTest):

def test_readline(self):
async def test_readline(self):
DATA = b'line1\nline2\nline3'

stream = asyncio.StreamReader(loop=self.loop)
Expand All @@ -102,7 +99,7 @@ async def reader():
data.append(line)
return data

data = self.loop.run_until_complete(reader())
data = await reader()
self.assertEqual(data, [b'line1\n', b'line2\n', b'line3'])


Expand Down Expand Up @@ -170,7 +167,7 @@ async def coro():
data = self.loop.run_until_complete(coro())
self.assertEqual(data, 'spam')

def test_task_print_stack(self):
async def test_task_print_stack(self):
T = None

async def foo():
Expand All @@ -185,9 +182,9 @@ async def runner():
T = asyncio.ensure_future(foo(), loop=self.loop)
await T

self.loop.run_until_complete(runner())
await runner()

def test_double_await(self):
async def test_double_await(self):
async def afunc():
await asyncio.sleep(0.1)

Expand All @@ -200,12 +197,12 @@ async def runner():
finally:
t.cancel()

self.loop.set_debug(True)
asyncio.get_running_loop().set_debug(True)
with self.assertRaises(
RuntimeError,
msg='coroutine is being awaited already'):

self.loop.run_until_complete(runner())
await runner()


if __name__ == '__main__':
Expand Down