"""Regression tests for what was in Python 2's "urllib" module"""

import urllib.parse
import urllib.request
import urllib.error
import http.client
import email.message
import io
import unittest
from unittest.mock import patch
from test import support
import os
try:
    import ssl
except ImportError:
    ssl = None
import sys
import tempfile
from nturl2path import url2pathname, pathname2url

from base64 import b64encode
import collections


def hexescape(char):
    """Escape char as RFC 2396 specifies"""
    hex_repr = hex(ord(char))[2:].upper()
    if len(hex_repr) == 1:
        hex_repr = "0%s" % hex_repr
    return "%" + hex_repr

# Shortcut for testing FancyURLopener
_urlopener = None


def urlopen(url, data=None, proxies=None):
    """urlopen(url [, data]) -> open file-like object"""
    global _urlopener
    if proxies is not None:
        opener = urllib.request.FancyURLopener(proxies=proxies)
    elif not _urlopener:
        opener = FancyURLopener()
        _urlopener = opener
    else:
        opener = _urlopener
    if data is None:
        return opener.open(url)
    else:
        return opener.open(url, data)


def FancyURLopener():
    with support.check_warnings(
            ('FancyURLopener style of invoking requests is deprecated.',
            DeprecationWarning)):
        return urllib.request.FancyURLopener()


def fakehttp(fakedata, mock_close=False):
    class FakeSocket(io.BytesIO):
        io_refs = 1

        def sendall(self, data):
            FakeHTTPConnection.buf = data

        def makefile(self, *args, **kwds):
            self.io_refs += 1
            return self

        def read(self, amt=None):
            if self.closed:
                return b""
            return io.BytesIO.read(self, amt)

        def readline(self, length=None):
            if self.closed:
                return b""
            return io.BytesIO.readline(self, length)

        def close(self):
            self.io_refs -= 1
            if self.io_refs == 0:
                io.BytesIO.close(self)

    class FakeHTTPConnection(http.client.HTTPConnection):

        # buffer to store data for verification in urlopen tests.
        buf = None

        def connect(self):
            self.sock = FakeSocket(self.fakedata)
            type(self).fakesock = self.sock

        if mock_close:
            # bpo-36918: HTTPConnection destructor calls close() which calls
            # flush(). Problem: flush() calls self.fp.flush() which raises
            # "ValueError: I/O operation on closed file" which is logged as an
            # "Exception ignored in". Override close() to silence this error.
            def close(self):
                pass
    FakeHTTPConnection.fakedata = fakedata

    return FakeHTTPConnection


class FakeHTTPMixin(object):
    def fakehttp(self, fakedata, mock_close=False):
        fake_http_class = fakehttp(fakedata, mock_close=mock_close)
        self._connection_class = http.client.HTTPConnection
        http.client.HTTPConnection = fake_http_class

    def unfakehttp(self):
        http.client.HTTPConnection = self._connection_class


class FakeFTPMixin(object):
    def fakeftp(self):
        class FakeFtpWrapper(object):
            def __init__(self,  user, passwd, host, port, dirs, timeout=None,
                     persistent=True):
                pass

            def retrfile(self, file, type):
                return io.BytesIO(), 0

            def close(self):
                pass

        self._ftpwrapper_class = urllib.request.ftpwrapper
        urllib.request.ftpwrapper = FakeFtpWrapper

    def unfakeftp(self):
        urllib.request.ftpwrapper = self._ftpwrapper_class


