diff -r ffa851df0825 -r 2fb8b9db1c86 symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_urllib2_localnet.py
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_urllib2_localnet.py	Fri Jul 31 15:01:17 2009 +0100
@@ -0,0 +1,491 @@
+#!/usr/bin/env python
+
+import mimetools
+import threading
+import urlparse
+import urllib2
+import BaseHTTPServer
+import unittest
+import hashlib
+from test import test_support
+
+# 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 not request_handler.headers.has_key('Proxy-Authorization'):
+            return self._return_auth_challenge(request_handler)
+        else:
+            auth_dict = self._create_auth_dict(
+                request_handler.headers['Proxy-Authorization']
+                )
+            if self._users.has_key(auth_dict["username"]):
+                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.
+    """
+
+    digest_auth_handler = DigestAuthHandler()
+
+    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 ProxyAuthTests(unittest.TestCase):
+    URL = "http://localhost"
+
+    USER = "tester"
+    PASSWD = "test123"
+    REALM = "TestRealm"
+
+    def setUp(self):
+        FakeProxyHandler.digest_auth_handler.set_users({
+            self.USER : self.PASSWD
+            })
+        FakeProxyHandler.digest_auth_handler.set_realm(self.REALM)
+
+        self.server = LoopbackHttpServerThread(FakeProxyHandler)
+        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._digest_auth_handler = urllib2.ProxyDigestAuthHandler()
+        self.opener = urllib2.build_opener(handler, self._digest_auth_handler)
+
+    def tearDown(self):
+        self.server.stop()
+
+    def test_proxy_with_bad_password_raises_httperror(self):
+        self._digest_auth_handler.add_password(self.REALM, self.URL,
+                                               self.USER, self.PASSWD+"bad")
+        FakeProxyHandler.digest_auth_handler.set_qop("auth")
+        self.assertRaises(urllib2.HTTPError,
+                          self.opener.open,
+                          self.URL)
+
+    def test_proxy_with_no_password_raises_httperror(self):
+        FakeProxyHandler.digest_auth_handler.set_qop("auth")
+        self.assertRaises(urllib2.HTTPError,
+                          self.opener.open,
+                          self.URL)
+
+    def test_proxy_qop_auth_works(self):
+        self._digest_auth_handler.add_password(self.REALM, self.URL,
+                                               self.USER, self.PASSWD)
+        FakeProxyHandler.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._digest_auth_handler.add_password(self.REALM, self.URL,
+                                               self.USER, self.PASSWD)
+        FakeProxyHandler.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(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 = '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 = '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.assertEquals(data, expected_response)
+            self.assertEquals(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.assertEquals(data, expected_response)
+            self.assertEquals(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.assertEquals(data, expected_response)
+            self.assertEquals(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.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, [], "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, [], "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
+    # localhost.  However, if this is a bad rationale, then uncomment
+    # the next line.
+    #test_support.requires("network")
+
+    test_support.run_unittest(ProxyAuthTests)
+    test_support.run_unittest(TestUrlopen)
+
+if __name__ == "__main__":
+    test_main()