bpo-40275: Adding threading_helper submodule in test.support (GH-20263)
diff --git a/Lib/test/test_imaplib.py b/Lib/test/test_imaplib.py
index d1e3550..0fcc1fb 100644
--- a/Lib/test/test_imaplib.py
+++ b/Lib/test/test_imaplib.py
@@ -10,9 +10,10 @@
import threading
import socket
-from test.support import (reap_threads, verbose,
+from test.support import (verbose,
run_with_tz, run_with_locale, cpython_only)
from test.support import hashlib_helper
+from test.support import threading_helper
import unittest
from unittest import mock
from datetime import datetime, timezone, timedelta
@@ -252,7 +253,7 @@
# cleanup the server
self.server.shutdown()
self.server.server_close()
- support.join_thread(self.thread)
+ threading_helper.join_thread(self.thread)
# Explicitly clear the attribute to prevent dangling thread
self.thread = None
@@ -641,13 +642,13 @@
finally:
client.logout()
- @reap_threads
+ @threading_helper.reap_threads
def test_connect(self):
with self.reaped_server(SimpleIMAPHandler) as server:
client = self.imap_class(*server.server_address)
client.shutdown()
- @reap_threads
+ @threading_helper.reap_threads
def test_bracket_flags(self):
# This violates RFC 3501, which disallows ']' characters in tag names,
@@ -696,7 +697,7 @@
typ, [data] = client.response('PERMANENTFLAGS')
self.assertIn(b'[test]', data)
- @reap_threads
+ @threading_helper.reap_threads
def test_issue5949(self):
class EOFHandler(socketserver.StreamRequestHandler):
@@ -708,7 +709,7 @@
self.assertRaises(imaplib.IMAP4.abort,
self.imap_class, *server.server_address)
- @reap_threads
+ @threading_helper.reap_threads
def test_line_termination(self):
class BadNewlineHandler(SimpleIMAPHandler):
@@ -732,7 +733,7 @@
self.server.response = yield
self._send_tagged(tag, 'OK', 'FAKEAUTH successful')
- @reap_threads
+ @threading_helper.reap_threads
def test_enable_raises_error_if_not_AUTH(self):
with self.reaped_pair(self.UTF8Server) as (server, client):
self.assertFalse(client.utf8_enabled)
@@ -741,14 +742,14 @@
# XXX Also need a test that enable after SELECT raises an error.
- @reap_threads
+ @threading_helper.reap_threads
def test_enable_raises_error_if_no_capability(self):
class NoEnableServer(self.UTF8Server):
capabilities = 'AUTH'
with self.reaped_pair(NoEnableServer) as (server, client):
self.assertRaises(imaplib.IMAP4.error, client.enable, 'foo')
- @reap_threads
+ @threading_helper.reap_threads
def test_enable_UTF8_raises_error_if_not_supported(self):
class NonUTF8Server(SimpleIMAPHandler):
pass
@@ -759,7 +760,7 @@
client.enable('UTF8=ACCEPT')
pass
- @reap_threads
+ @threading_helper.reap_threads
def test_enable_UTF8_True_append(self):
class UTF8AppendServer(self.UTF8Server):
@@ -789,7 +790,7 @@
# XXX also need a test that makes sure that the Literal and Untagged_status
# regexes uses unicode in UTF8 mode instead of the default ASCII.
- @reap_threads
+ @threading_helper.reap_threads
def test_search_disallows_charset_in_utf8_mode(self):
with self.reaped_pair(self.UTF8Server) as (server, client):
typ, _ = client.authenticate('MYAUTH', lambda x: b'fake')
@@ -799,7 +800,7 @@
self.assertTrue(client.utf8_enabled)
self.assertRaises(imaplib.IMAP4.error, client.search, 'foo', 'bar')
- @reap_threads
+ @threading_helper.reap_threads
def test_bad_auth_name(self):
class MyServer(SimpleIMAPHandler):
@@ -812,7 +813,7 @@
with self.assertRaises(imaplib.IMAP4.error):
client.authenticate('METHOD', lambda: 1)
- @reap_threads
+ @threading_helper.reap_threads
def test_invalid_authentication(self):
class MyServer(SimpleIMAPHandler):
@@ -826,7 +827,7 @@
with self.assertRaises(imaplib.IMAP4.error):
code, data = client.authenticate('MYAUTH', lambda x: b'fake')
- @reap_threads
+ @threading_helper.reap_threads
def test_valid_authentication(self):
class MyServer(SimpleIMAPHandler):
@@ -848,7 +849,7 @@
self.assertEqual(server.response,
b'ZmFrZQ==\r\n') # b64 encoded 'fake'
- @reap_threads
+ @threading_helper.reap_threads
@hashlib_helper.requires_hashdigest('md5')
def test_login_cram_md5(self):
@@ -877,7 +878,7 @@
self.assertEqual(ret, "OK")
- @reap_threads
+ @threading_helper.reap_threads
def test_aborted_authentication(self):
class MyServer(SimpleIMAPHandler):
@@ -906,14 +907,14 @@
self.assertRaises(imaplib.IMAP4.error,
self.imap_class, *server.server_address)
- @reap_threads
+ @threading_helper.reap_threads
def test_simple_with_statement(self):
# simplest call
with self.reaped_server(SimpleIMAPHandler) as server:
with self.imap_class(*server.server_address):
pass
- @reap_threads
+ @threading_helper.reap_threads
def test_with_statement(self):
with self.reaped_server(SimpleIMAPHandler) as server:
with self.imap_class(*server.server_address) as imap:
@@ -921,7 +922,7 @@
self.assertEqual(server.logged, 'user')
self.assertIsNone(server.logged)
- @reap_threads
+ @threading_helper.reap_threads
def test_with_statement_logout(self):
# what happens if already logout in the block?
with self.reaped_server(SimpleIMAPHandler) as server:
@@ -938,7 +939,7 @@
server_class = SecureTCPServer
imap_class = IMAP4_SSL
- @reap_threads
+ @threading_helper.reap_threads
def test_ssl_verified(self):
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.load_verify_locations(CAFILE)