Merge with Python 3.4

This commit is contained in:
Victor Stinner 2014-07-14 22:26:57 +02:00
commit bbbf45b08d
3 changed files with 37 additions and 17 deletions

View File

@ -3,6 +3,7 @@
import collections import collections
import contextlib import contextlib
import io import io
import logging
import os import os
import re import re
import socket import socket
@ -28,6 +29,7 @@ from . import futures
from . import selectors from . import selectors
from . import tasks from . import tasks
from .coroutines import coroutine from .coroutines import coroutine
from .log import logger
if sys.platform == 'win32': # pragma: no cover if sys.platform == 'win32': # pragma: no cover
@ -401,3 +403,17 @@ class TestCase(unittest.TestCase):
def tearDown(self): def tearDown(self):
events.set_event_loop(None) events.set_event_loop(None)
@contextlib.contextmanager
def disable_logger():
"""Context manager to disable asyncio logger.
For example, it can be used to ignore warnings in debug mode.
"""
old_level = logger.level
try:
logger.setLevel(logging.CRITICAL+1)
yield
finally:
logger.setLevel(old_level)

View File

@ -819,6 +819,7 @@ class EventLoopTestsMixin:
# no CA loaded # no CA loaded
f_c = self.loop.create_connection(MyProto, host, port, f_c = self.loop.create_connection(MyProto, host, port,
ssl=sslcontext_client) ssl=sslcontext_client)
with test_utils.disable_logger():
with self.assertRaisesRegex(ssl.SSLError, with self.assertRaisesRegex(ssl.SSLError,
'certificate verify failed '): 'certificate verify failed '):
self.loop.run_until_complete(f_c) self.loop.run_until_complete(f_c)
@ -845,6 +846,7 @@ class EventLoopTestsMixin:
f_c = self.loop.create_unix_connection(MyProto, path, f_c = self.loop.create_unix_connection(MyProto, path,
ssl=sslcontext_client, ssl=sslcontext_client,
server_hostname='invalid') server_hostname='invalid')
with test_utils.disable_logger():
with self.assertRaisesRegex(ssl.SSLError, with self.assertRaisesRegex(ssl.SSLError,
'certificate verify failed '): 'certificate verify failed '):
self.loop.run_until_complete(f_c) self.loop.run_until_complete(f_c)
@ -871,6 +873,7 @@ class EventLoopTestsMixin:
# incorrect server_hostname # incorrect server_hostname
f_c = self.loop.create_connection(MyProto, host, port, f_c = self.loop.create_connection(MyProto, host, port,
ssl=sslcontext_client) ssl=sslcontext_client)
with test_utils.disable_logger():
with self.assertRaisesRegex( with self.assertRaisesRegex(
ssl.CertificateError, ssl.CertificateError,
"hostname '127.0.0.1' doesn't match 'localhost'"): "hostname '127.0.0.1' doesn't match 'localhost'"):

View File

@ -1105,13 +1105,13 @@ class SelectorSslTransportTests(test_utils.TestCase):
def test_on_handshake_exc(self): def test_on_handshake_exc(self):
exc = ValueError() exc = ValueError()
self.sslsock.do_handshake.side_effect = exc self.sslsock.do_handshake.side_effect = exc
with test_utils.disable_logger():
waiter = asyncio.Future(loop=self.loop)
transport = _SelectorSslTransport( transport = _SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext) self.loop, self.sock, self.protocol, self.sslcontext, waiter)
transport._waiter = asyncio.Future(loop=self.loop) self.assertTrue(waiter.done())
transport._on_handshake(None) self.assertIs(exc, waiter.exception())
self.assertTrue(self.sslsock.close.called) self.assertTrue(self.sslsock.close.called)
self.assertTrue(transport._waiter.done())
self.assertIs(exc, transport._waiter.exception())
def test_on_handshake_base_exc(self): def test_on_handshake_base_exc(self):
transport = _SelectorSslTransport( transport = _SelectorSslTransport(
@ -1119,6 +1119,7 @@ class SelectorSslTransportTests(test_utils.TestCase):
transport._waiter = asyncio.Future(loop=self.loop) transport._waiter = asyncio.Future(loop=self.loop)
exc = BaseException() exc = BaseException()
self.sslsock.do_handshake.side_effect = exc self.sslsock.do_handshake.side_effect = exc
with test_utils.disable_logger():
self.assertRaises(BaseException, transport._on_handshake, None) self.assertRaises(BaseException, transport._on_handshake, None)
self.assertTrue(self.sslsock.close.called) self.assertTrue(self.sslsock.close.called)
self.assertTrue(transport._waiter.done()) self.assertTrue(transport._waiter.done())