import unittest
from test import support
from test import test_urllib

import os
import io
import socket
import array
import sys
import tempfile
import subprocess

import urllib.request
# The proxy bypass method imported below has logic specific to the OSX
# proxy config data structure but is testable on all platforms.
from urllib.request import (Request, OpenerDirector, HTTPBasicAuthHandler,
                            HTTPPasswordMgrWithPriorAuth, _parse_proxy,
                            _proxy_bypass_macosx_sysconf,
                            AbstractDigestAuthHandler)
from urllib.parse import urlparse
import urllib.error
import http.client

# XXX
# Request
# CacheFTPHandler (hard to write)
# parse_keqv_list, parse_http_list, HTTPDigestAuthHandler


class TrivialTests(unittest.TestCase):

    def test___all__(self):
        # Verify which names are exposed
        for module in 'request', 'response', 'parse', 'error', 'robotparser':
            context = {}
            exec('from urllib.%s import *' % module, context)
            del context['__builtins__']
            if module == 'request' and os.name == 'nt':
                u, p = context.pop('url2pathname'), context.pop('pathname2url')
                self.assertEqual(u.__module__, 'nturl2path')
                self.assertEqual(p.__module__, 'nturl2path')
            for k, v in context.items():
                self.assertEqual(v.__module__, 'urllib.%s' % module,
                    "%r is exposed in 'urllib.%s' but defined in %r" %
                    (k, module, v.__module__))

    def test_trivial(self):
        # A couple trivial tests

        self.assertRaises(ValueError, urllib.request.urlopen, 'bogus url')

        # XXX Name hacking to get this to work on Windows.
        fname = os.path.abspath(urllib.request.__file__).replace(os.sep, '/')

        if os.name == 'nt':
            file_url = "file:///%s" % fname
        else:
            file_url = "file://%s" % fname

        with urllib.request.urlopen(file_url) as f:
            f.read()

    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.assertEqual(urllib.request.parse_http_list(string), list)

    def test_URLError_reasonstr(self):
        err = urllib.error.URLError('reason')
        self.assertIn(err.reason, str(err))


