Run test_urllib2_localnet tests using unittest.main().

Capture threads in the setUpModule and cleanup threads in the tearDownModule.
This commit is contained in:
Senthil Kumaran 2014-04-15 16:36:43 -04:00
parent 484f8a875b
commit d943fdee87
1 changed files with 17 additions and 7 deletions

View File

@ -5,8 +5,11 @@ import urllib.request
import http.server import http.server
import unittest import unittest
import hashlib import hashlib
from test import support from test import support
threading = support.import_module('threading') threading = support.import_module('threading')
try: try:
import ssl import ssl
except ImportError: except ImportError:
@ -57,14 +60,11 @@ class LoopbackHttpServerThread(threading.Thread):
request_handler.protocol_version = "HTTP/1.0" request_handler.protocol_version = "HTTP/1.0"
self.httpd = LoopbackHttpServer(("127.0.0.1", 0), self.httpd = LoopbackHttpServer(("127.0.0.1", 0),
request_handler) request_handler)
#print "Serving HTTP on %s port %s" % (self.httpd.server_name,
# self.httpd.server_port)
self.port = self.httpd.server_port self.port = self.httpd.server_port
def stop(self): def stop(self):
"""Stops the webserver if it's currently running.""" """Stops the webserver if it's currently running."""
# Set the stop flag.
self._stop_server = True self._stop_server = True
self.join() self.join()
@ -232,6 +232,7 @@ class FakeProxyHandler(http.server.BaseHTTPRequestHandler):
# Test cases # Test cases
@unittest.skipUnless(threading, "Threading required for this test.")
class ProxyAuthTests(unittest.TestCase): class ProxyAuthTests(unittest.TestCase):
URL = "http://localhost" URL = "http://localhost"
@ -343,6 +344,7 @@ def GetRequestHandler(responses):
return FakeHTTPRequestHandler return FakeHTTPRequestHandler
@unittest.skipUnless(threading, "Threading required for this test.")
class TestUrlopen(unittest.TestCase): class TestUrlopen(unittest.TestCase):
"""Tests urllib.request.urlopen using the network. """Tests urllib.request.urlopen using the network.
@ -590,9 +592,17 @@ class TestUrlopen(unittest.TestCase):
self.assertEqual(index + 1, len(lines)) self.assertEqual(index + 1, len(lines))
@support.reap_threads threads_key = None
def test_main():
support.run_unittest(ProxyAuthTests, TestUrlopen) def setUpModule():
# Store the threading_setup in a key and ensure that it is cleaned up
# in the tearDown
global threads_key
threads_key = support.threading_setup()
def tearDownModule():
if threads_key:
support.threading_cleanup(threads_key)
if __name__ == "__main__": if __name__ == "__main__":
test_main() unittest.main()