import unittest

from test import test_support

import os

import socket

import StringIO

import urllib2

from urllib2 import Request, OpenerDirector

# XXX

# Request

# CacheFTPHandler (hard to write)

# parse_keqv_list, parse_http_list, HTTPDigestAuthHandler

class TrivialTests(unittest.TestCase):

    def test_trivial(self):

        # A couple trivial tests

        self.assertRaises(ValueError, urllib2.urlopen, 'bogus url')

        # XXX Name hacking to get this to work on Windows.

        fname = os.path.abspath(urllib2.__file__).replace('\\', '/')

        if fname[1:2] == ":":

            fname = fname[2:]

        # And more hacking to get it to work on MacOS. This assumes

        # urllib.pathname2url works, unfortunately...

        if os.name == 'mac':

            fname = '/' + fname.replace(':', '/')

        elif os.name == 'riscos':

            import string

            fname = os.expand(fname)

            fname = fname.translate(string.maketrans("/.", "./"))

        file_url = "file://%s" % fname

        f = urllib2.urlopen(file_url)

        buf = f.read()

        f.close()

    def test_parse_http_list(self):

        tests = [('a,b,c', ['a', 'b', 'c']),

                 ('path"o,l"og"i"cal, example', ['path"o,l"og"i"cal', 'example']),

                 ('a, b, "c", "d", "e,f", g, h', ['a', 'b', '"c"', '"d"', '"e,f"', 'g', 'h']),

                 ('a="b\\"c", d="e\\,f", g="h\\\\i"', ['a="b"c"', 'd="e,f"', 'g="h\\i"'])]

        for string, list in tests:

            self.assertEquals(urllib2.parse_http_list(string), list)

def test_request_headers_dict():

    """

    The Request.headers dictionary is not a documented interface.  It should

    stay that way, because the complete set of headers are only accessible

    through the .get_header(), .has_header(), .header_items() interface.

    However, .headers pre-dates those methods, and so real code will be using

    the dictionary.

    The introduction in 2.4 of those methods was a mistake for the same reason:

    code that previously saw all (urllib2 user)-provided headers in .headers

    now sees only a subset (and the function interface is ugly and incomplete).

    A better change would have been to replace .headers dict with a dict

    subclass (or UserDict.DictMixin instance?)  that preserved the .headers

    interface and also provided access to the "unredirected" headers.  It's

    probably too late to fix that, though.

    Check .capitalize() case normalization:

    >>> url = "http://example.com"

    >>> Request(url, headers={"Spam-eggs": "blah"}).headers["Spam-eggs"]

    'blah'

    >>> Request(url, headers={"spam-EggS": "blah"}).headers["Spam-eggs"]

    'blah'

    Currently, Request(url, "Spam-eggs").headers["Spam-Eggs"] raises KeyError,

    but that could be changed in future.

    """

def test_request_headers_methods():

    """

    Note the case normalization of header names here, to .capitalize()-case.

    This should be preserved for backwards-compatibility.  (In the HTTP case,

    normalization to .title()-case is done by urllib2 before sending headers to

    httplib).

    >>> url = "http://example.com"

    >>> r = Request(url, headers={"Spam-eggs": "blah"})

    >>> r.has_header("Spam-eggs")

    True

    >>> r.header_items()

    [('Spam-eggs', 'blah')]

    >>> r.add_header("Foo-Bar", "baz")

    >>> items = r.header_items()

    >>> items.sort()

    >>> items

    [('Foo-bar', 'baz'), ('Spam-eggs', 'blah')]

    Note that e.g. r.has_header("spam-EggS") is currently False, and

    r.get_header("spam-EggS") returns None, but that could be changed in

    future.

    >>> r.has_header("Not-there")

    False

    >>> print r.get_header("Not-there")

    None

    >>> r.get_header("Not-there", "default")

    'default'

    """

