import collections
from collections.abc import Generator
import concurrent.futures
+import contextvars
import datetime
import functools
from functools import singledispatch
from tornado.log import app_log
from tornado.util import TimeoutError
-try:
- import contextvars
-except ImportError:
- contextvars = None # type: ignore
-
import typing
from typing import (
Union,
# This function is type-annotated with a comment to work around
# https://bitbucket.org/pypy/pypy/issues/2868/segfault-with-args-type-annotation-in
future = _create_future()
- if contextvars is not None:
- ctx_run: Callable = contextvars.copy_context().run
- else:
- ctx_run = _fake_ctx_run
+ ctx_run: Callable = contextvars.copy_context().run
try:
result = ctx_run(func, *args, **kwargs)
except (Return, StopIteration) as e:
# and ensure that it doesn't blow up (e.g. with unicode/bytes issues in
# python 3)
-import unittest
+from unittest import mock
from tornado.auth import (
OpenIdMixin,
from tornado.testing import AsyncHTTPTestCase, ExpectLog
from tornado.web import RequestHandler, Application, HTTPError
-try:
- from unittest import mock
-except ImportError:
- mock = None # type: ignore
-
class OpenIdClientLoginHandler(RequestHandler, OpenIdMixin):
def initialize(self, test):
response.headers["Set-Cookie"],
)
- @unittest.skipIf(mock is None, "mock package not present")
def test_oauth10a_redirect_error(self):
with mock.patch.object(OAuth1ServerRequestTokenHandler, "get") as get:
get.side_effect = Exception("boom")
import asyncio
from concurrent import futures
+import contextvars
import gc
import datetime
import platform
from tornado import gen
-try:
- import contextvars
-except ImportError:
- contextvars = None # type: ignore
-
class GenBasicTest(AsyncTestCase):
@gen.coroutine
self.assertEqual(result, [None, None])
-if contextvars is not None:
- ctx_var: contextvars.ContextVar[int] = contextvars.ContextVar("ctx_var")
+ctx_var: contextvars.ContextVar[int] = contextvars.ContextVar("ctx_var")
-@unittest.skipIf(contextvars is None, "contextvars module not present")
class ContextVarsTest(AsyncTestCase):
async def native_root(self, x):
ctx_var.set(x)