import threading
import time
import unittest
-import unittest.mock
+from unittest import mock
from http.server import HTTPServer
from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
def MockCallback(**kwargs):
- return unittest.mock.Mock(spec=['__call__'], **kwargs)
+ return mock.Mock(spec=['__call__'], **kwargs)
class MockPattern(str):
import sys
import time
import unittest
-import unittest.mock
+from unittest import mock
from test.support import find_unused_port, IPV6_ENABLED
import asyncio
from asyncio import test_utils
-MOCK_ANY = unittest.mock.ANY
+MOCK_ANY = mock.ANY
PY34 = sys.version_info >= (3, 4)
def setUp(self):
self.loop = base_events.BaseEventLoop()
- self.loop._selector = unittest.mock.Mock()
+ self.loop._selector = mock.Mock()
asyncio.set_event_loop(None)
def test_not_implemented(self):
- m = unittest.mock.Mock()
+ m = mock.Mock()
self.assertRaises(
NotImplementedError,
self.loop._make_socket_transport, m, m)
self.assertFalse(self.loop._ready)
def test_set_default_executor(self):
- executor = unittest.mock.Mock()
+ executor = mock.Mock()
self.loop.set_default_executor(executor)
self.assertIs(executor, self.loop._default_executor)
def test_getnameinfo(self):
- sockaddr = unittest.mock.Mock()
- self.loop.run_in_executor = unittest.mock.Mock()
+ sockaddr = mock.Mock()
+ self.loop.run_in_executor = mock.Mock()
self.loop.getnameinfo(sockaddr)
self.assertEqual(
(None, socket.getnameinfo, sockaddr, 0),
def cb(arg):
calls.append(arg)
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
self.loop.call_later(-1, cb, 'a')
self.loop.call_later(-2, cb, 'b')
test_utils.run_briefly(self.loop)
def cb():
self.loop.stop()
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
delay = 0.1
when = self.loop.time() + delay
pass
h = asyncio.Handle(cb, (), self.loop)
f = asyncio.Future(loop=self.loop)
- executor = unittest.mock.Mock()
+ executor = mock.Mock()
executor.submit.return_value = f
self.loop.set_default_executor(executor)
res = self.loop.run_in_executor(None, h)
self.assertIs(f, res)
- executor = unittest.mock.Mock()
+ executor = mock.Mock()
executor.submit.return_value = f
res = self.loop.run_in_executor(executor, h)
self.assertIs(f, res)
h1.cancel()
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
self.loop._scheduled.append(h1)
self.loop._scheduled.append(h2)
self.loop._run_once()
self.loop.set_debug(False)
self.assertFalse(self.loop.get_debug())
- @unittest.mock.patch('asyncio.base_events.time')
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.time')
+ @mock.patch('asyncio.base_events.logger')
def test__run_once_logging(self, m_logger, m_time):
# Log to INFO level if timeout > 1.0 sec.
idx = -1
self.loop._scheduled.append(
asyncio.TimerHandle(11.0, lambda: True, (), self.loop))
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
self.loop._run_once()
self.assertEqual(logging.INFO, m_logger.log.call_args[0][0])
h = asyncio.TimerHandle(time.monotonic() - 1, cb, (self.loop,),
self.loop)
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
self.loop._scheduled.append(h)
self.loop._run_once()
asyncio.SubprocessProtocol, 'exit 0', bufsize=4096)
def test_default_exc_handler_callback(self):
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
def zero_error(fut):
fut.set_result(True)
1/0
# Test call_soon (events.Handle)
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
fut = asyncio.Future(loop=self.loop)
self.loop.call_soon(zero_error, fut)
fut.add_done_callback(lambda fut: self.loop.stop())
exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
# Test call_later (events.TimerHandle)
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
fut = asyncio.Future(loop=self.loop)
self.loop.call_later(0.01, zero_error, fut)
fut.add_done_callback(lambda fut: self.loop.stop())
exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
def test_default_exc_handler_coro(self):
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
@asyncio.coroutine
def zero_error_coro():
1/0
# Test Future.__del__
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
fut = asyncio.async(zero_error_coro(), loop=self.loop)
fut.add_done_callback(lambda *args: self.loop.stop())
self.loop.run_forever()
self.loop.call_soon(zero_error)
self.loop._run_once()
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
- mock_handler = unittest.mock.Mock()
+ mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
run_loop()
mock_handler.assert_called_with(self.loop, {
mock_handler.reset_mock()
self.loop.set_exception_handler(None)
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
run_loop()
log.error.assert_called_with(
test_utils.MockPattern(
def handler(loop, context):
raise AttributeError('spam')
- self.loop._process_events = unittest.mock.Mock()
+ self.loop._process_events = mock.Mock()
self.loop.set_exception_handler(handler)
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
run_loop()
log.error.assert_called_with(
test_utils.MockPattern(
class Loop(base_events.BaseEventLoop):
- _selector = unittest.mock.Mock()
- _process_events = unittest.mock.Mock()
+ _selector = mock.Mock()
+ _process_events = mock.Mock()
def default_exception_handler(self, context):
nonlocal _context
loop.call_soon(zero_error)
loop._run_once()
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
run_loop()
log.error.assert_called_with(
'Exception in default exception handler',
_context = None
loop.set_exception_handler(custom_handler)
- with unittest.mock.patch('asyncio.base_events.logger') as log:
+ with mock.patch('asyncio.base_events.logger') as log:
run_loop()
log.error.assert_called_with(
test_utils.MockPattern('Exception in default exception.*'
def tearDown(self):
self.loop.close()
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_connection_multiple_errors(self, m_socket):
class MyProto(asyncio.Protocol):
return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
self.loop.getaddrinfo = getaddrinfo_task
- self.loop.sock_connect = unittest.mock.Mock()
+ self.loop.sock_connect = mock.Mock()
self.loop.sock_connect.side_effect = OSError
coro = self.loop.create_connection(MyProto, 'example.com', 80)
return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
self.loop.getaddrinfo = getaddrinfo_task
- self.loop.sock_connect = unittest.mock.Mock()
+ self.loop.sock_connect = mock.Mock()
self.loop.sock_connect.side_effect = OSError
coro = self.loop.create_connection(
with self.assertRaises(OSError):
self.loop.run_until_complete(coro)
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_connection_multiple_errors_local_addr(self, m_socket):
def bind(addr):
return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
self.loop.getaddrinfo = getaddrinfo_task
- self.loop.sock_connect = unittest.mock.Mock()
+ self.loop.sock_connect = mock.Mock()
self.loop.sock_connect.side_effect = OSError('Err2')
coro = self.loop.create_connection(
OSError, self.loop.run_until_complete, coro)
def test_create_connection_ssl_server_hostname_default(self):
- self.loop.getaddrinfo = unittest.mock.Mock()
+ self.loop.getaddrinfo = mock.Mock()
def mock_getaddrinfo(*args, **kwds):
f = asyncio.Future(loop=self.loop)
return f
self.loop.getaddrinfo.side_effect = mock_getaddrinfo
- self.loop.sock_connect = unittest.mock.Mock()
+ self.loop.sock_connect = mock.Mock()
self.loop.sock_connect.return_value = ()
- self.loop._make_ssl_transport = unittest.mock.Mock()
+ self.loop._make_ssl_transport = mock.Mock()
class _SelectorTransportMock:
_sock = None
return transport
self.loop._make_ssl_transport.side_effect = mock_make_ssl_transport
- ANY = unittest.mock.ANY
+ ANY = mock.ANY
# First try the default server_hostname.
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True)
self.assertRaises(ValueError, self.loop.run_until_complete, fut)
def test_create_server_no_getaddrinfo(self):
- getaddrinfo = self.loop.getaddrinfo = unittest.mock.Mock()
+ getaddrinfo = self.loop.getaddrinfo = mock.Mock()
getaddrinfo.return_value = []
f = self.loop.create_server(MyProto, '0.0.0.0', 0)
self.assertRaises(OSError, self.loop.run_until_complete, f)
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_server_cant_bind(self, m_socket):
class Err(OSError):
m_socket.getaddrinfo.return_value = [
(2, 1, 6, '', ('127.0.0.1', 10100))]
m_socket.getaddrinfo._is_coroutine = False
- m_sock = m_socket.socket.return_value = unittest.mock.Mock()
+ m_sock = m_socket.socket.return_value = mock.Mock()
m_sock.bind.side_effect = Err
fut = self.loop.create_server(MyProto, '0.0.0.0', 0)
self.assertRaises(OSError, self.loop.run_until_complete, fut)
self.assertTrue(m_sock.close.called)
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_datagram_endpoint_no_addrinfo(self, m_socket):
m_socket.getaddrinfo.return_value = []
m_socket.getaddrinfo._is_coroutine = False
AssertionError, self.loop.run_until_complete, coro)
def test_create_datagram_endpoint_connect_err(self):
- self.loop.sock_connect = unittest.mock.Mock()
+ self.loop.sock_connect = mock.Mock()
self.loop.sock_connect.side_effect = OSError
coro = self.loop.create_datagram_endpoint(
self.assertRaises(
OSError, self.loop.run_until_complete, coro)
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_datagram_endpoint_socket_err(self, m_socket):
m_socket.getaddrinfo = socket.getaddrinfo
m_socket.socket.side_effect = OSError
self.assertRaises(
ValueError, self.loop.run_until_complete, coro)
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_datagram_endpoint_setblk_err(self, m_socket):
m_socket.socket.return_value.setblocking.side_effect = OSError
asyncio.DatagramProtocol)
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
- @unittest.mock.patch('asyncio.base_events.socket')
+ @mock.patch('asyncio.base_events.socket')
def test_create_datagram_endpoint_cant_bind(self, m_socket):
class Err(OSError):
pass
m_socket.AF_INET6 = socket.AF_INET6
m_socket.getaddrinfo = socket.getaddrinfo
- m_sock = m_socket.socket.return_value = unittest.mock.Mock()
+ m_sock = m_socket.socket.return_value = mock.Mock()
m_sock.bind.side_effect = Err
fut = self.loop.create_datagram_endpoint(
self.assertTrue(m_sock.close.called)
def test_accept_connection_retry(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.accept.side_effect = BlockingIOError()
self.loop._accept_connection(MyProto, sock)
self.assertFalse(sock.close.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_accept_connection_exception(self, m_log):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.accept.side_effect = OSError(errno.EMFILE, 'Too many open files')
- self.loop.remove_reader = unittest.mock.Mock()
- self.loop.call_later = unittest.mock.Mock()
+ self.loop.remove_reader = mock.Mock()
+ self.loop.call_later = mock.Mock()
self.loop._accept_connection(MyProto, sock)
self.assertTrue(m_log.error.called)
self.loop.remove_reader.assert_called_with(10)
self.loop.call_later.assert_called_with(constants.ACCEPT_RETRY_DELAY,
# self.loop._start_serving
- unittest.mock.ANY,
+ mock.ANY,
MyProto, sock, None, None)
def test_call_coroutine(self):
import time
import errno
import unittest
-import unittest.mock
+from unittest import mock
from test import support # find_unused_port, IPV6_ENABLED, TEST_HOME_DIR
return args
args = ()
- h = asyncio.Handle(callback, args, unittest.mock.Mock())
+ h = asyncio.Handle(callback, args, mock.Mock())
self.assertIs(h._callback, callback)
self.assertIs(h._args, args)
self.assertFalse(h._cancelled)
def callback():
raise ValueError()
- m_loop = unittest.mock.Mock()
- m_loop.call_exception_handler = unittest.mock.Mock()
+ m_loop = mock.Mock()
+ m_loop.call_exception_handler = mock.Mock()
h = asyncio.Handle(callback, (), m_loop)
h._run()
m_loop.call_exception_handler.assert_called_with({
'message': test_utils.MockPattern('Exception in callback.*'),
- 'exception': unittest.mock.ANY,
+ 'exception': mock.ANY,
'handle': h
})
def test_hash(self):
when = time.monotonic()
h = asyncio.TimerHandle(when, lambda: False, (),
- unittest.mock.Mock())
+ mock.Mock())
self.assertEqual(hash(h), hash(when))
def test_timer(self):
args = ()
when = time.monotonic()
- h = asyncio.TimerHandle(when, callback, args, unittest.mock.Mock())
+ h = asyncio.TimerHandle(when, callback, args, mock.Mock())
self.assertIs(h._callback, callback)
self.assertIs(h._args, args)
self.assertFalse(h._cancelled)
self.assertRaises(AssertionError,
asyncio.TimerHandle, None, callback, args,
- unittest.mock.Mock())
+ mock.Mock())
def test_timer_comparison(self):
- loop = unittest.mock.Mock()
+ loop = mock.Mock()
def callback(*args):
return args
class AbstractEventLoopTests(unittest.TestCase):
def test_not_implemented(self):
- f = unittest.mock.Mock()
+ f = mock.Mock()
loop = asyncio.AbstractEventLoop()
self.assertRaises(
NotImplementedError, loop.run_forever)
NotImplementedError, loop.remove_signal_handler, 1)
self.assertRaises(
NotImplementedError, loop.connect_read_pipe, f,
- unittest.mock.sentinel.pipe)
+ mock.sentinel.pipe)
self.assertRaises(
NotImplementedError, loop.connect_write_pipe, f,
- unittest.mock.sentinel.pipe)
+ mock.sentinel.pipe)
self.assertRaises(
NotImplementedError, loop.subprocess_shell, f,
- unittest.mock.sentinel)
+ mock.sentinel)
self.assertRaises(
NotImplementedError, loop.subprocess_exec, f)
class ProtocolsAbsTests(unittest.TestCase):
def test_empty(self):
- f = unittest.mock.Mock()
+ f = mock.Mock()
p = asyncio.Protocol()
self.assertIsNone(p.connection_made(f))
self.assertIsNone(p.connection_lost(f))
def test_get_event_loop_calls_set_event_loop(self):
policy = asyncio.DefaultEventLoopPolicy()
- with unittest.mock.patch.object(
+ with mock.patch.object(
policy, "set_event_loop",
wraps=policy.set_event_loop) as m_set_event_loop:
policy.set_event_loop(None)
self.assertRaises(AssertionError, policy.get_event_loop)
- @unittest.mock.patch('asyncio.events.threading.current_thread')
+ @mock.patch('asyncio.events.threading.current_thread')
def test_get_event_loop_thread(self, m_current_thread):
def f():
import concurrent.futures
import threading
import unittest
-import unittest.mock
+from unittest import mock
import asyncio
from asyncio import test_utils
self.assertRaises(AssertionError, test)
fut.cancel()
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_tb_logger_abandoned(self, m_log):
fut = asyncio.Future(loop=self.loop)
del fut
self.assertFalse(m_log.error.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_tb_logger_result_unretrieved(self, m_log):
fut = asyncio.Future(loop=self.loop)
fut.set_result(42)
del fut
self.assertFalse(m_log.error.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_tb_logger_result_retrieved(self, m_log):
fut = asyncio.Future(loop=self.loop)
fut.set_result(42)
del fut
self.assertFalse(m_log.error.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_tb_logger_exception_unretrieved(self, m_log):
fut = asyncio.Future(loop=self.loop)
fut.set_exception(RuntimeError('boom'))
test_utils.run_briefly(self.loop)
self.assertTrue(m_log.error.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_tb_logger_exception_retrieved(self, m_log):
fut = asyncio.Future(loop=self.loop)
fut.set_exception(RuntimeError('boom'))
del fut
self.assertFalse(m_log.error.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_tb_logger_exception_result_retrieved(self, m_log):
fut = asyncio.Future(loop=self.loop)
fut.set_exception(RuntimeError('boom'))
f2 = asyncio.wrap_future(f1)
self.assertIs(f1, f2)
- @unittest.mock.patch('asyncio.futures.events')
+ @mock.patch('asyncio.futures.events')
def test_wrap_future_use_global_loop(self, m_events):
def run(arg):
return (arg, threading.get_ident())
"""Tests for lock.py"""
import unittest
-import unittest.mock
+from unittest import mock
import re
import asyncio
self.loop.close()
def test_ctor_loop(self):
- loop = unittest.mock.Mock()
+ loop = mock.Mock()
lock = asyncio.Lock(loop=loop)
self.assertIs(lock._loop, loop)
self.loop.close()
def test_ctor_loop(self):
- loop = unittest.mock.Mock()
+ loop = mock.Mock()
ev = asyncio.Event(loop=loop)
self.assertIs(ev._loop, loop)
self.assertTrue(repr(ev).endswith('[set]>'))
self.assertTrue(RGX_REPR.match(repr(ev)))
- ev._waiters.append(unittest.mock.Mock())
+ ev._waiters.append(mock.Mock())
self.assertTrue('waiters:1' in repr(ev))
self.assertTrue(RGX_REPR.match(repr(ev)))
self.loop.close()
def test_ctor_loop(self):
- loop = unittest.mock.Mock()
+ loop = mock.Mock()
cond = asyncio.Condition(loop=loop)
self.assertIs(cond._loop, loop)
self.loop.run_until_complete(cond.acquire())
self.assertTrue('locked' in repr(cond))
- cond._waiters.append(unittest.mock.Mock())
+ cond._waiters.append(mock.Mock())
self.assertTrue('waiters:1' in repr(cond))
self.assertTrue(RGX_REPR.match(repr(cond)))
- cond._waiters.append(unittest.mock.Mock())
+ cond._waiters.append(mock.Mock())
self.assertTrue('waiters:2' in repr(cond))
self.assertTrue(RGX_REPR.match(repr(cond)))
self.loop.close()
def test_ctor_loop(self):
- loop = unittest.mock.Mock()
+ loop = mock.Mock()
sem = asyncio.Semaphore(loop=loop)
self.assertIs(sem._loop, loop)
self.assertTrue('waiters' not in repr(sem))
self.assertTrue(RGX_REPR.match(repr(sem)))
- sem._waiters.append(unittest.mock.Mock())
+ sem._waiters.append(mock.Mock())
self.assertTrue('waiters:1' in repr(sem))
self.assertTrue(RGX_REPR.match(repr(sem)))
- sem._waiters.append(unittest.mock.Mock())
+ sem._waiters.append(mock.Mock())
self.assertTrue('waiters:2' in repr(sem))
self.assertTrue(RGX_REPR.match(repr(sem)))
import socket
import unittest
-import unittest.mock
+from unittest import mock
import asyncio
from asyncio.proactor_events import BaseProactorEventLoop
def setUp(self):
self.loop = test_utils.TestLoop()
- self.proactor = unittest.mock.Mock()
+ self.proactor = mock.Mock()
self.loop._proactor = self.proactor
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
- self.sock = unittest.mock.Mock(socket.socket)
+ self.sock = mock.Mock(socket.socket)
def test_ctor(self):
fut = asyncio.Future(loop=self.loop)
self.assertRaises(AssertionError, tr._loop_reading, res)
- tr.close = unittest.mock.Mock()
+ tr.close = mock.Mock()
tr._read_fut = res
tr._loop_reading(res)
self.assertFalse(self.loop._proactor.recv.called)
err = self.loop._proactor.recv.side_effect = ConnectionAbortedError()
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._fatal_error = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
tr._loop_reading()
tr._fatal_error.assert_called_with(
err,
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
tr._closing = True
- tr._fatal_error = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
tr._loop_reading()
self.assertFalse(tr._fatal_error.called)
self.loop._proactor.recv.side_effect = ConnectionAbortedError()
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
tr._closing = False
- tr._fatal_error = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
tr._loop_reading()
self.assertTrue(tr._fatal_error.called)
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
tr._closing = False
- tr._fatal_error = unittest.mock.Mock()
- tr._force_close = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
+ tr._force_close = mock.Mock()
tr._loop_reading()
self.assertFalse(tr._fatal_error.called)
tr._force_close.assert_called_with(err)
err = self.loop._proactor.recv.side_effect = (OSError())
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._fatal_error = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
tr._loop_reading()
tr._fatal_error.assert_called_with(
err,
def test_write(self):
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._loop_writing = unittest.mock.Mock()
+ tr._loop_writing = mock.Mock()
tr.write(b'data')
self.assertEqual(tr._buffer, None)
tr._loop_writing.assert_called_with(data=b'data')
def test_write_more(self):
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._write_fut = unittest.mock.Mock()
- tr._loop_writing = unittest.mock.Mock()
+ tr._write_fut = mock.Mock()
+ tr._loop_writing = mock.Mock()
tr.write(b'data')
self.assertEqual(tr._buffer, b'data')
self.assertFalse(tr._loop_writing.called)
self.loop._proactor.send.return_value.add_done_callback.\
assert_called_with(tr._loop_writing)
- @unittest.mock.patch('asyncio.proactor_events.logger')
+ @mock.patch('asyncio.proactor_events.logger')
def test_loop_writing_err(self, m_log):
err = self.loop._proactor.send.side_effect = OSError()
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._fatal_error = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
tr._buffer = [b'da', b'ta']
tr._loop_writing()
tr._fatal_error.assert_called_with(
def test_abort(self):
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._force_close = unittest.mock.Mock()
+ tr._force_close = mock.Mock()
tr.abort()
tr._force_close.assert_called_with(None)
def test_close_write_fut(self):
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._write_fut = unittest.mock.Mock()
+ tr._write_fut = mock.Mock()
tr.close()
test_utils.run_briefly(self.loop)
self.assertFalse(self.protocol.connection_lost.called)
test_utils.run_briefly(self.loop)
self.assertFalse(self.protocol.connection_lost.called)
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_fatal_error(self, m_logging):
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
- tr._force_close = unittest.mock.Mock()
+ tr._force_close = mock.Mock()
tr._fatal_error(None)
self.assertTrue(tr._force_close.called)
self.assertTrue(m_logging.error.called)
def test_force_close(self):
tr = _ProactorSocketTransport(self.loop, self.sock, self.protocol)
tr._buffer = [b'data']
- read_fut = tr._read_fut = unittest.mock.Mock()
- write_fut = tr._write_fut = unittest.mock.Mock()
+ read_fut = tr._read_fut = mock.Mock()
+ write_fut = tr._write_fut = mock.Mock()
tr._force_close(None)
read_fut.cancel.assert_called_with()
class BaseProactorEventLoopTests(unittest.TestCase):
def setUp(self):
- self.sock = unittest.mock.Mock(socket.socket)
- self.proactor = unittest.mock.Mock()
+ self.sock = mock.Mock(socket.socket)
+ self.proactor = mock.Mock()
- self.ssock, self.csock = unittest.mock.Mock(), unittest.mock.Mock()
+ self.ssock, self.csock = mock.Mock(), mock.Mock()
class EventLoop(BaseProactorEventLoop):
def _socketpair(s):
self.loop = EventLoop(self.proactor)
- @unittest.mock.patch.object(BaseProactorEventLoop, 'call_soon')
- @unittest.mock.patch.object(BaseProactorEventLoop, '_socketpair')
+ @mock.patch.object(BaseProactorEventLoop, 'call_soon')
+ @mock.patch.object(BaseProactorEventLoop, '_socketpair')
def test_ctor(self, socketpair, call_soon):
ssock, csock = socketpair.return_value = (
- unittest.mock.Mock(), unittest.mock.Mock())
+ mock.Mock(), mock.Mock())
loop = BaseProactorEventLoop(self.proactor)
self.assertIs(loop._ssock, ssock)
self.assertIs(loop._csock, csock)
self.assertIsNone(self.loop._csock)
def test_close(self):
- self.loop._close_self_pipe = unittest.mock.Mock()
+ self.loop._close_self_pipe = mock.Mock()
self.loop.close()
self.assertTrue(self.loop._close_self_pipe.called)
self.assertTrue(self.proactor.close.called)
self.loop._loop_self_reading)
def test_loop_self_reading_fut(self):
- fut = unittest.mock.Mock()
+ fut = mock.Mock()
self.loop._loop_self_reading(fut)
self.assertTrue(fut.result.called)
self.proactor.recv.assert_called_with(self.ssock, 4096)
self.loop._loop_self_reading)
def test_loop_self_reading_exception(self):
- self.loop.close = unittest.mock.Mock()
+ self.loop.close = mock.Mock()
self.proactor.recv.side_effect = OSError()
self.assertRaises(OSError, self.loop._loop_self_reading)
self.assertTrue(self.loop.close.called)
def test_process_events(self):
self.loop._process_events([])
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_create_server(self, m_log):
- pf = unittest.mock.Mock()
- call_soon = self.loop.call_soon = unittest.mock.Mock()
+ pf = mock.Mock()
+ call_soon = self.loop.call_soon = mock.Mock()
self.loop._start_serving(pf, self.sock)
self.assertTrue(call_soon.called)
self.proactor.accept.assert_called_with(self.sock)
# conn
- fut = unittest.mock.Mock()
- fut.result.return_value = (unittest.mock.Mock(), unittest.mock.Mock())
+ fut = mock.Mock()
+ fut.result.return_value = (mock.Mock(), mock.Mock())
- make_tr = self.loop._make_socket_transport = unittest.mock.Mock()
+ make_tr = self.loop._make_socket_transport = mock.Mock()
loop(fut)
self.assertTrue(fut.result.called)
self.assertTrue(make_tr.called)
self.assertTrue(m_log.error.called)
def test_create_server_cancel(self):
- pf = unittest.mock.Mock()
- call_soon = self.loop.call_soon = unittest.mock.Mock()
+ pf = mock.Mock()
+ call_soon = self.loop.call_soon = mock.Mock()
self.loop._start_serving(pf, self.sock)
loop = call_soon.call_args[0][0]
self.assertTrue(self.sock.close.called)
def test_stop_serving(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
self.loop._stop_serving(sock)
self.assertTrue(sock.close.called)
self.proactor._stop_serving.assert_called_with(sock)
"""Tests for queues.py"""
import unittest
-import unittest.mock
+from unittest import mock
import asyncio
from asyncio import test_utils
self.assertTrue('_queue=[1]' in fn(q))
def test_ctor_loop(self):
- loop = unittest.mock.Mock()
+ loop = mock.Mock()
q = asyncio.Queue(loop=loop)
self.assertIs(q._loop, loop)
import socket
import sys
import unittest
-import unittest.mock
+from unittest import mock
try:
import ssl
except ImportError:
from asyncio.selector_events import _SelectorDatagramTransport
-MOCK_ANY = unittest.mock.ANY
+MOCK_ANY = mock.ANY
class TestBaseSelectorEventLoop(BaseSelectorEventLoop):
def _make_self_pipe(self):
- self._ssock = unittest.mock.Mock()
- self._csock = unittest.mock.Mock()
+ self._ssock = mock.Mock()
+ self._csock = mock.Mock()
self._internal_fds += 1
class BaseSelectorEventLoopTests(unittest.TestCase):
def setUp(self):
- selector = unittest.mock.Mock()
+ selector = mock.Mock()
self.loop = TestBaseSelectorEventLoop(selector)
def test_make_socket_transport(self):
- m = unittest.mock.Mock()
- self.loop.add_reader = unittest.mock.Mock()
+ m = mock.Mock()
+ self.loop.add_reader = mock.Mock()
transport = self.loop._make_socket_transport(m, asyncio.Protocol())
self.assertIsInstance(transport, _SelectorSocketTransport)
@unittest.skipIf(ssl is None, 'No ssl module')
def test_make_ssl_transport(self):
- m = unittest.mock.Mock()
- self.loop.add_reader = unittest.mock.Mock()
- self.loop.add_writer = unittest.mock.Mock()
- self.loop.remove_reader = unittest.mock.Mock()
- self.loop.remove_writer = unittest.mock.Mock()
+ m = mock.Mock()
+ self.loop.add_reader = mock.Mock()
+ self.loop.add_writer = mock.Mock()
+ self.loop.remove_reader = mock.Mock()
+ self.loop.remove_writer = mock.Mock()
waiter = asyncio.Future(loop=self.loop)
transport = self.loop._make_ssl_transport(
m, asyncio.Protocol(), m, waiter)
self.assertIsInstance(transport, _SelectorSslTransport)
- @unittest.mock.patch('asyncio.selector_events.ssl', None)
+ @mock.patch('asyncio.selector_events.ssl', None)
def test_make_ssl_transport_without_ssl_error(self):
- m = unittest.mock.Mock()
- self.loop.add_reader = unittest.mock.Mock()
- self.loop.add_writer = unittest.mock.Mock()
- self.loop.remove_reader = unittest.mock.Mock()
- self.loop.remove_writer = unittest.mock.Mock()
+ m = mock.Mock()
+ self.loop.add_reader = mock.Mock()
+ self.loop.add_writer = mock.Mock()
+ self.loop.remove_reader = mock.Mock()
+ self.loop.remove_writer = mock.Mock()
with self.assertRaises(RuntimeError):
self.loop._make_ssl_transport(m, m, m, m)
ssock.fileno.return_value = 7
csock = self.loop._csock
csock.fileno.return_value = 1
- remove_reader = self.loop.remove_reader = unittest.mock.Mock()
+ remove_reader = self.loop.remove_reader = mock.Mock()
self.loop._selector.close()
- self.loop._selector = selector = unittest.mock.Mock()
+ self.loop._selector = selector = mock.Mock()
self.loop.close()
self.assertIsNone(self.loop._selector)
self.assertIsNone(self.loop._csock)
def test_close_no_selector(self):
ssock = self.loop._ssock
csock = self.loop._csock
- remove_reader = self.loop.remove_reader = unittest.mock.Mock()
+ remove_reader = self.loop.remove_reader = mock.Mock()
self.loop._selector.close()
self.loop._selector = None
self.assertRaises(OSError, self.loop._write_to_self)
def test_sock_recv(self):
- sock = unittest.mock.Mock()
- self.loop._sock_recv = unittest.mock.Mock()
+ sock = mock.Mock()
+ self.loop._sock_recv = mock.Mock()
f = self.loop.sock_recv(sock, 1024)
self.assertIsInstance(f, asyncio.Future)
self.loop._sock_recv.assert_called_with(f, False, sock, 1024)
def test__sock_recv_canceled_fut(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
f.cancel()
self.assertFalse(sock.recv.called)
def test__sock_recv_unregister(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
f = asyncio.Future(loop=self.loop)
f.cancel()
- self.loop.remove_reader = unittest.mock.Mock()
+ self.loop.remove_reader = mock.Mock()
self.loop._sock_recv(f, True, sock, 1024)
self.assertEqual((10,), self.loop.remove_reader.call_args[0])
def test__sock_recv_tryagain(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.recv.side_effect = BlockingIOError
- self.loop.add_reader = unittest.mock.Mock()
+ self.loop.add_reader = mock.Mock()
self.loop._sock_recv(f, False, sock, 1024)
self.assertEqual((10, self.loop._sock_recv, f, True, sock, 1024),
self.loop.add_reader.call_args[0])
def test__sock_recv_exception(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
err = sock.recv.side_effect = OSError()
self.assertIs(err, f.exception())
def test_sock_sendall(self):
- sock = unittest.mock.Mock()
- self.loop._sock_sendall = unittest.mock.Mock()
+ sock = mock.Mock()
+ self.loop._sock_sendall = mock.Mock()
f = self.loop.sock_sendall(sock, b'data')
self.assertIsInstance(f, asyncio.Future)
self.loop._sock_sendall.call_args[0])
def test_sock_sendall_nodata(self):
- sock = unittest.mock.Mock()
- self.loop._sock_sendall = unittest.mock.Mock()
+ sock = mock.Mock()
+ self.loop._sock_sendall = mock.Mock()
f = self.loop.sock_sendall(sock, b'')
self.assertIsInstance(f, asyncio.Future)
self.assertFalse(self.loop._sock_sendall.called)
def test__sock_sendall_canceled_fut(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
f.cancel()
self.assertFalse(sock.send.called)
def test__sock_sendall_unregister(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
f = asyncio.Future(loop=self.loop)
f.cancel()
- self.loop.remove_writer = unittest.mock.Mock()
+ self.loop.remove_writer = mock.Mock()
self.loop._sock_sendall(f, True, sock, b'data')
self.assertEqual((10,), self.loop.remove_writer.call_args[0])
def test__sock_sendall_tryagain(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.send.side_effect = BlockingIOError
- self.loop.add_writer = unittest.mock.Mock()
+ self.loop.add_writer = mock.Mock()
self.loop._sock_sendall(f, False, sock, b'data')
self.assertEqual(
(10, self.loop._sock_sendall, f, True, sock, b'data'),
def test__sock_sendall_interrupted(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.send.side_effect = InterruptedError
- self.loop.add_writer = unittest.mock.Mock()
+ self.loop.add_writer = mock.Mock()
self.loop._sock_sendall(f, False, sock, b'data')
self.assertEqual(
(10, self.loop._sock_sendall, f, True, sock, b'data'),
def test__sock_sendall_exception(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
err = sock.send.side_effect = OSError()
self.assertIs(f.exception(), err)
def test__sock_sendall(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
sock.fileno.return_value = 10
self.assertIsNone(f.result())
def test__sock_sendall_partial(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
sock.fileno.return_value = 10
sock.send.return_value = 2
- self.loop.add_writer = unittest.mock.Mock()
+ self.loop.add_writer = mock.Mock()
self.loop._sock_sendall(f, False, sock, b'data')
self.assertFalse(f.done())
self.assertEqual(
self.loop.add_writer.call_args[0])
def test__sock_sendall_none(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
sock.fileno.return_value = 10
sock.send.return_value = 0
- self.loop.add_writer = unittest.mock.Mock()
+ self.loop.add_writer = mock.Mock()
self.loop._sock_sendall(f, False, sock, b'data')
self.assertFalse(f.done())
self.assertEqual(
self.loop.add_writer.call_args[0])
def test_sock_connect(self):
- sock = unittest.mock.Mock()
- self.loop._sock_connect = unittest.mock.Mock()
+ sock = mock.Mock()
+ self.loop._sock_connect = mock.Mock()
f = self.loop.sock_connect(sock, ('127.0.0.1', 8080))
self.assertIsInstance(f, asyncio.Future)
def test__sock_connect(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
self.loop._sock_connect(f, False, sock, ('127.0.0.1', 8080))
self.assertTrue(sock.connect.called)
def test__sock_connect_canceled_fut(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
f.cancel()
self.assertFalse(sock.connect.called)
def test__sock_connect_unregister(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
f = asyncio.Future(loop=self.loop)
f.cancel()
- self.loop.remove_writer = unittest.mock.Mock()
+ self.loop.remove_writer = mock.Mock()
self.loop._sock_connect(f, True, sock, ('127.0.0.1', 8080))
self.assertEqual((10,), self.loop.remove_writer.call_args[0])
def test__sock_connect_tryagain(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.getsockopt.return_value = errno.EAGAIN
- self.loop.add_writer = unittest.mock.Mock()
- self.loop.remove_writer = unittest.mock.Mock()
+ self.loop.add_writer = mock.Mock()
+ self.loop.remove_writer = mock.Mock()
self.loop._sock_connect(f, True, sock, ('127.0.0.1', 8080))
self.assertEqual(
def test__sock_connect_exception(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.getsockopt.return_value = errno.ENOTCONN
- self.loop.remove_writer = unittest.mock.Mock()
+ self.loop.remove_writer = mock.Mock()
self.loop._sock_connect(f, True, sock, ('127.0.0.1', 8080))
self.assertIsInstance(f.exception(), OSError)
def test_sock_accept(self):
- sock = unittest.mock.Mock()
- self.loop._sock_accept = unittest.mock.Mock()
+ sock = mock.Mock()
+ self.loop._sock_accept = mock.Mock()
f = self.loop.sock_accept(sock)
self.assertIsInstance(f, asyncio.Future)
def test__sock_accept(self):
f = asyncio.Future(loop=self.loop)
- conn = unittest.mock.Mock()
+ conn = mock.Mock()
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.accept.return_value = conn, ('127.0.0.1', 1000)
self.assertEqual((False,), conn.setblocking.call_args[0])
def test__sock_accept_canceled_fut(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
f = asyncio.Future(loop=self.loop)
f.cancel()
self.assertFalse(sock.accept.called)
def test__sock_accept_unregister(self):
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
f = asyncio.Future(loop=self.loop)
f.cancel()
- self.loop.remove_reader = unittest.mock.Mock()
+ self.loop.remove_reader = mock.Mock()
self.loop._sock_accept(f, True, sock)
self.assertEqual((10,), self.loop.remove_reader.call_args[0])
def test__sock_accept_tryagain(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
sock.accept.side_effect = BlockingIOError
- self.loop.add_reader = unittest.mock.Mock()
+ self.loop.add_reader = mock.Mock()
self.loop._sock_accept(f, False, sock)
self.assertEqual(
(10, self.loop._sock_accept, f, True, sock),
def test__sock_accept_exception(self):
f = asyncio.Future(loop=self.loop)
- sock = unittest.mock.Mock()
+ sock = mock.Mock()
sock.fileno.return_value = 10
err = sock.accept.side_effect = OSError()
self.assertIsNone(w)
def test_add_reader_existing(self):
- reader = unittest.mock.Mock()
- writer = unittest.mock.Mock()
+ reader = mock.Mock()
+ writer = mock.Mock()
self.loop._selector.get_key.return_value = selectors.SelectorKey(
1, 1, selectors.EVENT_WRITE, (reader, writer))
cb = lambda: True
self.assertEqual(writer, w)
def test_add_reader_existing_writer(self):
- writer = unittest.mock.Mock()
+ writer = mock.Mock()
self.loop._selector.get_key.return_value = selectors.SelectorKey(
1, 1, selectors.EVENT_WRITE, (None, writer))
cb = lambda: True
self.assertTrue(self.loop._selector.unregister.called)
def test_remove_reader_read_write(self):
- reader = unittest.mock.Mock()
- writer = unittest.mock.Mock()
+ reader = mock.Mock()
+ writer = mock.Mock()
self.loop._selector.get_key.return_value = selectors.SelectorKey(
1, 1, selectors.EVENT_READ | selectors.EVENT_WRITE,
(reader, writer))
self.assertEqual(cb, w._callback)
def test_add_writer_existing(self):
- reader = unittest.mock.Mock()
- writer = unittest.mock.Mock()
+ reader = mock.Mock()
+ writer = mock.Mock()
self.loop._selector.get_key.return_value = selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (reader, writer))
cb = lambda: True
self.assertTrue(self.loop._selector.unregister.called)
def test_remove_writer_read_write(self):
- reader = unittest.mock.Mock()
- writer = unittest.mock.Mock()
+ reader = mock.Mock()
+ writer = mock.Mock()
self.loop._selector.get_key.return_value = selectors.SelectorKey(
1, 1, selectors.EVENT_READ | selectors.EVENT_WRITE,
(reader, writer))
self.loop.remove_writer(1))
def test_process_events_read(self):
- reader = unittest.mock.Mock()
+ reader = mock.Mock()
reader._cancelled = False
- self.loop._add_callback = unittest.mock.Mock()
+ self.loop._add_callback = mock.Mock()
self.loop._process_events(
[(selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (reader, None)),
self.loop._add_callback.assert_called_with(reader)
def test_process_events_read_cancelled(self):
- reader = unittest.mock.Mock()
+ reader = mock.Mock()
reader.cancelled = True
- self.loop.remove_reader = unittest.mock.Mock()
+ self.loop.remove_reader = mock.Mock()
self.loop._process_events(
[(selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (reader, None)),
self.loop.remove_reader.assert_called_with(1)
def test_process_events_write(self):
- writer = unittest.mock.Mock()
+ writer = mock.Mock()
writer._cancelled = False
- self.loop._add_callback = unittest.mock.Mock()
+ self.loop._add_callback = mock.Mock()
self.loop._process_events(
[(selectors.SelectorKey(1, 1, selectors.EVENT_WRITE,
(None, writer)),
self.loop._add_callback.assert_called_with(writer)
def test_process_events_write_cancelled(self):
- writer = unittest.mock.Mock()
+ writer = mock.Mock()
writer.cancelled = True
- self.loop.remove_writer = unittest.mock.Mock()
+ self.loop.remove_writer = mock.Mock()
self.loop._process_events(
[(selectors.SelectorKey(1, 1, selectors.EVENT_WRITE,
def setUp(self):
self.loop = test_utils.TestLoop()
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
- self.sock = unittest.mock.Mock(socket.socket)
+ self.sock = mock.Mock(socket.socket)
self.sock.fileno.return_value = 7
def test_ctor(self):
def test_abort(self):
tr = _SelectorTransport(self.loop, self.sock, self.protocol, None)
- tr._force_close = unittest.mock.Mock()
+ tr._force_close = mock.Mock()
tr.abort()
tr._force_close.assert_called_with(None)
def test_force_close(self):
tr = _SelectorTransport(self.loop, self.sock, self.protocol, None)
tr._buffer.extend(b'1')
- self.loop.add_reader(7, unittest.mock.sentinel)
- self.loop.add_writer(7, unittest.mock.sentinel)
+ self.loop.add_reader(7, mock.sentinel)
+ self.loop.add_writer(7, mock.sentinel)
tr._force_close(None)
self.assertTrue(tr._closing)
self.assertFalse(self.loop.readers)
self.assertEqual(1, self.loop.remove_reader_count[7])
- @unittest.mock.patch('asyncio.log.logger.error')
+ @mock.patch('asyncio.log.logger.error')
def test_fatal_error(self, m_exc):
exc = OSError()
tr = _SelectorTransport(self.loop, self.sock, self.protocol, None)
- tr._force_close = unittest.mock.Mock()
+ tr._force_close = mock.Mock()
tr._fatal_error(exc)
m_exc.assert_called_with(
def setUp(self):
self.loop = test_utils.TestLoop()
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
- self.sock = unittest.mock.Mock(socket.socket)
+ self.sock = mock.Mock(socket.socket)
self.sock_fd = self.sock.fileno.return_value = 7
def test_ctor(self):
def test_read_ready_eof(self):
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport.close = unittest.mock.Mock()
+ transport.close = mock.Mock()
self.sock.recv.return_value = b''
transport._read_ready()
def test_read_ready_eof_keep_open(self):
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport.close = unittest.mock.Mock()
+ transport.close = mock.Mock()
self.sock.recv.return_value = b''
self.protocol.eof_received.return_value = True
self.protocol.eof_received.assert_called_with()
self.assertFalse(transport.close.called)
- @unittest.mock.patch('logging.exception')
+ @mock.patch('logging.exception')
def test_read_ready_tryagain(self, m_exc):
self.sock.recv.side_effect = BlockingIOError
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
self.assertFalse(transport._fatal_error.called)
- @unittest.mock.patch('logging.exception')
+ @mock.patch('logging.exception')
def test_read_ready_tryagain_interrupted(self, m_exc):
self.sock.recv.side_effect = InterruptedError
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
self.assertFalse(transport._fatal_error.called)
- @unittest.mock.patch('logging.exception')
+ @mock.patch('logging.exception')
def test_read_ready_conn_reset(self, m_exc):
err = self.sock.recv.side_effect = ConnectionResetError()
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport._force_close = unittest.mock.Mock()
+ transport._force_close = mock.Mock()
transport._read_ready()
transport._force_close.assert_called_with(err)
- @unittest.mock.patch('logging.exception')
+ @mock.patch('logging.exception')
def test_read_ready_err(self, m_exc):
err = self.sock.recv.side_effect = OSError()
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
transport._fatal_error.assert_called_with(
self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)
- @unittest.mock.patch('asyncio.selector_events.logger')
+ @mock.patch('asyncio.selector_events.logger')
def test_write_exception(self, m_log):
err = self.sock.send.side_effect = OSError()
data = b'data'
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport.write(data)
transport._fatal_error.assert_called_with(
err,
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._buffer.extend(b'data')
transport._write_ready()
transport._fatal_error.assert_called_with(
err,
'Fatal write error on socket transport')
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.base_events.logger')
def test_write_ready_exception_and_close(self, m_log):
self.sock.send.side_effect = OSError()
- remove_writer = self.loop.remove_writer = unittest.mock.Mock()
+ remove_writer = self.loop.remove_writer = mock.Mock()
transport = _SelectorSocketTransport(
self.loop, self.sock, self.protocol)
def setUp(self):
self.loop = test_utils.TestLoop()
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
- self.sock = unittest.mock.Mock(socket.socket)
+ self.sock = mock.Mock(socket.socket)
self.sock.fileno.return_value = 7
- self.sslsock = unittest.mock.Mock()
+ self.sslsock = mock.Mock()
self.sslsock.fileno.return_value = 1
- self.sslcontext = unittest.mock.Mock()
+ self.sslcontext = mock.Mock()
self.sslcontext.wrap_socket.return_value = self.sslsock
def _make_one(self, create_waiter=None):
transport.write(b'data')
self.assertEqual(transport._conn_lost, 2)
- @unittest.mock.patch('asyncio.selector_events.logger')
+ @mock.patch('asyncio.selector_events.logger')
def test_write_exception(self, m_log):
transport = self._make_one()
transport._conn_lost = 1
self.assertEqual((b'data',), self.protocol.data_received.call_args[0])
def test_read_ready_write_wants_read(self):
- self.loop.add_writer = unittest.mock.Mock()
+ self.loop.add_writer = mock.Mock()
self.sslsock.recv.side_effect = BlockingIOError
transport = self._make_one()
transport._write_wants_read = True
- transport._write_ready = unittest.mock.Mock()
+ transport._write_ready = mock.Mock()
transport._buffer.extend(b'data')
transport._read_ready()
def test_read_ready_recv_eof(self):
self.sslsock.recv.return_value = b''
transport = self._make_one()
- transport.close = unittest.mock.Mock()
+ transport.close = mock.Mock()
transport._read_ready()
transport.close.assert_called_with()
self.protocol.eof_received.assert_called_with()
def test_read_ready_recv_conn_reset(self):
err = self.sslsock.recv.side_effect = ConnectionResetError()
transport = self._make_one()
- transport._force_close = unittest.mock.Mock()
+ transport._force_close = mock.Mock()
transport._read_ready()
transport._force_close.assert_called_with(err)
self.assertFalse(self.protocol.data_received.called)
def test_read_ready_recv_write(self):
- self.loop.remove_reader = unittest.mock.Mock()
- self.loop.add_writer = unittest.mock.Mock()
+ self.loop.remove_reader = mock.Mock()
+ self.loop.add_writer = mock.Mock()
self.sslsock.recv.side_effect = ssl.SSLWantWriteError
transport = self._make_one()
transport._read_ready()
def test_read_ready_recv_exc(self):
err = self.sslsock.recv.side_effect = OSError()
transport = self._make_one()
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
transport._fatal_error.assert_called_with(
err,
transport = self._make_one()
transport._buffer = list_to_buffer([b'data'])
- self.loop.remove_writer = unittest.mock.Mock()
+ self.loop.remove_writer = mock.Mock()
self.sslsock.send.side_effect = ssl.SSLWantReadError
transport._write_ready()
self.assertFalse(self.protocol.data_received.called)
transport = self._make_one()
transport._buffer = list_to_buffer([b'data'])
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._write_ready()
transport._fatal_error.assert_called_with(
err,
self.assertEqual(list_to_buffer(), transport._buffer)
def test_write_ready_read_wants_write(self):
- self.loop.add_reader = unittest.mock.Mock()
+ self.loop.add_reader = mock.Mock()
self.sslsock.send.side_effect = BlockingIOError
transport = self._make_one()
transport._read_wants_write = True
- transport._read_ready = unittest.mock.Mock()
+ transport._read_ready = mock.Mock()
transport._write_ready()
self.assertFalse(transport._read_wants_write)
class SelectorSslWithoutSslTransportTests(unittest.TestCase):
- @unittest.mock.patch('asyncio.selector_events.ssl', None)
+ @mock.patch('asyncio.selector_events.ssl', None)
def test_ssl_transport_requires_ssl_module(self):
- Mock = unittest.mock.Mock
+ Mock = mock.Mock
with self.assertRaises(RuntimeError):
transport = _SelectorSslTransport(Mock(), Mock(), Mock(), Mock())
def setUp(self):
self.loop = test_utils.TestLoop()
self.protocol = test_utils.make_test_protocol(asyncio.DatagramProtocol)
- self.sock = unittest.mock.Mock(spec_set=socket.socket)
+ self.sock = mock.Mock(spec_set=socket.socket)
self.sock.fileno.return_value = 7
def test_read_ready(self):
self.loop, self.sock, self.protocol)
self.sock.recvfrom.side_effect = BlockingIOError
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
self.assertFalse(transport._fatal_error.called)
self.loop, self.sock, self.protocol)
err = self.sock.recvfrom.side_effect = RuntimeError()
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
transport._fatal_error.assert_called_with(
self.loop, self.sock, self.protocol)
err = self.sock.recvfrom.side_effect = OSError()
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._read_ready()
self.assertFalse(transport._fatal_error.called)
self.assertEqual(
[(b'data', ('0.0.0.0', 12345))], list(transport._buffer))
- @unittest.mock.patch('asyncio.selector_events.logger')
+ @mock.patch('asyncio.selector_events.logger')
def test_sendto_exception(self, m_log):
data = b'data'
err = self.sock.sendto.side_effect = RuntimeError()
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport.sendto(data, ())
self.assertTrue(transport._fatal_error.called)
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport.sendto(data, ())
self.assertEqual(transport._conn_lost, 0)
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol, ('0.0.0.0', 1))
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport.sendto(data)
self.assertFalse(transport._fatal_error.called)
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._buffer.append((b'data', ()))
transport._sendto_ready()
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol)
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._buffer.append((b'data', ()))
transport._sendto_ready()
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol, ('0.0.0.0', 1))
- transport._fatal_error = unittest.mock.Mock()
+ transport._fatal_error = mock.Mock()
transport._buffer.append((b'data', ()))
transport._sendto_ready()
self.assertFalse(transport._fatal_error.called)
self.assertTrue(self.protocol.error_received.called)
- @unittest.mock.patch('asyncio.base_events.logger.error')
+ @mock.patch('asyncio.base_events.logger.error')
def test_fatal_error_connected(self, m_exc):
transport = _SelectorDatagramTransport(
self.loop, self.sock, self.protocol, ('0.0.0.0', 1))
import gc
import socket
import unittest
-import unittest.mock
+from unittest import mock
try:
import ssl
except ImportError:
self.loop.close()
gc.collect()
- @unittest.mock.patch('asyncio.streams.events')
+ @mock.patch('asyncio.streams.events')
def test_ctor_global_loop(self, m_events):
stream = asyncio.StreamReader()
self.assertIs(stream._loop, m_events.get_event_loop.return_value)
"""Tests for transports.py."""
import unittest
-import unittest.mock
+from unittest import mock
import asyncio
from asyncio import transports
def test_writelines(self):
transport = asyncio.Transport()
- transport.write = unittest.mock.Mock()
+ transport.write = mock.Mock()
transport.writelines([b'line1',
bytearray(b'line2'),
return 512
transport = MyTransport()
- transport._protocol = unittest.mock.Mock()
+ transport._protocol = mock.Mock()
self.assertFalse(transport._protocol_paused)
import tempfile
import threading
import unittest
-import unittest.mock
+from unittest import mock
if sys.platform == 'win32':
raise unittest.SkipTest('UNIX only')
from asyncio import unix_events
-MOCK_ANY = unittest.mock.ANY
+MOCK_ANY = mock.ANY
@unittest.skipUnless(signal, 'Signals are not supported')
self.loop._handle_signal(signal.NSIG + 1, ())
def test_handle_signal_cancelled_handler(self):
- h = asyncio.Handle(unittest.mock.Mock(), (),
- loop=unittest.mock.Mock())
+ h = asyncio.Handle(mock.Mock(), (),
+ loop=mock.Mock())
h.cancel()
self.loop._signal_handlers[signal.NSIG + 1] = h
- self.loop.remove_signal_handler = unittest.mock.Mock()
+ self.loop.remove_signal_handler = mock.Mock()
self.loop._handle_signal(signal.NSIG + 1, ())
self.loop.remove_signal_handler.assert_called_with(signal.NSIG + 1)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_add_signal_handler_setup_error(self, m_signal):
m_signal.NSIG = signal.NSIG
m_signal.set_wakeup_fd.side_effect = ValueError
self.loop.add_signal_handler,
signal.SIGINT, lambda: True)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_add_signal_handler(self, m_signal):
m_signal.NSIG = signal.NSIG
self.assertIsInstance(h, asyncio.Handle)
self.assertEqual(h._callback, cb)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_add_signal_handler_install_error(self, m_signal):
m_signal.NSIG = signal.NSIG
self.loop.add_signal_handler,
signal.SIGINT, lambda: True)
- @unittest.mock.patch('asyncio.unix_events.signal')
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.base_events.logger')
def test_add_signal_handler_install_error2(self, m_logging, m_signal):
m_signal.NSIG = signal.NSIG
self.assertFalse(m_logging.info.called)
self.assertEqual(1, m_signal.set_wakeup_fd.call_count)
- @unittest.mock.patch('asyncio.unix_events.signal')
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.base_events.logger')
def test_add_signal_handler_install_error3(self, m_logging, m_signal):
class Err(OSError):
errno = errno.EINVAL
self.assertFalse(m_logging.info.called)
self.assertEqual(2, m_signal.set_wakeup_fd.call_count)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_remove_signal_handler(self, m_signal):
m_signal.NSIG = signal.NSIG
self.assertEqual(
(signal.SIGHUP, m_signal.SIG_DFL), m_signal.signal.call_args[0])
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_remove_signal_handler_2(self, m_signal):
m_signal.NSIG = signal.NSIG
m_signal.SIGINT = signal.SIGINT
(signal.SIGINT, m_signal.default_int_handler),
m_signal.signal.call_args[0])
- @unittest.mock.patch('asyncio.unix_events.signal')
- @unittest.mock.patch('asyncio.base_events.logger')
+ @mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.base_events.logger')
def test_remove_signal_handler_cleanup_error(self, m_logging, m_signal):
m_signal.NSIG = signal.NSIG
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)
self.loop.remove_signal_handler(signal.SIGHUP)
self.assertTrue(m_logging.info)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_remove_signal_handler_error(self, m_signal):
m_signal.NSIG = signal.NSIG
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)
self.assertRaises(
OSError, self.loop.remove_signal_handler, signal.SIGHUP)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_remove_signal_handler_error2(self, m_signal):
m_signal.NSIG = signal.NSIG
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)
self.assertRaises(
RuntimeError, self.loop.remove_signal_handler, signal.SIGHUP)
- @unittest.mock.patch('asyncio.unix_events.signal')
+ @mock.patch('asyncio.unix_events.signal')
def test_close(self, m_signal):
m_signal.NSIG = signal.NSIG
def setUp(self):
self.loop = test_utils.TestLoop()
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
- self.pipe = unittest.mock.Mock(spec_set=io.RawIOBase)
+ self.pipe = mock.Mock(spec_set=io.RawIOBase)
self.pipe.fileno.return_value = 5
- fcntl_patcher = unittest.mock.patch('fcntl.fcntl')
+ fcntl_patcher = mock.patch('fcntl.fcntl')
fcntl_patcher.start()
self.addCleanup(fcntl_patcher.stop)
- fstat_patcher = unittest.mock.patch('os.fstat')
+ fstat_patcher = mock.patch('os.fstat')
m_fstat = fstat_patcher.start()
- st = unittest.mock.Mock()
+ st = mock.Mock()
st.st_mode = stat.S_IFIFO
m_fstat.return_value = st
self.addCleanup(fstat_patcher.stop)
test_utils.run_briefly(self.loop)
self.assertIsNone(fut.result())
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test__read_ready(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
m_read.assert_called_with(5, tr.max_size)
self.protocol.data_received.assert_called_with(b'data')
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test__read_ready_eof(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
self.protocol.eof_received.assert_called_with()
self.protocol.connection_lost.assert_called_with(None)
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test__read_ready_blocked(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
test_utils.run_briefly(self.loop)
self.assertFalse(self.protocol.data_received.called)
- @unittest.mock.patch('asyncio.log.logger.error')
- @unittest.mock.patch('os.read')
+ @mock.patch('asyncio.log.logger.error')
+ @mock.patch('os.read')
def test__read_ready_error(self, m_read, m_logexc):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
err = OSError()
m_read.side_effect = err
- tr._close = unittest.mock.Mock()
+ tr._close = mock.Mock()
tr._read_ready()
m_read.assert_called_with(5, tr.max_size)
'\nprotocol:.*\ntransport:.*'),
exc_info=(OSError, MOCK_ANY, MOCK_ANY))
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test_pause_reading(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
- m = unittest.mock.Mock()
+ m = mock.Mock()
self.loop.add_reader(5, m)
tr.pause_reading()
self.assertFalse(self.loop.readers)
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test_resume_reading(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
tr.resume_reading()
self.loop.assert_reader(5, tr._read_ready)
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test_close(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
- tr._close = unittest.mock.Mock()
+ tr._close = mock.Mock()
tr.close()
tr._close.assert_called_with(None)
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test_close_already_closing(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
tr._closing = True
- tr._close = unittest.mock.Mock()
+ tr._close = mock.Mock()
tr.close()
self.assertFalse(tr._close.called)
- @unittest.mock.patch('os.read')
+ @mock.patch('os.read')
def test__close(self, m_read):
tr = unix_events._UnixReadPipeTransport(
self.loop, self.pipe, self.protocol)
def setUp(self):
self.loop = test_utils.TestLoop()
self.protocol = test_utils.make_test_protocol(asyncio.BaseProtocol)
- self.pipe = unittest.mock.Mock(spec_set=io.RawIOBase)
+ self.pipe = mock.Mock(spec_set=io.RawIOBase)
self.pipe.fileno.return_value = 5
- fcntl_patcher = unittest.mock.patch('fcntl.fcntl')
+ fcntl_patcher = mock.patch('fcntl.fcntl')
fcntl_patcher.start()
self.addCleanup(fcntl_patcher.stop)
- fstat_patcher = unittest.mock.patch('os.fstat')
+ fstat_patcher = mock.patch('os.fstat')
m_fstat = fstat_patcher.start()
- st = unittest.mock.Mock()
+ st = mock.Mock()
st.st_mode = stat.S_IFSOCK
m_fstat.return_value = st
self.addCleanup(fstat_patcher.stop)
self.loop, self.pipe, self.protocol)
self.assertTrue(tr.can_write_eof())
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_write(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.assertFalse(self.loop.writers)
self.assertEqual([], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_write_no_data(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.assertFalse(self.loop.writers)
self.assertEqual([], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_write_partial(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual([b'ta'], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_write_buffer(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual([b'previous', b'data'], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_write_again(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual([b'data'], tr._buffer)
- @unittest.mock.patch('asyncio.unix_events.logger')
- @unittest.mock.patch('os.write')
+ @mock.patch('asyncio.unix_events.logger')
+ @mock.patch('os.write')
def test_write_err(self, m_write, m_log):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
err = OSError()
m_write.side_effect = err
- tr._fatal_error = unittest.mock.Mock()
+ tr._fatal_error = mock.Mock()
tr.write(b'data')
m_write.assert_called_with(5, b'data')
self.assertFalse(self.loop.writers)
m_log.warning.assert_called_with(
'pipe closed by peer or os.write(pipe, data) raised exception.')
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_write_close(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
test_utils.run_briefly(self.loop)
self.protocol.connection_lost.assert_called_with(None)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test__write_ready(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.assertFalse(self.loop.writers)
self.assertEqual([], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test__write_ready_partial(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual([b'a'], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test__write_ready_again(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual([b'data'], tr._buffer)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test__write_ready_empty(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual([b'data'], tr._buffer)
- @unittest.mock.patch('asyncio.log.logger.error')
- @unittest.mock.patch('os.write')
+ @mock.patch('asyncio.log.logger.error')
+ @mock.patch('os.write')
def test__write_ready_err(self, m_write, m_logexc):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
test_utils.run_briefly(self.loop)
self.protocol.connection_lost.assert_called_with(err)
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test__write_ready_closing(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
self.protocol.connection_lost.assert_called_with(None)
self.pipe.close.assert_called_with()
- @unittest.mock.patch('os.write')
+ @mock.patch('os.write')
def test_abort(self, m_write):
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
- tr.write_eof = unittest.mock.Mock()
+ tr.write_eof = mock.Mock()
tr.close()
tr.write_eof.assert_called_with()
tr = unix_events._UnixWritePipeTransport(
self.loop, self.pipe, self.protocol)
- tr.write_eof = unittest.mock.Mock()
+ tr.write_eof = mock.Mock()
tr._closing = True
tr.close()
self.assertFalse(tr.write_eof.called)
class AbstractChildWatcherTests(unittest.TestCase):
def test_not_implemented(self):
- f = unittest.mock.Mock()
+ f = mock.Mock()
watcher = asyncio.AbstractChildWatcher()
self.assertRaises(
NotImplementedError, watcher.add_child_handler, f, f)
class BaseChildWatcherTests(unittest.TestCase):
def test_not_implemented(self):
- f = unittest.mock.Mock()
+ f = mock.Mock()
watcher = unix_events.BaseChildWatcher()
self.assertRaises(
NotImplementedError, watcher._do_waitpid, f)
class ChildWatcherTestsMixin:
- ignore_warnings = unittest.mock.patch.object(log.logger, "warning")
+ ignore_warnings = mock.patch.object(log.logger, "warning")
def setUp(self):
self.loop = test_utils.TestLoop()
self.running = False
self.zombies = {}
- with unittest.mock.patch.object(
+ with mock.patch.object(
self.loop, "add_signal_handler") as self.m_add_signal_handler:
self.watcher = self.create_watcher()
self.watcher.attach_loop(self.loop)
def waitpid_mocks(func):
def wrapped_func(self):
def patch(target, wrapper):
- return unittest.mock.patch(target, wraps=wrapper,
- new_callable=unittest.mock.Mock)
+ return mock.patch(target, wraps=wrapper,
+ new_callable=mock.Mock)
with patch('os.WTERMSIG', self.WTERMSIG) as m_WTERMSIG, \
patch('os.WEXITSTATUS', self.WEXITSTATUS) as m_WEXITSTATUS, \
@waitpid_mocks
def test_sigchld(self, m):
# register a child
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
with self.watcher:
self.running = True
@waitpid_mocks
def test_sigchld_two_children(self, m):
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
# register child 1
with self.watcher:
@waitpid_mocks
def test_sigchld_two_children_terminating_together(self, m):
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
# register child 1
with self.watcher:
@waitpid_mocks
def test_sigchld_race_condition(self, m):
# register a child
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
with self.watcher:
# child terminates before being registered
@waitpid_mocks
def test_sigchld_replace_handler(self, m):
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
# register a child
with self.watcher:
@waitpid_mocks
def test_sigchld_remove_handler(self, m):
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
# register a child
with self.watcher:
@waitpid_mocks
def test_sigchld_unknown_status(self, m):
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
# register a child
with self.watcher:
@waitpid_mocks
def test_remove_child_handler(self, m):
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
- callback3 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
+ callback3 = mock.Mock()
# register children
with self.watcher:
@waitpid_mocks
def test_sigchld_unhandled_exception(self, m):
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
# register a child
with self.watcher:
# raise an exception
m.waitpid.side_effect = ValueError
- with unittest.mock.patch.object(log.logger,
+ with mock.patch.object(log.logger,
'error') as m_error:
self.assertEqual(self.watcher._sig_chld(), None)
@waitpid_mocks
def test_sigchld_child_reaped_elsewhere(self, m):
# register a child
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
with self.watcher:
self.running = True
@waitpid_mocks
def test_sigchld_unknown_pid_during_registration(self, m):
# register two children
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
with self.ignore_warnings, self.watcher:
self.running = True
@waitpid_mocks
def test_set_loop(self, m):
# register a child
- callback = unittest.mock.Mock()
+ callback = mock.Mock()
with self.watcher:
self.running = True
old_loop = self.loop
self.loop = test_utils.TestLoop()
- with unittest.mock.patch.object(
+ with mock.patch.object(
old_loop,
"remove_signal_handler") as m_old_remove_signal_handler, \
- unittest.mock.patch.object(
+ mock.patch.object(
self.loop,
"add_signal_handler") as m_new_add_signal_handler:
@waitpid_mocks
def test_set_loop_race_condition(self, m):
# register 3 children
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
- callback3 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
+ callback3 = mock.Mock()
with self.watcher:
self.running = True
old_loop = self.loop
self.loop = None
- with unittest.mock.patch.object(
+ with mock.patch.object(
old_loop, "remove_signal_handler") as m_remove_signal_handler:
self.watcher.attach_loop(None)
# attach a new loop
self.loop = test_utils.TestLoop()
- with unittest.mock.patch.object(
+ with mock.patch.object(
self.loop, "add_signal_handler") as m_add_signal_handler:
self.watcher.attach_loop(self.loop)
@waitpid_mocks
def test_close(self, m):
# register two children
- callback1 = unittest.mock.Mock()
- callback2 = unittest.mock.Mock()
+ callback1 = mock.Mock()
+ callback2 = mock.Mock()
with self.watcher:
self.running = True
if isinstance(self.watcher, asyncio.FastChildWatcher):
self.assertEqual(len(self.watcher._zombies), 1)
- with unittest.mock.patch.object(
+ with mock.patch.object(
self.loop,
"remove_signal_handler") as m_remove_signal_handler:
import sys
import test.support
import unittest
-import unittest.mock
+from unittest import mock
if sys.platform != 'win32':
raise unittest.SkipTest('Windows only')
csock.close()
ssock.close()
- @unittest.mock.patch('asyncio.windows_utils.socket')
+ @mock.patch('asyncio.windows_utils.socket')
def test_winsocketpair_exc(self, m_socket):
m_socket.socket.return_value.getsockname.return_value = ('', 12345)
m_socket.socket.return_value.accept.return_value = object(), object()
projects / python / commitdiff