class urlopen_FileTests(unittest.TestCase):
    """Test urlopen() opening a temporary file.

    Try to test as much functionality as possible so as to cut down on reliance
    on connecting to the Net for testing.

    """

    def setUp(self):
        # Create a temp file to use for testing
        self.text = bytes("test_urllib: %s\n" % self.__class__.__name__,
                          "ascii")
        f = open(support.TESTFN, 'wb')
        try:
            f.write(self.text)
        finally:
            f.close()
        self.pathname = support.TESTFN
        self.returned_obj = urlopen("file:%s" % self.pathname)

    def tearDown(self):
        """Shut down the open object"""
        self.returned_obj.close()
        os.remove(support.TESTFN)

    def test_interface(self):
        # Make sure object returned by urlopen() has the specified methods
        for attr in ("read", "readline", "readlines", "fileno",
                     "close", "info", "geturl", "getcode", "__iter__"):
            self.assertTrue(hasattr(self.returned_obj, attr),
                         "object returned by urlopen() lacks %s attribute" %
                         attr)

    def test_read(self):
        self.assertEqual(self.text, self.returned_obj.read())

    def test_readline(self):
        self.assertEqual(self.text, self.returned_obj.readline())
        self.assertEqual(b'', self.returned_obj.readline(),
                         "calling readline() after exhausting the file did not"
                         " return an empty string")

    def test_readlines(self):
        lines_list = self.returned_obj.readlines()
        self.assertEqual(len(lines_list), 1,
                         "readlines() returned the wrong number of lines")
        self.assertEqual(lines_list[0], self.text,
                         "readlines() returned improper text")

    def test_fileno(self):
        file_num = self.returned_obj.fileno()
        self.assertIsInstance(file_num, int, "fileno() did not return an int")
        self.assertEqual(os.read(file_num, len(self.text)), self.text,
                         "Reading on the file descriptor returned by fileno() "
                         "did not return the expected text")

    def test_close(self):
        # Test close() by calling it here and then having it be called again
        # by the tearDown() method for the test
        self.returned_obj.close()

    def test_info(self):
        self.assertIsInstance(self.returned_obj.info(), email.message.Message)

    def test_geturl(self):
        self.assertEqual(self.returned_obj.geturl(), self.pathname)

    def test_getcode(self):
        self.assertIsNone(self.returned_obj.getcode())

    def test_iter(self):
        # Test iterator
        # Don't need to count number of iterations since test would fail the
        # instant it returned anything beyond the first line from the
        # comparison.
        # Use the iterator in the usual implicit way to test for ticket #4608.
        for line in self.returned_obj:
            self.assertEqual(line, self.text)

    def test_relativelocalfile(self):
        self.assertRaises(ValueError,urllib.request.urlopen,'./' + self.pathname)


class ProxyTests(unittest.TestCase):

    def setUp(self):
        # Records changes to env vars
        self.env = support.EnvironmentVarGuard()
        # Delete all proxy related env vars
        for k in list(os.environ):
            if 'proxy' in k.lower():
                self.env.unset(k)

    def tearDown(self):
        # Restore all proxy related env vars
        self.env.__exit__()
        del self.env

    def test_getproxies_environment_keep_no_proxies(self):
        self.env.set('NO_PROXY', 'localhost')
        proxies = urllib.request.getproxies_environment()
        # getproxies_environment use lowered case truncated (no '_proxy') keys
        self.assertEqual('localhost', proxies['no'])
        # List of no_proxies with space.
        self.env.set('NO_PROXY', 'localhost, anotherdomain.com, newdomain.com:1234')
        self.assertTrue(urllib.request.proxy_bypass_environment('anotherdomain.com'))
        self.assertTrue(urllib.request.proxy_bypass_environment('anotherdomain.com:8888'))
        self.assertTrue(urllib.request.proxy_bypass_environment('newdomain.com:1234'))

    def test_proxy_cgi_ignore(self):
        try:
            self.env.set('HTTP_PROXY', 'http://somewhere:3128')
            proxies = urllib.request.getproxies_environment()
            self.assertEqual('http://somewhere:3128', proxies['http'])
            self.env.set('REQUEST_METHOD', 'GET')
            proxies = urllib.request.getproxies_environment()
            self.assertNotIn('http', proxies)
        finally:
            self.env.unset('REQUEST_METHOD')
            self.env.unset('HTTP_PROXY')

    def test_proxy_bypass_environment_host_match(self):
        bypass = urllib.request.proxy_bypass_environment
        self.env.set('NO_PROXY',
                     'localhost, anotherdomain.com, newdomain.com:1234, .d.o.t')
        self.assertTrue(bypass('localhost'))
        self.assertTrue(bypass('LocalHost'))                 # MixedCase
        self.assertTrue(bypass('LOCALHOST'))                 # UPPERCASE
        self.assertTrue(bypass('.localhost'))
        self.assertTrue(bypass('newdomain.com:1234'))
        self.assertTrue(bypass('.newdomain.com:1234'))
        self.assertTrue(bypass('foo.d.o.t'))                 # issue 29142
        self.assertTrue(bypass('d.o.t'))
        self.assertTrue(bypass('anotherdomain.com:8888'))
        self.assertTrue(bypass('.anotherdomain.com:8888'))
        self.assertTrue(bypass('www.newdomain.com:1234'))
        self.assertFalse(bypass('prelocalhost'))
        self.assertFalse(bypass('newdomain.com'))            # no port
        self.assertFalse(bypass('newdomain.com:1235'))       # wrong port

    def test_proxy_bypass_environment_always_match(self):
        bypass = urllib.request.proxy_bypass_environment
        self.env.set('NO_PROXY', '*')
        self.assertTrue(bypass('newdomain.com'))
        self.assertTrue(bypass('newdomain.com:1234'))
        self.env.set('NO_PROXY', '*, anotherdomain.com')
        self.assertTrue(bypass('anotherdomain.com'))
        self.assertFalse(bypass('newdomain.com'))
        self.assertFalse(bypass('newdomain.com:1234'))

    def test_proxy_bypass_environment_newline(self):
        bypass = urllib.request.proxy_bypass_environment
        self.env.set('NO_PROXY',
                     'localhost, anotherdomain.com, newdomain.com:1234')
        self.assertFalse(bypass('localhost\n'))
        self.assertFalse(bypass('anotherdomain.com:8888\n'))
        self.assertFalse(bypass('newdomain.com:1234\n'))


