Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

bpo-40275: Use new test.support helper submodules in tests #21449

Merged
merged 3 commits into from
Aug 3, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Lib/test/test__opcode.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import dis
from test.support import import_module
from test.support.import_helper import import_module
import unittest

_opcode = import_module("_opcode")
Expand Down
26 changes: 14 additions & 12 deletions Lib/test/test_asyncore.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,10 @@
import threading

from test import support
from test.support import os_helper
from test.support import socket_helper
from test.support import threading_helper
from test.support import warnings_helper
from io import BytesIO

if support.PGO:
Expand Down Expand Up @@ -92,7 +94,7 @@ def bind_af_aware(sock, addr):
"""Helper function to bind a socket according to its family."""
if HAS_UNIX_SOCKETS and sock.family == socket.AF_UNIX:
# Make sure the path doesn't exist.
support.unlink(addr)
os_helper.unlink(addr)
socket_helper.bind_unix_socket(sock, addr)
else:
sock.bind(addr)
Expand Down Expand Up @@ -369,14 +371,14 @@ def test_send(self):
class FileWrapperTest(unittest.TestCase):
def setUp(self):
self.d = b"It's not dead, it's sleeping!"
with open(support.TESTFN, 'wb') as file:
with open(os_helper.TESTFN, 'wb') as file:
file.write(self.d)

def tearDown(self):
support.unlink(support.TESTFN)
os_helper.unlink(os_helper.TESTFN)

def test_recv(self):
fd = os.open(support.TESTFN, os.O_RDONLY)
fd = os.open(os_helper.TESTFN, os.O_RDONLY)
w = asyncore.file_wrapper(fd)
os.close(fd)

Expand All @@ -390,20 +392,20 @@ def test_recv(self):
def test_send(self):
d1 = b"Come again?"
d2 = b"I want to buy some cheese."
fd = os.open(support.TESTFN, os.O_WRONLY | os.O_APPEND)
fd = os.open(os_helper.TESTFN, os.O_WRONLY | os.O_APPEND)
w = asyncore.file_wrapper(fd)
os.close(fd)

w.write(d1)
w.send(d2)
w.close()
with open(support.TESTFN, 'rb') as file:
with open(os_helper.TESTFN, 'rb') as file:
self.assertEqual(file.read(), self.d + d1 + d2)

@unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),
'asyncore.file_dispatcher required')
def test_dispatcher(self):
fd = os.open(support.TESTFN, os.O_RDONLY)
fd = os.open(os_helper.TESTFN, os.O_RDONLY)
data = []
class FileDispatcher(asyncore.file_dispatcher):
def handle_read(self):
Expand All @@ -415,16 +417,16 @@ def handle_read(self):

def test_resource_warning(self):
# Issue #11453
fd = os.open(support.TESTFN, os.O_RDONLY)
fd = os.open(os_helper.TESTFN, os.O_RDONLY)
f = asyncore.file_wrapper(fd)

os.close(fd)
with support.check_warnings(('', ResourceWarning)):
with warnings_helper.check_warnings(('', ResourceWarning)):
f = None
support.gc_collect()

def test_close_twice(self):
fd = os.open(support.TESTFN, os.O_RDONLY)
fd = os.open(os_helper.TESTFN, os.O_RDONLY)
f = asyncore.file_wrapper(fd)
os.close(fd)

Expand Down Expand Up @@ -804,10 +806,10 @@ class TestAPI_UseIPv6Sockets(BaseTestAPI):
class TestAPI_UseUnixSockets(BaseTestAPI):
if HAS_UNIX_SOCKETS:
family = socket.AF_UNIX
addr = support.TESTFN
addr = os_helper.TESTFN

def tearDown(self):
support.unlink(self.addr)
os_helper.unlink(self.addr)
BaseTestAPI.tearDown(self)

class TestAPI_UseIPv4Select(TestAPI_UseIPv4Sockets, unittest.TestCase):
Expand Down
15 changes: 8 additions & 7 deletions Lib/test/test_binascii.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@
import binascii
import array
import re
from test import support
from test.support import warnings_helper


