Rooba
/
uvloop
mirror of https://github.com/MagicStack/uvloop.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix code style in unittest

This commit is contained in:
claws 2016-05-08 00:35:33 +09:30 committed by Yury Selivanov
parent 77dc53e7b3
commit f894ac7403
10 changed files with 38 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/test_aiohttp.py
View File

@ -6,8 +6,6 @@ except ImportError:
else:
skip_tests = False
import asyncio
import uvloop
import unittest
from uvloop import _testbase as tb

30
tests/test_base.py
View File

@ -131,8 +131,8 @@ class _TestBase:
loop.set_debug(debug)
if debug:
msg = ("Non-thread-safe operation invoked on an event loop other "
"than the current one")
msg = ("Non-thread-safe operation invoked on an "
"event loop other than the current one")
with self.assertRaisesRegex(RuntimeError, msg):
loop.call_soon(cb)
with self.assertRaisesRegex(RuntimeError, msg):
@ -189,6 +189,7 @@ class _TestBase:
def test_run_once_in_executor_plain(self):
called = []
def cb(arg):
called.append(arg)
@ -206,18 +207,17 @@ class _TestBase:
self.assertFalse(self.loop.get_debug())
def test_run_until_complete_type_error(self):
self.assertRaises(TypeError,
self.loop.run_until_complete, 'blah')
self.assertRaises(
TypeError, self.loop.run_until_complete, 'blah')
def test_run_until_complete_loop(self):
task = asyncio.Future(loop=self.loop)
other_loop = self.new_loop()
self.addCleanup(other_loop.close)
self.assertRaises(ValueError,
other_loop.run_until_complete, task)
self.assertRaises(
ValueError, other_loop.run_until_complete, task)
def test_run_until_complete_error(self):
task = asyncio.Future(loop=self.loop)
async def foo():
raise ValueError('aaa')
with self.assertRaisesRegex(ValueError, 'aaa'):
@ -228,7 +228,7 @@ class _TestBase:
def zero_error(fut):
fut.set_result(True)
1/0
1 / 0
logger = logging.getLogger('asyncio')
@ -258,11 +258,12 @@ class _TestBase:
def run_loop():
def zero_error():
self.loop.stop()
1/0
1 / 0
self.loop.call_soon(zero_error)
self.loop.run_forever()
errors = []
def handler(loop, exc):
errors.append(exc)
@ -278,8 +279,8 @@ class _TestBase:
with mock.patch.object(logger, 'error') as log:
run_loop()
log.assert_called_with(
self.mock_pattern('Exception in callback.*zero'),
exc_info=mock.ANY)
self.mock_pattern('Exception in callback.*zero'),
exc_info=mock.ANY)
self.assertEqual(len(errors), 1)
@ -289,7 +290,7 @@ class _TestBase:
def run_loop():
def zero_error():
self.loop.stop()
1/0
1 / 0
self.loop.call_soon(zero_error)
self.loop.run_forever()
@ -328,7 +329,7 @@ class _TestBase:
def run_loop():
def zero_error():
loop.stop()
1/0
1 / 0
loop.call_soon(zero_error)
loop.run_forever()
@ -358,7 +359,8 @@ class _TestBase:
def test_set_task_factory_invalid(self):
with self.assertRaisesRegex(
TypeError, 'task factory must be a callable or None'):
TypeError,
'task factory must be a callable or None'):
self.loop.set_task_factory(1)

4
tests/test_dns.py
View File

@ -1,6 +1,4 @@
import asyncio
import socket
import uvloop
import unittest
from uvloop import _testbase as tb
@ -136,6 +134,6 @@ class Test_UV_DNS(BaseTestDNS, tb.UVTestCase):
class Test_AIO_DNS(BaseTestDNS, tb.AIOTestCase):
pass
def test_getaddrinfo_11(self):
self._test_getaddrinfo(_HOST.encode(), str(_PORT))

2
tests/test_pipes.py
View File

@ -147,6 +147,7 @@ class _BasePipeTest:
transport.write(b'1')
data = bytearray()
def reader(data):
chunk = os.read(rpipe, 1024)
data += chunk
@ -205,6 +206,7 @@ class _BasePipeTest:
transport.write(b'1')
data = bytearray()
def reader(data):
chunk = os.read(master, 1024)
data += chunk

56
tests/test_process.py
View File