class RequestHdrsTests(unittest.TestCase):

    def test_request_headers_dict(self):
        """
        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.

        """
        url = "http://example.com"
        self.assertEqual(Request(url,
                                 headers={"Spam-eggs": "blah"}
                                 ).headers["Spam-eggs"], "blah")
        self.assertEqual(Request(url,
                                 headers={"spam-EggS": "blah"}
                                 ).headers["Spam-eggs"], "blah")

    def test_request_headers_methods(self):
        """
        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
        http.client).

        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.

        Method r.remove_header should remove items both from r.headers and
        r.unredirected_hdrs dictionaries
        """
        url = "http://example.com"
        req = Request(url, headers={"Spam-eggs": "blah"})
        self.assertTrue(req.has_header("Spam-eggs"))
        self.assertEqual(req.header_items(), [('Spam-eggs', 'blah')])

        req.add_header("Foo-Bar", "baz")
        self.assertEqual(sorted(req.header_items()),
                         [('Foo-bar', 'baz'), ('Spam-eggs', 'blah')])
        self.assertFalse(req.has_header("Not-there"))
        self.assertIsNone(req.get_header("Not-there"))
        self.assertEqual(req.get_header("Not-there", "default"), "default")

        req.remove_header("Spam-eggs")
        self.assertFalse(req.has_header("Spam-eggs"))

        req.add_unredirected_header("Unredirected-spam", "Eggs")
        self.assertTrue(req.has_header("Unredirected-spam"))

        req.remove_header("Unredirected-spam")
        self.assertFalse(req.has_header("Unredirected-spam"))

    def test_password_manager(self):
        mgr = urllib.request.HTTPPasswordMgr()
        add = mgr.add_password
        find_user_pass = mgr.find_user_password

        add("Some Realm", "http://example.com/", "joe", "password")
        add("Some Realm", "http://example.com/ni", "ni", "ni")
        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")

        # For the same realm, password set the highest path is the winner.
        self.assertEqual(find_user_pass("Some Realm", "example.com"),
                         ('joe', 'password'))
        self.assertEqual(find_user_pass("Some Realm", "http://example.com/ni"),
                         ('joe', 'password'))
        self.assertEqual(find_user_pass("Some Realm", "http://example.com"),
                         ('joe', 'password'))
        self.assertEqual(find_user_pass("Some Realm", "http://example.com/"),
                         ('joe', 'password'))
        self.assertEqual(find_user_pass("Some Realm",
                                        "http://example.com/spam"),
                         ('joe', 'password'))

        self.assertEqual(find_user_pass("Some Realm",
                                        "http://example.com/spam/spam"),
                         ('joe', 'password'))

        # You can have different passwords for different paths.

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

        self.assertEqual(find_user_pass("c", "http://example.com/foo"),
                         ('foo', 'ni'))

        self.assertEqual(find_user_pass("c", "http://example.com/bar"),
                         ('bar', 'nini'))

        # For the same path, newer password should be considered.

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

        self.assertEqual(find_user_pass("b", "http://example.com/"),
                         ('second', 'spam'))

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

        add("a", "http://example.com", "1", "a")
        self.assertEqual(find_user_pass("a", "http://example.com/"),
                         ('1', 'a'))

        self.assertEqual(find_user_pass("a", "http://a.example.com/"),
                         (None, None))

        # Ports:

        self.assertEqual(find_user_pass("Some Realm", "c.example.com"),
                         (None, None))
        self.assertEqual(find_user_pass("Some Realm", "c.example.com:3128"),
                         ('3', 'c'))
        self.assertEqual(
            find_user_pass("Some Realm", "http://c.example.com:3128"),
            ('3', 'c'))
        self.assertEqual(find_user_pass("Some Realm", "d.example.com"),
                         ('4', 'd'))
        self.assertEqual(find_user_pass("Some Realm", "e.example.com:3128"),
                         ('5', 'e'))

    def test_password_manager_default_port(self):
        """
        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.
        """
        mgr = urllib.request.HTTPPasswordMgr()
        add = mgr.add_password
        find_user_pass = mgr.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")
        self.assertEqual(find_user_pass("f", "g.example.com:100"),
                         (None, None))
        self.assertEqual(find_user_pass("f", "g.example.com:80"),
                         ('10', 'j'))
        self.assertEqual(find_user_pass("f", "g.example.com"),
                         (None, None))
        self.assertEqual(find_user_pass("f", "http://g.example.com:100"),
                         (None, None))
        self.assertEqual(find_user_pass("f", "http://g.example.com:80"),
                         ('10', 'j'))
        self.assertEqual(find_user_pass("f", "http://g.example.com"),
                         ('10', 'j'))
        self.assertEqual(find_user_pass("g", "h.example.com"), ('11', 'k'))
        self.assertEqual(find_user_pass("g", "h.example.com:80"), ('11', 'k'))
        self.assertEqual(find_user_pass("g", "http://h.example.com:80"),
                         ('11', 'k'))
        self.assertEqual(find_user_pass("h", "i.example.com"), (None, None))
        self.assertEqual(find_user_pass("h", "i.example.com:80"), ('12', 'l'))
        self.assertEqual(find_user_pass("h", "http://i.example.com:80"),
                         ('12', 'l'))
        self.assertEqual(find_user_pass("i", "j.example.com"), ('13', 'm'))
        self.assertEqual(find_user_pass("i", "j.example.com:80"),
                         (None, None))
        self.assertEqual(find_user_pass("i", "http://j.example.com"),
                         ('13', 'm'))
        self.assertEqual(find_user_pass("i", "http://j.example.com:80"),
                         (None, None))


class MockOpener:
    addheaders = []

    def open(self, req, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
        self.req, self.data, self.timeout = req, data, timeout

    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 list(self.values())


class MockResponse(io.StringIO):
    def __init__(self, code, msg, headers, data, url=None):
        io.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 MockHTTPResponse(io.IOBase):
    def __init__(self, fp, msg, status, reason):
        self.fp = fp
        self.msg = msg
        self.status = status
        self.reason = reason
        self.code = 200

    def read(self):
        return ''

    def info(self):
        return {}

    def geturl(self):
        return self.url


class MockHTTPClass:
    def __init__(self):
        self.level = 0
        self.req_headers = []
        self.data = None
        self.raise_on_endheaders = False
        self.sock = None
        self._tunnel_headers = {}

    def __call__(self, host, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
        self.host = host
        self.timeout = timeout
        return self

    def set_debuglevel(self, level):
        self.level = level

    def set_tunnel(self, host, port=None, headers=None):
        self._tunnel_host = host
        self._tunnel_port = port
        if headers:
            self._tunnel_headers = headers