iframe-proxy | Sunbelt Computer Software

History

156 lines (134 loc) · 5.73 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

import asyncio

import contextvars

import inspect

import warnings

from .case import TestCase

__unittest = True

class IsolatedAsyncioTestCase(TestCase):

# Names intentionally have a long prefix

# to reduce a chance of clashing with user-defined attributes

# from inherited test case

# The class doesn't call loop.run_until_complete(self.setUp()) and family

# but uses a different approach:

# 1. create a long-running task that reads self.setUp()

# awaitable from queue along with a future

# 2. await the awaitable object passing in and set the result

# into the future object

# 3. Outer code puts the awaitable and the future object into a queue

# with waiting for the future

# The trick is necessary because every run_until_complete() call

# creates a new task with embedded ContextVar context.

# To share contextvars between setUp(), test and tearDown() we need to execute

# them inside the same task.

# Note: the test case sets the per-thread event loop unless

# loop_factory=asyncio.EventLoop is set.

# I believe this is not an issue in user level tests but python itself for testing

# should reset the event loop in every test module

# by calling asyncio.set_event_loop(None) in tearDownModule()

# or set loop_factory=asyncio.EventLoop

loop_factory = None

def __init__(self, methodName='runTest'):

super().__init__(methodName)

self._asyncioRunner = None

self._asyncioTestContext = contextvars.copy_context()

async def asyncSetUp(self):

pass

async def asyncTearDown(self):

pass

def addAsyncCleanup(self, func, /, *args, **kwargs):

# A trivial trampoline to addCleanup()

# the function exists because it has a different semantics

# and signature:

# addCleanup() accepts regular functions

# but addAsyncCleanup() accepts coroutines

# We intentionally don't add inspect.iscoroutinefunction() check

# for func argument because there is no way

# to check for async function reliably:

# 1. It can be "async def func()" itself

# 2. Class can implement "async def __call__()" method

# 3. Regular "def func()" that returns awaitable object

self.addCleanup(*(func, *args), **kwargs)

async def enterAsyncContext(self, cm):

"""Enters the supplied asynchronous context manager.

If successful, also adds its __aexit__ method as a cleanup

function and returns the result of the __aenter__ method.

"""

# We look up the special methods on the type to match the with

# statement.

cls = type(cm)

try:

enter = cls.__aenter__

exit = cls.__aexit__

except AttributeError:

msg = (f"'{cls.__module__}.{cls.__qualname__}' object does "

"not support the asynchronous context manager protocol")

try:

cls.__enter__

cls.__exit__

except AttributeError:

pass

else:

msg += (" but it supports the context manager protocol. "

"Did you mean to use enterContext()?")

raise TypeError(msg) from None

result = await enter(cm)

self.addAsyncCleanup(exit, cm, None, None, None)

return result

def _callSetUp(self):

# Force loop to be initialized and set as the current loop

# so that setUp functions can use get_event_loop() and get the

# correct loop instance.

self._asyncioRunner.get_loop()

self._asyncioTestContext.run(self.setUp)

self._callAsync(self.asyncSetUp)

def _callTestMethod(self, method):

result = self._callMaybeAsync(method)

if result is not None:

msg = (

f'It is deprecated to return a value that is not None '

f'from a test case ({method} returned {type(result).__name__!r})',

)

warnings.warn(msg, DeprecationWarning, stacklevel=4)

def _callTearDown(self):

self._callAsync(self.asyncTearDown)

self._asyncioTestContext.run(self.tearDown)

def _callCleanup(self, function, *args, **kwargs):

self._callMaybeAsync(function, *args, **kwargs)

def _callAsync(self, func, /, *args, **kwargs):

assert self._asyncioRunner is not None, 'asyncio runner is not initialized'

assert inspect.iscoroutinefunction(func), f'{func!r} is not an async function'

return self._asyncioRunner.run(

func(*args, **kwargs),

context=self._asyncioTestContext

)

def _callMaybeAsync(self, func, /, *args, **kwargs):

assert self._asyncioRunner is not None, 'asyncio runner is not initialized'

if inspect.iscoroutinefunction(func):

return self._asyncioRunner.run(

func(*args, **kwargs),

context=self._asyncioTestContext,

)

else:

return self._asyncioTestContext.run(func, *args, **kwargs)

def _setupAsyncioRunner(self):

assert self._asyncioRunner is None, 'asyncio runner is already initialized'

runner = asyncio.Runner(debug=True, loop_factory=self.loop_factory)

self._asyncioRunner = runner

def _tearDownAsyncioRunner(self):

runner = self._asyncioRunner

runner.close()

def run(self, result=None):

self._setupAsyncioRunner()

try:

return super().run(result)

finally:

self._tearDownAsyncioRunner()

def debug(self):

self._setupAsyncioRunner()

super().debug()

self._tearDownAsyncioRunner()

def __del__(self):

if self._asyncioRunner is not None:

self._tearDownAsyncioRunner()

Sunbelt Computer Software

PL/B Language Development and Support

Navigation Menu

Uh oh!

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Uh oh!

FilesExpand file tree

async_case.py

Latest commit

History

async_case.py

File metadata and controls