@ -1,15 +1,12 @@
import asyncio
import contextlib
import signal
import socket
import subprocess
import sys
import tempfile
import uvloop
from asyncio import test_utils
from uvloop import _testbase as tb
from test import support
class _TestProcess:
@ -276,12 +273,12 @@ class _AsyncioTests:
def len_message(message):
code = 'import sys; data = sys.stdin.read(); print(len(data))'
proc = yield from asyncio.create_subprocess_exec(
sys.executable, '-c', code,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
close_fds=False,
loop=self.loop)
sys.executable, '-c', code,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
close_fds=False,
loop=self.loop)
stdout, stderr = yield from proc.communicate(message)
exitcode = yield from proc.wait()
return (stdout, exitcode)
@ -296,10 +293,10 @@ class _AsyncioTests:
@asyncio.coroutine
def run(data):
proc = yield from asyncio.create_subprocess_exec(
*args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
loop=self.loop)
*args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
loop=self.loop)
# feed data
proc.stdin.write(data)
@ -323,10 +320,10 @@ class _AsyncioTests:
@asyncio.coroutine
def run(data):
proc = yield from asyncio.create_subprocess_exec(
*args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
loop=self.loop)
*args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
loop=self.loop)
stdout, stderr = yield from proc.communicate(data)
return proc.returncode, stdout
@ -389,35 +386,14 @@ class _AsyncioTests:
returncode = self.loop.run_until_complete(send_signal(proc))
self.assertEqual(-signal.SIGHUP, returncode)
def test_stdin_not_inheritable(self):
# asyncio issue #209: stdin must not be inheritable, otherwise
# the Process.communicate() hangs
@asyncio.coroutine
def len_message(message):
code = 'import sys; data = sys.stdin.read(); print(len(data))'
proc = yield from asyncio.create_subprocess_exec(
sys.executable, '-c', code,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
close_fds=False,
loop=self.loop)
stdout, stderr = yield from proc.communicate(message)
exitcode = yield from proc.wait()
return (stdout, exitcode)
output, exitcode = self.loop.run_until_complete(len_message(b'abc'))
self.assertEqual(output.rstrip(), b'3')
self.assertEqual(exitcode, 0)
def test_cancel_process_wait(self):
# Issue #23140: cancel Process.wait()
@asyncio.coroutine
def cancel_wait():
proc = yield from asyncio.create_subprocess_exec(
*self.PROGRAM_BLOCKED,
loop=self.loop)
*self.PROGRAM_BLOCKED,
loop=self.loop)
# Create an internal future waiting on the process exit
task = self.loop.create_task(proc.wait())

1
tests/test_signals.py
View File

@ -3,7 +3,6 @@ import signal
import subprocess
import sys
import time
import uvloop
from uvloop import _testbase as tb

2
tests/test_sockets.py
View File

@ -1,6 +1,4 @@
import asyncio
import socket
import uvloop
from uvloop import _testbase as tb

3
tests/test_tcp.py
View File

@ -1,10 +1,7 @@
import asyncio
import logging
import socket
import uvloop
import ssl
import sys
import warnings
from uvloop import _testbase as tb

3
tests/test_udp.py
View File

@ -1,10 +1,7 @@
import asyncio
import logging
import socket
import uvloop
import ssl
import sys
import warnings
from asyncio import test_utils
from uvloop import _testbase as tb

8
tests/test_unix.py
View File

@ -2,8 +2,6 @@ import asyncio
import os
import socket
import tempfile
import uvloop
import unittest.mock
from uvloop import _testbase as tb
@ -114,7 +112,6 @@ class _TestUnix:
for srv_sock in srv_socks:
self.assertEqual(srv_sock.fileno(), -1)
self.loop.run_until_complete(start_server())
self.assertEqual(CNT, TOTAL_CNT)
@ -246,7 +243,8 @@ class _TestUnix:
for _ in range(TOTAL_CNT):
tasks.append(coro(srv.addr))
self.loop.run_until_complete(asyncio.gather(*tasks, loop=self.loop))
self.loop.run_until_complete(
asyncio.gather(*tasks, loop=self.loop))
srv.join()
self.assertEqual(CNT, TOTAL_CNT)
@ -293,7 +291,7 @@ class _TestUnix:
self.assertEqual(len(excs), 1)
self.assertIn(excs[0].__class__,
(BrokenPipeError, ConnectionResetError))
(BrokenPipeError, ConnectionResetError))
def test_transport_fromsock_get_extra_info(self):
async def test(sock):