class ProxyTests_withOrderedEnv(unittest.TestCase):

    def setUp(self):
        # We need to test conditions, where variable order _is_ significant
        self._saved_env = os.environ
        # Monkey patch os.environ, start with empty fake environment
        os.environ = collections.OrderedDict()

    def tearDown(self):
        os.environ = self._saved_env

    def test_getproxies_environment_prefer_lowercase(self):
        # Test lowercase preference with removal
        os.environ['no_proxy'] = ''
        os.environ['No_Proxy'] = 'localhost'
        self.assertFalse(urllib.request.proxy_bypass_environment('localhost'))
        self.assertFalse(urllib.request.proxy_bypass_environment('arbitrary'))
        os.environ['http_proxy'] = ''
        os.environ['HTTP_PROXY'] = 'http://somewhere:3128'
        proxies = urllib.request.getproxies_environment()
        self.assertEqual({}, proxies)
        # Test lowercase preference of proxy bypass and correct matching including ports
        os.environ['no_proxy'] = 'localhost, noproxy.com, my.proxy:1234'
        os.environ['No_Proxy'] = 'xyz.com'
        self.assertTrue(urllib.request.proxy_bypass_environment('localhost'))
        self.assertTrue(urllib.request.proxy_bypass_environment('noproxy.com:5678'))
        self.assertTrue(urllib.request.proxy_bypass_environment('my.proxy:1234'))
        self.assertFalse(urllib.request.proxy_bypass_environment('my.proxy'))
        self.assertFalse(urllib.request.proxy_bypass_environment('arbitrary'))
        # Test lowercase preference with replacement
        os.environ['http_proxy'] = 'http://somewhere:3128'
        os.environ['Http_Proxy'] = 'http://somewhereelse:3128'
        proxies = urllib.request.getproxies_environment()
        self.assertEqual('http://somewhere:3128', proxies['http'])


class urlopen_HttpTests(unittest.TestCase, FakeHTTPMixin, FakeFTPMixin):
    """Test urlopen() opening a fake http connection."""

    def check_read(self, ver):
        self.fakehttp(b"HTTP/" + ver + b" 200 OK\r\n\r\nHello!")
        try:
            fp = urlopen("http://python.org/")
            self.assertEqual(fp.readline(), b"Hello!")
            self.assertEqual(fp.readline(), b"")
            self.assertEqual(fp.geturl(), 'http://python.org/')
            self.assertEqual(fp.getcode(), 200)
        finally:
            self.unfakehttp()

    def test_url_fragment(self):