def test_password_manager(self):

    """

    >>> mgr = urllib2.HTTPPasswordMgr()

    >>> add = mgr.add_password

    >>> add("Some Realm", "http://example.com/", "joe", "password")

    >>> add("Some Realm", "http://example.com/ni", "ni", "ni")

    >>> add("c", "http://example.com/foo", "foo", "ni")

    >>> add("c", "http://example.com/bar", "bar", "nini")

    >>> add("b", "http://example.com/", "first", "blah")

    >>> add("b", "http://example.com/", "second", "spam")

    >>> add("a", "http://example.com", "1", "a")

    >>> add("Some Realm", "http://c.example.com:3128", "3", "c")

    >>> add("Some Realm", "d.example.com", "4", "d")

    >>> add("Some Realm", "e.example.com:3128", "5", "e")

    >>> mgr.find_user_password("Some Realm", "example.com")

    ('joe', 'password')

    >>> mgr.find_user_password("Some Realm", "http://example.com")

    ('joe', 'password')

    >>> mgr.find_user_password("Some Realm", "http://example.com/")

    ('joe', 'password')

    >>> mgr.find_user_password("Some Realm", "http://example.com/spam")

    ('joe', 'password')

    >>> mgr.find_user_password("Some Realm", "http://example.com/spam/spam")

    ('joe', 'password')

    >>> mgr.find_user_password("c", "http://example.com/foo")

    ('foo', 'ni')

    >>> mgr.find_user_password("c", "http://example.com/bar")

    ('bar', 'nini')

    Actually, this is really undefined ATM

##     Currently, we use the highest-level path where more than one match:

##     >>> mgr.find_user_password("Some Realm", "http://example.com/ni")

##     ('joe', 'password')

    Use latest add_password() in case of conflict:

    >>> mgr.find_user_password("b", "http://example.com/")

    ('second', 'spam')

    No special relationship between a.example.com and example.com:

    >>> mgr.find_user_password("a", "http://example.com/")

    ('1', 'a')

    >>> mgr.find_user_password("a", "http://a.example.com/")

    (None, None)

    Ports:

    >>> mgr.find_user_password("Some Realm", "c.example.com")

    (None, None)

    >>> mgr.find_user_password("Some Realm", "c.example.com:3128")

    ('3', 'c')

    >>> mgr.find_user_password("Some Realm", "http://c.example.com:3128")

    ('3', 'c')

    >>> mgr.find_user_password("Some Realm", "d.example.com")

    ('4', 'd')

    >>> mgr.find_user_password("Some Realm", "e.example.com:3128")

    ('5', 'e')

    """

    pass

def test_password_manager_default_port(self):

    """

    >>> mgr = urllib2.HTTPPasswordMgr()

    >>> add = mgr.add_password

    The point to note here is that we can't guess the default port if there's

    no scheme.  This applies to both add_password and find_user_password.

    >>> add("f", "http://g.example.com:80", "10", "j")

    >>> add("g", "http://h.example.com", "11", "k")

    >>> add("h", "i.example.com:80", "12", "l")

    >>> add("i", "j.example.com", "13", "m")

    >>> mgr.find_user_password("f", "g.example.com:100")

    (None, None)

    >>> mgr.find_user_password("f", "g.example.com:80")

    ('10', 'j')

    >>> mgr.find_user_password("f", "g.example.com")

    (None, None)

    >>> mgr.find_user_password("f", "http://g.example.com:100")

    (None, None)

    >>> mgr.find_user_password("f", "http://g.example.com:80")

    ('10', 'j')

    >>> mgr.find_user_password("f", "http://g.example.com")

    ('10', 'j')

    >>> mgr.find_user_password("g", "h.example.com")

    ('11', 'k')

    >>> mgr.find_user_password("g", "h.example.com:80")

    ('11', 'k')

    >>> mgr.find_user_password("g", "http://h.example.com:80")

    ('11', 'k')

    >>> mgr.find_user_password("h", "i.example.com")

    (None, None)

    >>> mgr.find_user_password("h", "i.example.com:80")

    ('12', 'l')

    >>> mgr.find_user_password("h", "http://i.example.com:80")

    ('12', 'l')

    >>> mgr.find_user_password("i", "j.example.com")

    ('13', 'm')

    >>> mgr.find_user_password("i", "j.example.com:80")

    (None, None)

    >>> mgr.find_user_password("i", "http://j.example.com")

    ('13', 'm')

    >>> mgr.find_user_password("i", "http://j.example.com:80")

    (None, None)

    """

class MockOpener:

    addheaders = []

    def open(self, req, data=None):

        self.req, self.data = req, data

    def error(self, proto, *args):

        self.proto, self.args = proto, args

class MockFile:

    def read(self, count=None): pass

    def readline(self, count=None): pass

    def close(self): pass

class MockHeaders(dict):

    def getheaders(self, name):

        return self.values()

class MockResponse(StringIO.StringIO):

    def __init__(self, code, msg, headers, data, url=None):

        StringIO.StringIO.__init__(self, data)

        self.code, self.msg, self.headers, self.url = code, msg, headers, url

    def info(self):

        return self.headers

    def geturl(self):

        return self.url

