diff options
Diffstat (limited to 'Lib')
| -rwxr-xr-x | Lib/test/regrtest.py | 1 | ||||
| -rw-r--r-- | Lib/test/test_sax.py | 13 | ||||
| -rw-r--r-- | Lib/test/test_socket.py | 10 | ||||
| -rw-r--r-- | Lib/test/test_sqlite.py | 7 | ||||
| -rw-r--r-- | Lib/test/test_telnetlib.py | 14 | ||||
| -rw-r--r-- | Lib/test/test_threading.py | 8 | ||||
| -rw-r--r-- | Lib/test/test_urllib2_localnet.py | 200 | ||||
| -rw-r--r-- | Lib/test/test_urllib2net.py | 98 | ||||
| -rw-r--r-- | Lib/test/test_xmlrpc.py | 12 | ||||
| -rw-r--r-- | Lib/threading.py | 10 |
10 files changed, 254 insertions, 119 deletions
diff --git a/Lib/test/regrtest.py b/Lib/test/regrtest.py index 9b8b572..000f241 100755 --- a/Lib/test/regrtest.py +++ b/Lib/test/regrtest.py @@ -529,7 +529,6 @@ NOTTESTS = { 'test_support', 'test_future1', 'test_future2', - 'test_future3', } def findtests(testdir=None, stdtests=STDTESTS, nottests=NOTTESTS): diff --git a/Lib/test/test_sax.py b/Lib/test/test_sax.py index a411bfd..eb79581 100644 --- a/Lib/test/test_sax.py +++ b/Lib/test/test_sax.py @@ -446,7 +446,8 @@ class ExpatReaderTest(XmlTestBase): # ===== InputSource support - def test_expat_inpsource_filename(self): + def XXXtest_expat_inpsource_filename(self): + # FIXME: test blocks indefinitely parser = create_parser() result = StringIO() xmlgen = XMLGenerator(result) @@ -456,7 +457,8 @@ class ExpatReaderTest(XmlTestBase): self.assertEquals(result.getvalue(), xml_test_out) - def test_expat_inpsource_sysid(self): + def XXXtest_expat_inpsource_sysid(self): + # FIXME: test blocks indefinitely parser = create_parser() result = StringIO() xmlgen = XMLGenerator(result) @@ -529,7 +531,8 @@ class ExpatReaderTest(XmlTestBase): self.assertEquals(parser.getPublicId(), None) self.assertEquals(parser.getLineNumber(), 1) - def test_expat_locator_withinfo(self): + def XXXtest_expat_locator_withinfo(self): + # FIXME: test blocks indefinitely result = StringIO() xmlgen = XMLGenerator(result) parser = create_parser() @@ -684,7 +687,7 @@ class XmlReaderTest(XmlTestBase): self.assertRaises(SAXParseException, parser.parse, sio) -def unittest_main(): +def test_main(): run_unittest(MakeParserTest, SaxutilsTest, XmlgenTest, @@ -693,4 +696,4 @@ def unittest_main(): XmlReaderTest) if __name__ == "__main__": - unittest_main() + test_main() diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index d3b870f..2bec373 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -5,8 +5,9 @@ from test import test_support import socket import select -import time import thread, threading +import time +import traceback import Queue import sys import os @@ -1016,10 +1017,13 @@ class TCPTimeoutTest(SocketTCPTest): except Alarm: pass except: - self.fail("caught other exception instead of Alarm") + self.fail("caught other exception instead of Alarm:" + " %s(%s):\n%s" % + (sys.exc_info()[:2] + (traceback.format_exc(),))) else: self.fail("nothing caught") - signal.alarm(0) # shut off alarm + finally: + signal.alarm(0) # shut off alarm except Alarm: self.fail("got Alarm in wrong place") finally: diff --git a/Lib/test/test_sqlite.py b/Lib/test/test_sqlite.py index c1523e1..945dd51 100644 --- a/Lib/test/test_sqlite.py +++ b/Lib/test/test_sqlite.py @@ -5,12 +5,13 @@ try: except ImportError: raise TestSkipped('no sqlite available') from sqlite3.test import (dbapi, types, userfunctions, - factory, transactions, hooks, regression) + factory, transactions, hooks, regression, + dump) def test_main(): run_unittest(dbapi.suite(), types.suite(), userfunctions.suite(), - factory.suite(), transactions.suite(), hooks.suite(), - regression.suite()) + factory.suite(), transactions.suite(), + hooks.suite(), regression.suite(), dump.suite()) if __name__ == "__main__": test_main() diff --git a/Lib/test/test_telnetlib.py b/Lib/test/test_telnetlib.py index 3a53ad9..8eee666 100644 --- a/Lib/test/test_telnetlib.py +++ b/Lib/test/test_telnetlib.py @@ -6,12 +6,14 @@ import time from unittest import TestCase from test import test_support +PORT = 9091 def server(evt): serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) serv.settimeout(3) serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - serv.bind(("", 9091)) + global PORT + PORT = test_support.bind_port(serv, "", PORT) serv.listen(5) evt.set() try: @@ -36,24 +38,24 @@ class GeneralTests(TestCase): def testBasic(self): # connects - telnet = telnetlib.Telnet("localhost", 9091) + telnet = telnetlib.Telnet("localhost", PORT) telnet.sock.close() def testTimeoutDefault(self): # default - telnet = telnetlib.Telnet("localhost", 9091) + telnet = telnetlib.Telnet("localhost", PORT) self.assertTrue(telnet.sock.gettimeout() is None) telnet.sock.close() def testTimeoutValue(self): # a value - telnet = telnetlib.Telnet("localhost", 9091, timeout=30) + telnet = telnetlib.Telnet("localhost", PORT, timeout=30) self.assertEqual(telnet.sock.gettimeout(), 30) telnet.sock.close() def testTimeoutDifferentOrder(self): telnet = telnetlib.Telnet(timeout=30) - telnet.open("localhost", 9091) + telnet.open("localhost", PORT) self.assertEqual(telnet.sock.gettimeout(), 30) telnet.sock.close() @@ -62,7 +64,7 @@ class GeneralTests(TestCase): previous = socket.getdefaulttimeout() socket.setdefaulttimeout(30) try: - telnet = telnetlib.Telnet("localhost", 9091, timeout=None) + telnet = telnetlib.Telnet("localhost", PORT, timeout=None) finally: socket.setdefaulttimeout(previous) self.assertEqual(telnet.sock.gettimeout(), 30) diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py index 67d9ed9..ee0b197 100644 --- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -271,13 +271,17 @@ class ThreadTests(unittest.TestCase): weak_cyclic_object = weakref.ref(cyclic_object) cyclic_object.thread.join() del cyclic_object - self.assertEquals(None, weak_cyclic_object()) + self.assertEquals(None, weak_cyclic_object(), + msg=('%d references still around' % + sys.getrefcount(weak_cyclic_object()))) raising_cyclic_object = RunSelfFunction(should_raise=True) weak_raising_cyclic_object = weakref.ref(raising_cyclic_object) raising_cyclic_object.thread.join() del raising_cyclic_object - self.assertEquals(None, weak_raising_cyclic_object()) + self.assertEquals(None, weak_raising_cyclic_object(), + msg=('%d references still around' % + sys.getrefcount(weak_raising_cyclic_object()))) class ThreadingExceptionTests(unittest.TestCase): diff --git a/Lib/test/test_urllib2_localnet.py b/Lib/test/test_urllib2_localnet.py index 663330b..0815658 100644 --- a/Lib/test/test_urllib2_localnet.py +++ b/Lib/test/test_urllib2_localnet.py @@ -1,5 +1,6 @@ #!/usr/bin/env python +import mimetools import threading import urlparse import urllib2 @@ -217,7 +218,7 @@ class FakeProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler): # Test cases class ProxyAuthTests(unittest.TestCase): - URL = "http://www.foo.com" + URL = "http://localhost" USER = "tester" PASSWD = "test123" @@ -279,6 +280,202 @@ class ProxyAuthTests(unittest.TestCase): pass result.close() + +def GetRequestHandler(responses): + + class FakeHTTPRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): + + server_version = "TestHTTP/" + requests = [] + headers_received = [] + port = 80 + + def do_GET(self): + body = self.send_head() + if body: + self.wfile.write(body) + + def do_POST(self): + content_length = self.headers['Content-Length'] + post_data = self.rfile.read(int(content_length)) + self.do_GET() + self.requests.append(post_data) + + def send_head(self): + FakeHTTPRequestHandler.headers_received = self.headers + self.requests.append(self.path) + response_code, headers, body = responses.pop(0) + + self.send_response(response_code) + + for (header, value) in headers: + self.send_header(header, value % self.port) + if body: + self.send_header('Content-type', 'text/plain') + self.end_headers() + return body + self.end_headers() + + def log_message(self, *args): + pass + + + return FakeHTTPRequestHandler + + +class TestUrlopen(unittest.TestCase): + """Tests urllib2.urlopen using the network. + + These tests are not exhaustive. Assuming that testing using files does a + good job overall of some of the basic interface features. There are no + tests exercising the optional 'data' and 'proxies' arguments. No tests + for transparent redirection have been written. + """ + + def start_server(self, responses): + handler = GetRequestHandler(responses) + + self.server = LoopbackHttpServerThread(handler) + self.server.start() + self.server.ready.wait() + port = self.server.port + handler.port = port + return handler + + + def test_redirection(self): + expected_response = b'We got here...' + responses = [ + (302, [('Location', 'http://localhost:%s/somewhere_else')], ''), + (200, [], expected_response) + ] + + handler = self.start_server(responses) + + try: + f = urllib2.urlopen('http://localhost:%s/' % handler.port) + data = f.read() + f.close() + + self.assertEquals(data, expected_response) + self.assertEquals(handler.requests, ['/', '/somewhere_else']) + finally: + self.server.stop() + + + def test_404(self): + expected_response = b'Bad bad bad...' + handler = self.start_server([(404, [], expected_response)]) + + try: + try: + urllib2.urlopen('http://localhost:%s/weeble' % handler.port) + except urllib2.URLError as f: + data = f.read() + f.close() + else: + self.fail('404 should raise URLError') + + self.assertEquals(data, expected_response) + self.assertEquals(handler.requests, ['/weeble']) + finally: + self.server.stop() + + + def test_200(self): + expected_response = b'pycon 2008...' + handler = self.start_server([(200, [], expected_response)]) + + try: + f = urllib2.urlopen('http://localhost:%s/bizarre' % handler.port) + data = f.read() + f.close() + + self.assertEquals(data, expected_response) + self.assertEquals(handler.requests, ['/bizarre']) + finally: + self.server.stop() + + def test_200_with_parameters(self): + expected_response = b'pycon 2008...' + handler = self.start_server([(200, [], expected_response)]) + + try: + f = urllib2.urlopen('http://localhost:%s/bizarre' % handler.port, b'get=with_feeling') + data = f.read() + f.close() + + self.assertEquals(data, expected_response) + self.assertEquals(handler.requests, ['/bizarre', b'get=with_feeling']) + finally: + self.server.stop() + + + def test_sending_headers(self): + handler = self.start_server([(200, [], b"we don't care")]) + + try: + req = urllib2.Request("http://localhost:%s/" % handler.port, + headers={'Range': 'bytes=20-39'}) + urllib2.urlopen(req) + self.assertEqual(handler.headers_received['Range'], 'bytes=20-39') + finally: + self.server.stop() + + def test_basic(self): + handler = self.start_server([(200, [], b"we don't care")]) + + try: + open_url = urllib2.urlopen("http://localhost:%s" % handler.port) + for attr in ("read", "close", "info", "geturl"): + self.assert_(hasattr(open_url, attr), "object returned from " + "urlopen lacks the %s attribute" % attr) + try: + self.assert_(open_url.read(), "calling 'read' failed") + finally: + open_url.close() + finally: + self.server.stop() + + def test_info(self): + handler = self.start_server([(200, [], b"we don't care")]) + + try: + open_url = urllib2.urlopen("http://localhost:%s" % handler.port) + info_obj = open_url.info() + self.assert_(isinstance(info_obj, mimetools.Message), + "object returned by 'info' is not an instance of " + "mimetools.Message") + self.assertEqual(info_obj.getsubtype(), "plain") + finally: + self.server.stop() + + def test_geturl(self): + # Make sure same URL as opened is returned by geturl. + handler = self.start_server([(200, [], b"we don't care")]) + + try: + open_url = urllib2.urlopen("http://localhost:%s" % handler.port) + url = open_url.geturl() + self.assertEqual(url, "http://localhost:%s" % handler.port) + finally: + self.server.stop() + + + def test_bad_address(self): + # Make sure proper exception is raised when connecting to a bogus + # address. + self.assertRaises(IOError, + # SF patch 809915: In Sep 2003, VeriSign started + # highjacking invalid .com and .net addresses to + # boost traffic to their own site. This test + # started failing then. One hopes the .invalid + # domain will be spared to serve its defined + # purpose. + # urllib2.urlopen, "http://www.sadflkjsasadf.com/") + urllib2.urlopen, "http://www.python.invalid./") + + def test_main(): # We will NOT depend on the network resource flag # (Lib/test/regrtest.py -u network) since all tests here are only @@ -287,6 +484,7 @@ def test_main(): #test_support.requires("network") test_support.run_unittest(ProxyAuthTests) + test_support.run_unittest(TestUrlopen) if __name__ == "__main__": test_main() diff --git a/Lib/test/test_urllib2net.py b/Lib/test/test_urllib2net.py index dc3d884..75df9fd 100644 --- a/Lib/test/test_urllib2net.py +++ b/Lib/test/test_urllib2net.py @@ -25,20 +25,6 @@ def _urlopen_with_retry(host, *args, **kwargs): raise last_exc -class URLTimeoutTest(unittest.TestCase): - - TIMEOUT = 10.0 - - def setUp(self): - socket.setdefaulttimeout(self.TIMEOUT) - - def tearDown(self): - socket.setdefaulttimeout(None) - - def testURLread(self): - f = _urlopen_with_retry("http://www.python.org/") - x = f.read() - class AuthTests(unittest.TestCase): """Tests urllib2 authentication features.""" @@ -98,68 +84,6 @@ class CloseSocketTest(unittest.TestCase): response.close() self.assert_(fileobject.closed) -class urlopenNetworkTests(unittest.TestCase): - """Tests urllib2.urlopen using the network. - - These tests are not exhaustive. Assuming that testing using files does a - good job overall of some of the basic interface features. There are no - tests exercising the optional 'data' and 'proxies' arguments. No tests - for transparent redirection have been written. - - setUp is not used for always constructing a connection to - http://www.python.org/ since there a few tests that don't use that address - and making a connection is expensive enough to warrant minimizing unneeded - connections. - - """ - - def test_basic(self): - # Simple test expected to pass. - open_url = _urlopen_with_retry("http://www.python.org/") - for attr in ("read", "close", "info", "geturl"): - self.assert_(hasattr(open_url, attr), "object returned from " - "urlopen lacks the %s attribute" % attr) - try: - self.assert_(open_url.read(), "calling 'read' failed") - finally: - open_url.close() - - def test_info(self): - # Test 'info'. - open_url = _urlopen_with_retry("http://www.python.org/") - try: - info_obj = open_url.info() - finally: - open_url.close() - self.assert_(isinstance(info_obj, mimetools.Message), - "object returned by 'info' is not an instance of " - "mimetools.Message") - self.assertEqual(info_obj.getsubtype(), "html") - - def test_geturl(self): - # Make sure same URL as opened is returned by geturl. - URL = "http://www.python.org/" - open_url = _urlopen_with_retry(URL) - try: - gotten_url = open_url.geturl() - finally: - open_url.close() - self.assertEqual(gotten_url, URL) - - def test_bad_address(self): - # Make sure proper exception is raised when connecting to a bogus - # address. - self.assertRaises(IOError, - # SF patch 809915: In Sep 2003, VeriSign started - # highjacking invalid .com and .net addresses to - # boost traffic to their own site. This test - # started failing then. One hopes the .invalid - # domain will be spared to serve its defined - # purpose. - # urllib2.urlopen, "http://www.sadflkjsasadf.com/") - urllib2.urlopen, "http://www.python.invalid./") - - class OtherNetworkTests(unittest.TestCase): def setUp(self): if 0: # for debugging @@ -167,13 +91,6 @@ class OtherNetworkTests(unittest.TestCase): logger = logging.getLogger("test_urllib2net") logger.addHandler(logging.StreamHandler()) - def test_range (self): - req = urllib2.Request("http://www.python.org", - headers={'Range': 'bytes=20-39'}) - result = _urlopen_with_retry(req) - data = result.read() - self.assertEqual(len(data), 20) - # XXX The rest of these tests aren't very good -- they don't check much. # They do sometimes catch some major disasters, though. @@ -201,16 +118,6 @@ class OtherNetworkTests(unittest.TestCase): finally: os.remove(TESTFN) - def test_http(self): - urls = [ - 'http://www.espn.com/', # redirect - 'http://www.python.org/Spanish/Inquistion/', - ('http://www.python.org/cgi-bin/faqw.py', - 'query=pythonistas&querytype=simple&casefold=yes&req=search', None), - 'http://www.python.org/', - ] - self._test_urls(urls, self._extra_handlers()) - # XXX Following test depends on machine configurations that are internal # to CNRI. Need to set up a public server with the right authentication # configuration for test purposes. @@ -278,6 +185,7 @@ class OtherNetworkTests(unittest.TestCase): return handlers + class TimeoutTest(unittest.TestCase): def test_http_basic(self): u = _urlopen_with_retry("http://www.python.org") @@ -326,9 +234,7 @@ class TimeoutTest(unittest.TestCase): def test_main(): test_support.requires("network") - test_support.run_unittest(URLTimeoutTest, - urlopenNetworkTests, - AuthTests, + test_support.run_unittest(AuthTests, OtherNetworkTests, CloseSocketTest, TimeoutTest, diff --git a/Lib/test/test_xmlrpc.py b/Lib/test/test_xmlrpc.py index 77d4525..0691f23 100644 --- a/Lib/test/test_xmlrpc.py +++ b/Lib/test/test_xmlrpc.py @@ -238,9 +238,17 @@ def http_server(evt, numrequests): '''This is my function''' return True + class MyXMLRPCServer(SimpleXMLRPCServer.SimpleXMLRPCServer): + def get_request(self): + # Ensure the socket is always non-blocking. On Linux, socket + # attributes are not inherited like they are on *BSD and Windows. + s, port = self.socket.accept() + s.setblocking(True) + return s, port + try: - serv = SimpleXMLRPCServer.SimpleXMLRPCServer(("localhost", 0), - logRequests=False, bind_and_activate=False) + serv = MyXMLRPCServer(("localhost", 0), + logRequests=False, bind_and_activate=False) serv.server_bind() global PORT PORT = serv.socket.getsockname()[1] diff --git a/Lib/threading.py b/Lib/threading.py index d010b80..8661cde 100644 --- a/Lib/threading.py +++ b/Lib/threading.py @@ -391,6 +391,9 @@ class Thread(_Verbose): # shutdown and thus raises an exception about trying to perform some # operation on/with a NoneType __exc_info = _sys.exc_info + # Keep sys.exc_clear too to clear the exception just before + # allowing .join() to return. + #XXX __exc_clear = _sys.exc_clear def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, verbose=None): @@ -527,6 +530,13 @@ class Thread(_Verbose): else: if __debug__: self._note("%s._bootstrap(): normal return", self) + finally: + # Prevent a race in + # test_threading.test_no_refcycle_through_target when + # the exception keeps the target alive past when we + # assert that it's dead. + #XXX self.__exc_clear() + pass finally: with _active_limbo_lock: self._stop() |