# Note: "*_hex" functions are aliases for "(un)hexlify"
b2a_functions = ['b2a_base64', 'b2a_hex', 'b2a_hqx', 'b2a_qp', 'b2a_uu',
Expand Down Expand Up @@ -37,7 +38,7 @@ def test_functions(self):
self.assertTrue(hasattr(getattr(binascii, name), '__call__'))
self.assertRaises(TypeError, getattr(binascii, name))

@support.ignore_warnings(category=DeprecationWarning)
@warnings_helper.ignore_warnings(category=DeprecationWarning)
def test_returned_value(self):
# Limit to the minimum of all limits (b2a_uu)
MAX_ALL = 45
Expand Down Expand Up @@ -181,7 +182,7 @@ def test_uu(self):
with self.assertRaises(TypeError):
binascii.b2a_uu(b"", True)

@support.ignore_warnings(category=DeprecationWarning)
@warnings_helper.ignore_warnings(category=DeprecationWarning)
def test_crc_hqx(self):
crc = binascii.crc_hqx(self.type2test(b"Test the CRC-32 of"), 0)
crc = binascii.crc_hqx(self.type2test(b" this string."), crc)
Expand All @@ -201,7 +202,7 @@ def test_crc32(self):

self.assertRaises(TypeError, binascii.crc32)

@support.ignore_warnings(category=DeprecationWarning)
@warnings_helper.ignore_warnings(category=DeprecationWarning)
def test_hqx(self):
# Perform binhex4 style RLE-compression
# Then calculate the hexbin4 binary-to-ASCII translation
Expand All @@ -212,7 +213,7 @@ def test_hqx(self):
res = binascii.rledecode_hqx(b)
self.assertEqual(res, self.rawdata)

@support.ignore_warnings(category=DeprecationWarning)
@warnings_helper.ignore_warnings(category=DeprecationWarning)
def test_rle(self):
# test repetition with a repetition longer than the limit of 255
data = (b'a' * 100 + b'b' + b'c' * 300)
Expand Down Expand Up @@ -359,7 +360,7 @@ def test_qp(self):
self.assertEqual(b2a_qp(type2test(b'a.\n')), b'a.\n')
self.assertEqual(b2a_qp(type2test(b'.a')[:-1]), b'=2E')

@support.ignore_warnings(category=DeprecationWarning)
@warnings_helper.ignore_warnings(category=DeprecationWarning)
def test_empty_string(self):
# A test for SF bug #1022953. Make sure SystemError is not raised.
empty = self.type2test(b'')
Expand All @@ -384,7 +385,7 @@ def test_unicode_b2a(self):
# crc_hqx needs 2 arguments
self.assertRaises(TypeError, binascii.crc_hqx, "test", 0)

@support.ignore_warnings(category=DeprecationWarning)
@warnings_helper.ignore_warnings(category=DeprecationWarning)
def test_unicode_a2b(self):
# Unicode strings are accepted by a2b_* functions.
MAX_ALL = 45
Expand Down
7 changes: 4 additions & 3 deletions Lib/test/test_bisect.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import sys
import unittest
from test import support
from test.support import import_helper
from collections import UserList

py_bisect = support.import_fresh_module('bisect', blocked=['_bisect'])
c_bisect = support.import_fresh_module('bisect', fresh=['_bisect'])

py_bisect = import_helper.import_fresh_module('bisect', blocked=['_bisect'])
c_bisect = import_helper.import_fresh_module('bisect', fresh=['_bisect'])

class Range(object):
"""A trivial range()-like object that has an insert() method."""
Expand Down
6 changes: 3 additions & 3 deletions Lib/test/test_builtin.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,10 @@
from types import AsyncGeneratorType, FunctionType
from operator import neg
from test import support
from test.support import (
EnvironmentVarGuard, TESTFN, check_warnings, swap_attr, unlink,
maybe_get_event_loop_policy)
from test.support import (swap_attr, maybe_get_event_loop_policy)
from test.support.os_helper import (EnvironmentVarGuard, TESTFN, unlink)
from test.support.script_helper import assert_python_ok
from test.support.warnings_helper import check_warnings
from unittest.mock import MagicMock, patch
try:
import pty, signal
Expand Down
3 changes: 2 additions & 1 deletion Lib/test/test_concurrent_futures.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
from test import support
from test.support import import_helper
from test.support import threading_helper

# Skip tests if _multiprocessing wasn't built.
support.import_module('_multiprocessing')
import_helper.import_module('_multiprocessing')
# Skip tests if sem_open implementation is broken.
support.skip_if_broken_multiprocessing_synchronize()

Expand Down
7 changes: 4 additions & 3 deletions Lib/test/test_configparser.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import warnings

from test import support
from test.support import os_helper


class SortedDict(collections.UserDict):
Expand Down Expand Up @@ -1063,17 +1064,17 @@ def setUp(self):
cf.add_section(s)
for j in range(10):
cf.set(s, 'lovely_spam{}'.format(j), self.wonderful_spam)
with open(support.TESTFN, 'w') as f:
with open(os_helper.TESTFN, 'w') as f:
cf.write(f)

def tearDown(self):
os.unlink(support.TESTFN)
os.unlink(os_helper.TESTFN)

def test_dominating_multiline_values(self):
# We're reading from file because this is where the code changed
# during performance updates in Python 3.2
cf_from_file = self.newconfig()
with open(support.TESTFN) as f:
with open(os_helper.TESTFN) as f:
cf_from_file.read_file(f)
self.assertEqual(cf_from_file.get('section8', 'lovely_spam4'),
self.wonderful_spam.replace('\t\n', '\n'))
Expand Down
13 changes: 8 additions & 5 deletions Lib/test/test_coroutines.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
import unittest
import warnings
from test import support
from test.support import import_helper
from test.support import warnings_helper
from test.support.script_helper import assert_python_ok


Expand Down Expand Up @@ -2117,7 +2119,7 @@ class CoroAsyncIOCompatTest(unittest.TestCase):
def test_asyncio_1(self):
# asyncio cannot be imported when Python is compiled without thread
# support
asyncio = support.import_module('asyncio')
asyncio = import_helper.import_module('asyncio')

class MyException(Exception):
pass
Expand Down Expand Up @@ -2258,8 +2260,9 @@ async def corofn():
try:
warnings._warn_unawaited_coroutine = lambda coro: 1/0
with support.catch_unraisable_exception() as cm, \
support.check_warnings((r'coroutine .* was never awaited',
RuntimeWarning)):
warnings_helper.check_warnings(
(r'coroutine .* was never awaited',
RuntimeWarning)):
# only store repr() to avoid keeping the coroutine alive
coro = corofn()
coro_repr = repr(coro)
Expand All @@ -2272,8 +2275,8 @@ async def corofn():
self.assertEqual(cm.unraisable.exc_type, ZeroDivisionError)

del warnings._warn_unawaited_coroutine
with support.check_warnings((r'coroutine .* was never awaited',
RuntimeWarning)):
with warnings_helper.check_warnings(
(r'coroutine .* was never awaited', RuntimeWarning)):
corofn()
support.gc_collect()

Expand Down
3 changes: 2 additions & 1 deletion Lib/test/test_curses.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@
import tempfile
import unittest

from test.support import requires, import_module, verbose, SaveSignals
from test.support import requires, verbose, SaveSignals
from test.support.import_helper import import_module

# Optionally test curses module. This currently requires that the
# 'curses' resource be given on the regrtest command line using the -u
Expand Down
4 changes: 3 additions & 1 deletion Lib/test/test_datetime.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import unittest
import sys

from test.support import import_fresh_module, run_unittest
from test.support import run_unittest
from test.support.import_helper import import_fresh_module


TESTS = 'test.datetimetester'

Expand Down
5 changes: 3 additions & 2 deletions Lib/test/test_mmap.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from test.support import (TESTFN, import_module, unlink,
requires, _2G, _4G, gc_collect, cpython_only)
from test.support import (requires, _2G, _4G, gc_collect, cpython_only)
from test.support.import_helper import import_module
from test.support.os_helper import TESTFN, unlink
import unittest
import os
import re
Expand Down
3 changes: 2 additions & 1 deletion Lib/test/test_signal.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import time
import unittest
from test import support
from test.support import os_helper
from test.support.script_helper import assert_python_ok, spawn_python
try:
import _testcapi
Expand Down Expand Up @@ -154,7 +155,7 @@ def test_invalid_call(self):
signal.set_wakeup_fd(signal.SIGINT, False)

def test_invalid_fd(self):
fd = support.make_bad_fd()
fd = os_helper.make_bad_fd()
self.assertRaises((ValueError, OSError),
signal.set_wakeup_fd, fd)

Expand Down
Loading