| #!/usr/bin/env python | |
| import urlparse | |
| import urllib2 | |
| import BaseHTTPServer | |
| import unittest | |
| 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 | |
| class LoopbackHttpServer(BaseHTTPServer.HTTPServer): | |
| """HTTP server w/ a few modifications that make it useful for | |
| loopback testing purposes. | |
| """ | |
| def __init__(self, server_address, RequestHandlerClass): | |
| BaseHTTPServer.HTTPServer.__init__(self, | |
| server_address, | |
| RequestHandlerClass) | |
| # Set the timeout of our listening socket really low so | |
| # that we can stop the server easily. | |
| self.socket.settimeout(1.0) | |
| def get_request(self): | |
| """BaseHTTPServer method, overridden.""" | |
| request, client_address = self.socket.accept() | |
| # It's a loopback connection, so setting the timeout | |
| # really low shouldn't affect anything, but should make | |
| # deadlocks less likely to occur. | |
| request.settimeout(10.0) | |
| return (request, client_address) | |
| class LoopbackHttpServerThread(threading.Thread): | |
| """Stoppable thread that runs a loopback http server.""" | |
| def __init__(self, request_handler): | |
| threading.Thread.__init__(self) | |
| self._stop = False | |
| self.ready = threading.Event() | |
| request_handler.protocol_version = "HTTP/1.0" | |
| self.httpd = LoopbackHttpServer(('127.0.0.1', 0), | |
| request_handler) | |
| #print "Serving HTTP on %s port %s" % (self.httpd.server_name, | |
| # self.httpd.server_port) | |
| self.port = self.httpd.server_port | |
| def stop(self): | |
| """Stops the webserver if it's currently running.""" | |
| # Set the stop flag. | |
| self._stop = True | |
| self.join() | |
| def run(self): | |
| self.ready.set() | |
| while not self._stop: | |
| self.httpd.handle_request() | |
| # Authentication infrastructure | |
| class DigestAuthHandler: | |
| """Handler for performing digest authentication.""" | |
| def __init__(self): | |
| self._request_num = 0 | |
| self._nonces = [] | |
| self._users = {} | |
| self._realm_name = "Test Realm" | |
| self._qop = "auth" | |
| def set_qop(self, qop): | |
| self._qop = qop | |
| def set_users(self, users): | |
| assert isinstance(users, dict) | |
| self._users = users | |
| def set_realm(self, realm): | |
| self._realm_name = realm | |
| def _generate_nonce(self): | |
| self._request_num += 1 | |
| nonce = hashlib.md5(str(self._request_num)).hexdigest() | |
| self._nonces.append(nonce) | |
| return nonce | |
| def _create_auth_dict(self, auth_str): | |
| first_space_index = auth_str.find(" ") | |
| auth_str = auth_str[first_space_index+1:] | |
| parts = auth_str.split(",") | |
| auth_dict = {} | |
| for part in parts: | |
| name, value = part.split("=") | |
| name = name.strip() | |
| if value[0] == '"' and value[-1] == '"': | |
| value = value[1:-1] | |
| else: | |
| value = value.strip() | |
| auth_dict[name] = value | |
| return auth_dict | |
| def _validate_auth(self, auth_dict, password, method, uri): | |
| final_dict = {} | |
| final_dict.update(auth_dict) | |
| final_dict["password"] = password | |
| final_dict["method"] = method | |
| final_dict["uri"] = uri | |
| HA1_str = "%(username)s:%(realm)s:%(password)s" % final_dict | |
| HA1 = hashlib.md5(HA1_str).hexdigest() | |
| HA2_str = "%(method)s:%(uri)s" % final_dict | |
| HA2 = hashlib.md5(HA2_str).hexdigest() | |
| final_dict["HA1"] = HA1 | |
| final_dict["HA2"] = HA2 | |
| response_str = "%(HA1)s:%(nonce)s:%(nc)s:" \ | |
| "%(cnonce)s:%(qop)s:%(HA2)s" % final_dict | |
| response = hashlib.md5(response_str).hexdigest() | |
| return response == auth_dict["response"] | |
| def _return_auth_challenge(self, request_handler): | |
| request_handler.send_response(407, "Proxy Authentication Required") | |
| request_handler.send_header("Content-Type", "text/html") | |
| request_handler.send_header( | |
| 'Proxy-Authenticate', 'Digest realm="%s", ' | |
| 'qop="%s",' | |
| 'nonce="%s", ' % \ | |
| (self._realm_name, self._qop, self._generate_nonce())) | |
| # XXX: Not sure if we're supposed to add this next header or | |
| # not. | |
| #request_handler.send_header('Connection', 'close') | |
| request_handler.end_headers() | |
| request_handler.wfile.write("Proxy Authentication Required.") | |
| return False | |
| def handle_request(self, request_handler): | |
| """Performs digest authentication on the given HTTP request | |
| handler. Returns True if authentication was successful, False | |
| otherwise. | |
| If no users have been set, then digest auth is effectively | |
| disabled and this method will always return True. | |
| """ | |
| if len(self._users) == 0: | |
| return True | |
| if 'Proxy-Authorization' not in request_handler.headers: | |
| return self._return_auth_challenge(request_handler) | |
| else: | |
| auth_dict = self._create_auth_dict( | |
| request_handler.headers['Proxy-Authorization'] | |
| ) | |
| if auth_dict["username"] in self._users: | |
| password = self._users[ auth_dict["username"] ] | |
| else: | |
| return self._return_auth_challenge(request_handler) | |
| if not auth_dict.get("nonce") in self._nonces: | |
| return self._return_auth_challenge(request_handler) | |
| else: | |
| self._nonces.remove(auth_dict["nonce"]) | |
| auth_validated = False | |
| # MSIE uses short_path in its validation, but Python's | |
| # urllib2 uses the full path, so we're going to see if | |
| # either of them works here. | |
| for path in [request_handler.path, request_handler.short_path]: | |
| if self._validate_auth(auth_dict, | |
| password, | |
| request_handler.command, | |
| path): | |
| auth_validated = True | |
| if not auth_validated: | |
| return self._return_auth_challenge(request_handler) | |
| return True | |
| # Proxy test infrastructure | |
| class FakeProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler): | |
| """This is a 'fake proxy' that makes it look like the entire | |
| internet has gone down due to a sudden zombie invasion. It main | |
| utility is in providing us with authentication support for | |
| testing. | |
| """ | |
| def __init__(self, digest_auth_handler, *args, **kwargs): | |
| # This has to be set before calling our parent's __init__(), which will | |
| # try to call do_GET(). | |
| self.digest_auth_handler = digest_auth_handler | |
| BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args, **kwargs) | |
| def log_message(self, format, *args): | |
| # Uncomment the next line for debugging. | |
| #sys.stderr.write(format % args) | |
| pass | |
| def do_GET(self): | |
| (scm, netloc, path, params, query, fragment) = urlparse.urlparse( | |
| self.path, 'http') | |
| self.short_path = path | |
| if self.digest_auth_handler.handle_request(self): | |
| self.send_response(200, "OK") | |
| self.send_header("Content-Type", "text/html") | |
| self.end_headers() | |
| self.wfile.write("You've reached %s!<BR>" % self.path) | |
| self.wfile.write("Our apologies, but our server is down due to " | |
| "a sudden zombie invasion.") | |
| # Test cases | |
| class BaseTestCase(unittest.TestCase): | |
| def setUp(self): | |
| self._threads = test_support.threading_setup() | |
| def tearDown(self): | |
| test_support.threading_cleanup(*self._threads) | |
| class ProxyAuthTests(BaseTestCase): | |
| URL = "http://localhost" | |
| USER = "tester" | |
| PASSWD = "test123" | |
| REALM = "TestRealm" | |
| def setUp(self): | |
| super(ProxyAuthTests, self).setUp() | |
| self.digest_auth_handler = DigestAuthHandler() | |
| self.digest_auth_handler.set_users({self.USER: self.PASSWD}) | |
| self.digest_auth_handler.set_realm(self.REALM) | |
| def create_fake_proxy_handler(*args, **kwargs): | |
| return FakeProxyHandler(self.digest_auth_handler, *args, **kwargs) | |
| self.server = LoopbackHttpServerThread(create_fake_proxy_handler) | |
| self.server.start() | |
| self.server.ready.wait() | |
| proxy_url = "http://127.0.0.1:%d" % self.server.port | |
| handler = urllib2.ProxyHandler({"http" : proxy_url}) | |
| self.proxy_digest_handler = urllib2.ProxyDigestAuthHandler() | |
| self.opener = urllib2.build_opener(handler, self.proxy_digest_handler) | |
| def tearDown(self): | |
| self.server.stop() | |
| super(ProxyAuthTests, self).tearDown() | |
| def test_proxy_with_bad_password_raises_httperror(self): | |
| self.proxy_digest_handler.add_password(self.REALM, self.URL, | |
| self.USER, self.PASSWD+"bad") | |
| self.digest_auth_handler.set_qop("auth") | |
| self.assertRaises(urllib2.HTTPError, | |
| self.opener.open, | |
| self.URL) | |
| def test_proxy_with_no_password_raises_httperror(self): | |
| self.digest_auth_handler.set_qop("auth") | |
| self.assertRaises(urllib2.HTTPError, | |
| self.opener.open, | |
| self.URL) | |
| def test_proxy_qop_auth_works(self): | |
| self.proxy_digest_handler.add_password(self.REALM, self.URL, | |
| self.USER, self.PASSWD) | |
| self.digest_auth_handler.set_qop("auth") | |
| result = self.opener.open(self.URL) | |
| while result.read(): | |
| pass | |
| result.close() | |
| def test_proxy_qop_auth_int_works_or_throws_urlerror(self): | |
| self.proxy_digest_handler.add_password(self.REALM, self.URL, | |
| self.USER, self.PASSWD) | |
| self.digest_auth_handler.set_qop("auth-int") | |
| try: | |
| result = self.opener.open(self.URL) | |
| except urllib2.URLError: | |
| # It's okay if we don't support auth-int, but we certainly | |
| # shouldn't receive any kind of exception here other than | |
| # a URLError. | |
| result = None | |
| if result: | |
| while result.read(): | |
| 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(BaseTestCase): | |
| """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 = '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.assertEqual(data, expected_response) | |
| self.assertEqual(handler.requests, ['/', '/somewhere_else']) | |
| finally: | |
| self.server.stop() | |
| def test_404(self): | |
| expected_response = 'Bad bad bad...' | |
| handler = self.start_server([(404, [], expected_response)]) | |
| try: | |
| try: | |
| urllib2.urlopen('http://localhost:%s/weeble' % handler.port) | |
| except urllib2.URLError, f: | |
| pass | |
| else: | |
| self.fail('404 should raise URLError') | |
| data = f.read() | |
| f.close() | |
| self.assertEqual(data, expected_response) | |
| self.assertEqual(handler.requests, ['/weeble']) | |
| finally: | |
| self.server.stop() | |
| def test_200(self): | |
| expected_response = '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.assertEqual(data, expected_response) | |
| self.assertEqual(handler.requests, ['/bizarre']) | |
| finally: | |
| self.server.stop() | |
| def test_200_with_parameters(self): | |
| expected_response = 'pycon 2008...' | |
| handler = self.start_server([(200, [], expected_response)]) | |
| try: | |
| f = urllib2.urlopen('http://localhost:%s/bizarre' % handler.port, 'get=with_feeling') | |
| data = f.read() | |
| f.close() | |
| self.assertEqual(data, expected_response) | |
| self.assertEqual(handler.requests, ['/bizarre', 'get=with_feeling']) | |
| finally: | |
| self.server.stop() | |
| def test_sending_headers(self): | |
| handler = self.start_server([(200, [], "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, [], "we don't care")]) | |
| try: | |
| open_url = urllib2.urlopen("http://localhost:%s" % handler.port) | |
| for attr in ("read", "close", "info", "geturl"): | |
| self.assertTrue(hasattr(open_url, attr), "object returned from " | |
| "urlopen lacks the %s attribute" % attr) | |
| try: | |
| self.assertTrue(open_url.read(), "calling 'read' failed") | |
| finally: | |
| open_url.close() | |
| finally: | |
| self.server.stop() | |
| def test_info(self): | |
| handler = self.start_server([(200, [], "we don't care")]) | |
| try: | |
| open_url = urllib2.urlopen("http://localhost:%s" % handler.port) | |
| info_obj = open_url.info() | |
| self.assertIsInstance(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, [], "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, | |
| # Given that both VeriSign and various ISPs have in | |
| # the past or are presently hijacking various invalid | |
| # domain name requests in an attempt to boost traffic | |
| # to their own sites, finding a domain name to use | |
| # for this test is difficult. RFC2606 leads one to | |
| # believe that '.invalid' should work, but experience | |
| # seemed to indicate otherwise. Single character | |
| # TLDs are likely to remain invalid, so this seems to | |
| # be the best choice. The trailing '.' prevents a | |
| # related problem: The normal DNS resolver appends | |
| # the domain names from the search path if there is | |
| # no '.' the end and, and if one of those domains | |
| # implements a '*' rule a result is returned. | |
| # However, none of this will prevent the test from | |
| # failing if the ISP hijacks all invalid domain | |
| # requests. The real solution would be to be able to | |
| # parameterize the framework with a mock resolver. | |
| urllib2.urlopen, "http://sadflkjsasf.i.nvali.d./") | |
| def test_iteration(self): | |
| expected_response = "pycon 2008..." | |
| handler = self.start_server([(200, [], expected_response)]) | |
| try: | |
| data = urllib2.urlopen("http://localhost:%s" % handler.port) | |
| for line in data: | |
| self.assertEqual(line, expected_response) | |
| finally: | |
| self.server.stop() | |
| def ztest_line_iteration(self): | |
| lines = ["We\n", "got\n", "here\n", "verylong " * 8192 + "\n"] | |
| expected_response = "".join(lines) | |
| handler = self.start_server([(200, [], expected_response)]) | |
| try: | |
| data = urllib2.urlopen("http://localhost:%s" % handler.port) | |
| for index, line in enumerate(data): | |
| self.assertEqual(line, lines[index], | |
| "Fetched line number %s doesn't match expected:\n" | |
| " Expected length was %s, got %s" % | |
| (index, len(lines[index]), len(line))) | |
| finally: | |
| self.server.stop() | |
| self.assertEqual(index + 1, len(lines)) | |
| def test_main(): | |
| # We will NOT depend on the network resource flag | |
| # (Lib/test/regrtest.py -u network) since all tests here are only | |
| # localhost. However, if this is a bad rationale, then uncomment | |
| # the next line. | |
| #test_support.requires("network") | |
| test_support.run_unittest(ProxyAuthTests, TestUrlopen) | |
| if __name__ == "__main__": | |
| test_main() |