import unittest
import sys
-import threading
import sqlite3 as sqlite
+try:
+ import threading
+except ImportError:
+ threading = None
class ModuleTests(unittest.TestCase):
def CheckAPILevel(self):
except TypeError:
pass
+@unittest.skipUnless(threading, 'This test requires threading.')
class ThreadTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
"""This test case provides support for checking forking and wait behavior.
-To test different wait behavior, overrise the wait_impl method.
+To test different wait behavior, override the wait_impl method.
We want fork1() semantics -- only the forking thread survives in the
child after a fork().
the same application, the present example should work just fine. DC
"""
-import os, sys, time, thread, unittest
+import os, sys, time, unittest
+import test.test_support as test_support
+thread = test_support.import_module('thread')
LONGSLEEP = 2
SHORTSLEEP = 0.5
import select
import os
import socket
-import threading
import sys
import time
from test.test_support import TESTFN, run_unittest, unlink
from StringIO import StringIO
+try:
+ import threading
+except ImportError:
+ threading = None
+
HOST = test_support.HOST
class dummysocket:
def tearDown(self):
asyncore.close_all()
+ @unittest.skipUnless(threading, 'Threading required for this test.')
@test_support.reap_threads
def test_send(self):
evt = threading.Event()
import os
import subprocess
import sys
-import threading
+
+try:
+ import threading
+except ImportError:
+ threading = None
bz2 = import_module('bz2')
from bz2 import BZ2File, BZ2Compressor, BZ2Decompressor
else:
self.fail("1 // 0 didn't raise an exception")
+ @unittest.skipUnless(threading, 'Threading required for this test.')
def testThreading(self):
# Using a BZ2File from several threads doesn't deadlock (issue #7205).
data = "1" * 2**20
import sys
import tempfile
import unittest
-import threading
from contextlib import * # Tests __all__
from test import test_support
+try:
+ import threading
+except ImportError:
+ threading = None
class ContextManagerTestCase(unittest.TestCase):
finally:
test_support.unlink(tfn)
+@unittest.skipUnless(threading, 'Threading required for this test.')
class LockContextTestCase(unittest.TestCase):
def boilerPlate(self, lock, locked):
import httplib
import sys
from test import test_support
-import threading
+threading = test_support.import_module('threading')
import time
import socket
import unittest
import unittest
import itertools
import time
-import threading
from array import array
from weakref import proxy
+try:
+ import threading
+except ImportError:
+ threading = None
from test import test_support
from test.test_support import TESTFN, run_unittest
self.assertTrue(f.subclass_closed)
+@unittest.skipUnless(threading, 'Threading required for this test.')
class FileThreadingTests(unittest.TestCase):
# These tests check the ability to call various methods of file objects
# (including close()) concurrently without crashing the Python interpreter.
import signal
import sys
import time
-import threading
from test.fork_wait import ForkWait
-from test.test_support import run_unittest, reap_children, get_attribute
+from test.test_support import run_unittest, reap_children, get_attribute, import_module
+threading = import_module('threading')
#Skip test if fork does not exist.
get_attribute(os, 'fork')
# environment
import ftplib
-import threading
import asyncore
import asynchat
import socket
from unittest import TestCase
from test import test_support
from test.test_support import HOST
+threading = test_support.import_module('threading')
# the dummy data returned by server over the data channel when
import urllib
import httplib
import tempfile
-import threading
import unittest
from test import test_support
+threading = test_support.import_module('threading')
class NoLogRequestHandler:
import sys
import time
import array
-import threading
import random
import unittest
import weakref
import codecs
import io # C implementation of io
import _pyio as pyio # Python implementation of io
+try:
+ import threading
+except ImportError:
+ threading = None
__metaclass__ = type
bytes = support.py3k_bytes
self.assertEquals(b"abcdefg", bufio.read())
+ @unittest.skipUnless(threading, 'Threading required for this test.')
def test_threads(self):
try:
# Write out many bytes with exactly the same number of 0's,
with self.open(support.TESTFN, "rb", buffering=0) as f:
self.assertEqual(f.read(), b"abc")
+ @unittest.skipUnless(threading, 'Threading required for this test.')
def test_threads(self):
try:
# Write out many bytes from many threads and test they were
with self.open(support.TESTFN, "w", errors="replace") as f:
self.assertEqual(f.errors, "replace")
-
+ @unittest.skipUnless(threading, 'Threading required for this test.')
def test_threads_write(self):
# Issue6750: concurrent writes could duplicate data
event = threading.Event()
from test.test_support import captured_stdout, run_with_locale, run_unittest,\
find_unused_port
import textwrap
-import threading
import unittest
import warnings
import weakref
-
+try:
+ import threading
+except ImportError:
+ threading = None
class BaseTest(unittest.TestCase):
self.server_close()
+@unittest.skipUnless(threading, 'Threading required for this test.')
class SocketHandlerTest(BaseTest):
"""Test for SocketHandler objects."""
def test_config13_failure(self):
self.assertRaises(StandardError, self.apply_config, self.config13)
+ @unittest.skipUnless(threading, 'listen() needs threading to work')
def setup_via_listener(self, text):
port = find_unused_port()
t = logging.config.listen(port)
# a real test suite
import poplib
-import threading
import asyncore
import asynchat
import socket
from unittest import TestCase
from test import test_support
from test.test_support import HOST
+threading = test_support.import_module('threading')
# the dummy data returned by server when LIST and RETR commands are issued
# Some simple queue module tests, plus some failure conditions
# to ensure the Queue locks remain stable.
import Queue
-import threading
import time
import unittest
from test import test_support
+threading = test_support.import_module('threading')
QUEUE_SIZE = 5
import asyncore
import email.utils
import socket
-import threading
import smtpd
import smtplib
import StringIO
import time
import select
-from unittest import TestCase
+import unittest
from test import test_support
+try:
+ import threading
+except ImportError:
+ threading = None
+
HOST = test_support.HOST
def server(evt, buf, serv):
serv.close()
evt.set()
-class GeneralTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class GeneralTests(unittest.TestCase):
def setUp(self):
self._threads = test_support.threading_setup()
# test server times out, causing the test to fail.
# Test behavior of smtpd.DebuggingServer
-class DebuggingServerTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class DebuggingServerTests(unittest.TestCase):
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.assertEqual(self.output.getvalue(), mexpect)
-class NonConnectingTests(TestCase):
+class NonConnectingTests(unittest.TestCase):
def testNotConnected(self):
# Test various operations on an unconnected SMTP object that
# test response of client to a non-successful HELO message
-class BadHELOServerTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class BadHELOServerTests(unittest.TestCase):
def setUp(self):
self.old_stdout = sys.stdout
# Test various SMTP & ESMTP commands/behaviors that require a simulated server
# (i.e., something with more features than DebuggingServer)
-class SMTPSimTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class SMTPSimTests(unittest.TestCase):
def setUp(self):
self._threads = test_support.threading_setup()
import errno
import socket
import select
-import thread, threading
import time
import traceback
import Queue
from weakref import proxy
import signal
+try:
+ import thread
+ import threading
+except ImportError:
+ thread = None
+ threading = None
+
HOST = test_support.HOST
MSG = 'Michael Gilfix was here\n'
s.ioctl(socket.SIO_KEEPALIVE_VALS, (1, 100, 100))
+@unittest.skipUnless(thread, 'Threading required for this test.')
class BasicTCPTest(SocketConnectedTest):
def __init__(self, methodName='runTest'):
self.serv_conn.send(MSG)
self.serv_conn.shutdown(2)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class BasicUDPTest(ThreadedUDPSocketTest):
def __init__(self, methodName='runTest'):
def _testRecvFromNegative(self):
self.cli.sendto(MSG, 0, (HOST, self.port))
+@unittest.skipUnless(thread, 'Threading required for this test.')
class TCPCloserTest(ThreadedTCPSocketTest):
def testClose(self):
self.cli.connect((HOST, self.port))
time.sleep(1.0)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class BasicSocketPairTest(SocketPairTest):
def __init__(self, methodName='runTest'):
msg = self.cli.recv(1024)
self.assertEqual(msg, MSG)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class NonBlockingTCPTests(ThreadedTCPSocketTest):
def __init__(self, methodName='runTest'):
time.sleep(0.1)
self.cli.send(MSG)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class FileObjectClassTestCase(SocketConnectedTest):
bufsize = -1 # Use default buffer size
lambda: socket.create_connection((HOST, port))
)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):
def __init__(self, methodName='runTest'):
self.cli = socket.create_connection((HOST, self.port), 30)
self.assertEqual(self.cli.gettimeout(), 30)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class NetworkConnectionBehaviourTest(SocketTCPTest, ThreadableTest):
def __init__(self, methodName='runTest'):
self.assertRaises(socket.error, s.bind, address)
+@unittest.skipUnless(thread, 'Threading required for this test.')
class BufferIOTest(SocketConnectedTest):
"""
Test the buffer versions of socket.recv() and socket.send().
import signal
import socket
import tempfile
-import threading
import unittest
import SocketServer
import test.test_support
from test.test_support import reap_children, reap_threads, verbose
+try:
+ import threading
+except ImportError:
+ threading = None
test.test_support.requires("network")
self.assertEquals(server.server_address, server.socket.getsockname())
return server
+ @unittest.skipUnless(threading, 'Threading required for this test.')
@reap_threads
def run_server(self, svrcls, hdlrbase, testfunc):
server = self.make_server(self.pickaddr(svrcls.address_family),
import socket
-import threading
import telnetlib
import time
import Queue
from unittest import TestCase
from test import test_support
+threading = test_support.import_module('threading')
HOST = test_support.HOST
EOF_sigil = object()
import unittest
import random
from test import test_support
-import thread
+thread = test_support.import_module('thread')
import time
import sys
import weakref
# complains several times about module random having no attribute
# randrange, and then Python hangs.
-import thread
import unittest
-from test.test_support import verbose, TestFailed
+from test.test_support import verbose, TestFailed, import_module
+thread = import_module('thread')
critical_section = thread.allocate_lock()
done = thread.allocate_lock()
NUM_THREADS = 20
FILES_PER_THREAD = 50
-import thread # If this fails, we can't test this module
-import threading
import tempfile
-from test.test_support import threading_setup, threading_cleanup, run_unittest
+from test.test_support import threading_setup, threading_cleanup, run_unittest, import_module
+threading = import_module('threading')
import unittest
import StringIO
from traceback import print_exc
import random
import re
import sys
-import threading
-import thread
+thread = test.test_support.import_module('thread')
+threading = test.test_support.import_module('threading')
import time
import unittest
import weakref
import unittest
from doctest import DocTestSuite
from test import test_support
-import threading
+threading = test_support.import_module('threading')
import weakref
import gc
"""PyUnit testing that threads honor our signal semantics"""
import unittest
-import thread
import signal
import os
import sys
-from test.test_support import run_unittest
+from test.test_support import run_unittest, import_module
+thread = import_module('thread')
if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':
raise unittest.SkipTest, "Can't test signal on %s" % sys.platform
#!/usr/bin/env python
-import threading
import urlparse
import urllib2
import BaseHTTPServer
import hashlib
from test import test_support
mimetools = test_support.import_module('mimetools', deprecated=True)
+threading = test_support.import_module('threading')
# Loopback http server infrastructure
Tests
-----
-- Issue #7449: Fix many tests to support Python compiled without thread support
+- Issue #7449: Fix many tests to support Python compiled without thread
+ support. Patches written by Jerry Seutter.
- Issue #8108: test_ftplib's non-blocking SSL server now has proper handling
of SSL shutdowns.
projects / python / commitdiff