class MockCookieJar:

    def add_cookie_header(self, request):

        self.ach_req = request

    def extract_cookies(self, response, request):

        self.ec_req, self.ec_r = request, response

class FakeMethod:

    def __init__(self, meth_name, action, handle):

        self.meth_name = meth_name

        self.handle = handle

        self.action = action

    def __call__(self, *args):

        return self.handle(self.meth_name, self.action, *args)

class MockHandler:

    # useful for testing handler machinery

    # see add_ordered_mock_handlers() docstring

    handler_order = 500

    def __init__(self, methods):

        self._define_methods(methods)

    def _define_methods(self, methods):

        for spec in methods:

            if len(spec) == 2: name, action = spec

            else: name, action = spec, None

            meth = FakeMethod(name, action, self.handle)

            setattr(self.__class__, name, meth)

    def handle(self, fn_name, action, *args, **kwds):

        self.parent.calls.append((self, fn_name, args, kwds))

        if action is None:

            return None

        elif action == "return self":

            return self

        elif action == "return response":

            res = MockResponse(200, "OK", {}, "")

            return res

        elif action == "return request":

            return Request("http://blah/")

        elif action.startswith("error"):

            code = action[action.rfind(" ")+1:]

            try:

                code = int(code)

            except ValueError:

                pass

            res = MockResponse(200, "OK", {}, "")

            return self.parent.error("http", args[0], res, code, "", {})

        elif action == "raise":

            raise urllib2.URLError("blah")

        assert False

    def close(self): pass

    def add_parent(self, parent):

        self.parent = parent

        self.parent.calls = []

    def __lt__(self, other):

        if not hasattr(other, "handler_order"):

            # No handler_order, leave in original order.  Yuck.

            return True

        return self.handler_order < other.handler_order

def add_ordered_mock_handlers(opener, meth_spec):

    """Create MockHandlers and add them to an OpenerDirector.

    meth_spec: list of lists of tuples and strings defining methods to define

    on handlers.  eg:

    [["http_error", "ftp_open"], ["http_open"]]

    defines methods .http_error() and .ftp_open() on one handler, and

    .http_open() on another.  These methods just record their arguments and

    return None.  Using a tuple instead of a string causes the method to

    perform some action (see MockHandler.handle()), eg:

    [["http_error"], [("http_open", "return request")]]

    defines .http_error() on one handler (which simply returns None), and

    .http_open() on another handler, which returns a Request object.

    """

    handlers = []

    count = 0

    for meths in meth_spec:

        class MockHandlerSubclass(MockHandler): pass

        h = MockHandlerSubclass(meths)

        h.handler_order += count

        h.add_parent(opener)

        count = count + 1

        handlers.append(h)

        opener.add_handler(h)

    return handlers

def build_test_opener(*handler_instances):

    opener = OpenerDirector()

    for h in handler_instances:

        opener.add_handler(h)

    return opener

class MockHTTPHandler(urllib2.BaseHandler):

    # useful for testing redirections and auth

    # sends supplied headers and code as first response

    # sends 200 OK as second response

    def __init__(self, code, headers):

        self.code = code

        self.headers = headers

        self.reset()

    def reset(self):

        self._count = 0

        self.requests = []

    def http_open(self, req):

        import mimetools, httplib, copy

        from StringIO import StringIO

        self.requests.append(copy.deepcopy(req))

        if self._count == 0:

            self._count = self._count + 1

            name = httplib.responses[self.code]

            msg = mimetools.Message(StringIO(self.headers))

            return self.parent.error(

                "http", req, MockFile(), self.code, name, msg)

        else:

            self.req = req

            msg = mimetools.Message(StringIO("\r\n\r\n"))

            return MockResponse(200, "OK", msg, "", req.get_full_url())

class MockPasswordManager:

    def add_password(self, realm, uri, user, password):

        self.realm = realm

        self.url = uri

        self.user = user

        self.password = password

    def find_user_password(self, realm, authuri):

        self.target_realm = realm

        self.target_url = authuri

        return self.user, self.password

class OpenerDirectorTests(unittest.TestCase):

    def test_add_non_handler(self):

        class NonHandler(object):

            pass

        self.assertRaises(TypeError,

                          OpenerDirector().add_handler, NonHandler())

    def test_badly_named_methods(self):

        # test work-around for three methods that accidentally follow the

        # naming conventions for handler methods

        # (*_open() / *_request() / *_response())

        # These used to call the accidentally-named methods, causing a

        # TypeError in real code; here, returning self from these mock

        # methods would either cause no exception, or AttributeError.

        from urllib2 import URLError

        o = OpenerDirector()

        meth_spec = [

            [("do_open", "return self"), ("proxy_open", "return self")],

            [("redirect_request", "return self")],

            ]

        handlers = add_ordered_mock_handlers(o, meth_spec)

        o.add_handler(urllib2.UnknownHandler())

        for scheme in "do", "proxy", "redirect":

            self.assertRaises(URLError, o.open, scheme+"://example.com/")

    def test_handled(self):

        # handler returning non-None means no more handlers will be called

        o = OpenerDirector()

        meth_spec = [

            ["http_open", "ftp_open", "http_error_302"],

            ["ftp_open"],

            [("http_open", "return self")],

            [("http_open", "return self")],

            ]

        handlers = add_ordered_mock_handlers(o, meth_spec)

        req = Request("http://example.com/")

        r = o.open(req)

        # Second .http_open() gets called, third doesn't, since second returned

        # non-None.  Handlers without .http_open() never get any methods called

        # on them.

        # In fact, second mock handler defining .http_open() returns self

        # (instead of response), which becomes the OpenerDirector's return

        # value.

        self.assertEqual(r, handlers[2])

        calls = [(handlers[0], "http_open"), (handlers[2], "http_open")]

        for expected, got in zip(calls, o.calls):

            handler, name, args, kwds = got

            self.assertEqual((handler, name), expected)

            self.assertEqual(args, (req,))

    def test_handler_order(self):

        o = OpenerDirector()

        handlers = []

        for meths, handler_order in [

            ([("http_open", "return self")], 500),

            (["http_open"], 0),

            ]:

            class MockHandlerSubclass(MockHandler): pass

            h = MockHandlerSubclass(meths)

            h.handler_order = handler_order

            handlers.append(h)

            o.add_handler(h)

        r = o.open("http://example.com/")

        # handlers called in reverse order, thanks to their sort order

        self.assertEqual(o.calls[0][0], handlers[1])

        self.assertEqual(o.calls[1][0], handlers[0])

    def test_raise(self):

        # raising URLError stops processing of request

        o = OpenerDirector()

        meth_spec = [

            [("http_open", "raise")],

            [("http_open", "return self")],

            ]

        handlers = add_ordered_mock_handlers(o, meth_spec)

        req = Request("http://example.com/")

        self.assertRaises(urllib2.URLError, o.open, req)

        self.assertEqual(o.calls, [(handlers[0], "http_open", (req,), {})])

##     def test_error(self):

##         # XXX this doesn't actually seem to be used in standard library,

##         #  but should really be tested anyway...

    def test_http_error(self):

        # XXX http_error_default

        # http errors are a special case

        o = OpenerDirector()

        meth_spec = [

            [("http_open", "error 302")],

            [("http_error_400", "raise"), "http_open"],

            [("http_error_302", "return response"), "http_error_303",

             "http_error"],

            [("http_error_302")],

            ]

        handlers = add_ordered_mock_handlers(o, meth_spec)

        class Unknown:

            def __eq__(self, other): return True

        req = Request("http://example.com/")

        r = o.open(req)

        assert len(o.calls) == 2

        calls = [(handlers[0], "http_open", (req,)),

                 (handlers[2], "http_error_302",

                  (req, Unknown(), 302, "", {}))]

        for expected, got in zip(calls, o.calls):

            handler, method_name, args = expected

            self.assertEqual((handler, method_name), got[:2])

            self.assertEqual(args, got[2])

    def test_processors(self):

        # *_request / *_response methods get called appropriately

        o = OpenerDirector()

        meth_spec = [

            [("http_request", "return request"),

             ("http_response", "return response")],

            [("http_request", "return request"),

             ("http_response", "return response")],

            ]

        handlers = add_ordered_mock_handlers(o, meth_spec)

        req = Request("http://example.com/")

        r = o.open(req)

        # processor methods are called on *all* handlers that define them,

        # not just the first handler that handles the request

        calls = [

            (handlers[0], "http_request"), (handlers[1], "http_request"),

            (handlers[0], "http_response"), (handlers[1], "http_response")]

        for i, (handler, name, args, kwds) in enumerate(o.calls):

            if i < 2:

                # *_request

                self.assertEqual((handler, name), calls[i])

                self.assertEqual(len(args), 1)

                self.assert_(isinstance(args[0], Request))

            else:

                # *_response

                self.assertEqual((handler, name), calls[i])

                self.assertEqual(len(args), 2)