import warnings
import unittest
import urllib.request
import urllib.error
import urllib.parse

from openid import fetchers

# XXX: make these separate test cases


def _assertEqual(v1, v2, extra):
    try:
        assert v1 == v2
    except AssertionError:
        raise AssertionError("%r != %r ; context %r" % (v1, v2, extra))


def failUnlessResponseExpected(expected, actual, extra):
    _assertEqual(expected.final_url, actual.final_url, extra)
    _assertEqual(expected.status, actual.status, extra)
    _assertEqual(expected.body, actual.body, extra)
    got_headers = dict(actual.headers)

    del got_headers['date']
    del got_headers['server']

    for k, v in expected.headers.items():
        assert got_headers[k] == v, (k, v, got_headers[k], extra)


def test_fetcher(fetcher, should_raise_exc, server):

    def geturl(path):
        host, port = server.server_address
        return 'http://%s:%s%s' % (host, port, path)

    expected_headers = {'content-type': 'text/plain'}

    def plain(path, code):
        path = '/' + path
        expected = fetchers.HTTPResponse(
            geturl(path), code, expected_headers, path)
        return (path, expected)

    expect_success = fetchers.HTTPResponse(
        geturl('/success'), 200, expected_headers, '/success')
    cases = [
        ('/success', expect_success),
        ('/301redirect', expect_success),
        ('/302redirect', expect_success),
        ('/303redirect', expect_success),
        ('/307redirect', expect_success),
        plain('notfound', 404),
        plain('badreq', 400),
        plain('forbidden', 403),
        plain('error', 500),
        plain('server_error', 503),
        ]

    for path, expected in cases:
        fetch_url = geturl(path)
        try:
            actual = fetcher.fetch(fetch_url)
        except (SystemExit, KeyboardInterrupt):
            pass
        except Exception as e:
            raise AssertionError((fetcher, fetch_url, e))
        else:
            failUnlessResponseExpected(expected, actual, extra=locals())

    for err_url in [
            geturl('/closed'),
            'http://invalid.janrain.com/',
            'not:a/url',
            'ftp://janrain.com/pub/',
            ]:
        try:
            result = fetcher.fetch(err_url)
        except (KeyboardInterrupt, SystemExit):
            raise
        except fetchers.HTTPError:
            # This is raised by the Curl fetcher for bad cases
            # detected by the fetchers module, but it's a subclass of
            # HTTPFetchingError, so we have to catch it explicitly.
            assert should_raise_exc
        except fetchers.HTTPFetchingError:
            assert not should_raise_exc, (fetcher, should_raise_exc, server)
        except Exception as e:
            assert should_raise_exc
        else:
            assert False, 'An exception was expected for %r (%r)' % (
                fetcher, result)


def run_fetcher_tests(server):
    exc_fetchers = []
    for klass, library_name in [
        (fetchers.Urllib2Fetcher, 'urllib2'),
        (fetchers.CurlHTTPFetcher, 'pycurl'),
        (fetchers.HTTPLib2Fetcher, 'httplib2'),
        ]:
        try:
            exc_fetchers.append(klass())
        except RuntimeError as why:
            if str(why).startswith('Cannot find %s library' % (library_name,)):
                try:
                    __import__(library_name)
                except ImportError:
                    warnings.warn(
                        'Skipping tests for %r fetcher because '
                        'the library did not import.' % (library_name,))
                    pass
                else:
                    assert False, ('%s present but not detected' % (
                        library_name,))
            else:
                raise

    non_exc_fetchers = []
    for f in exc_fetchers:
        non_exc_fetchers.append(fetchers.ExceptionWrappingFetcher(f))

    for f in exc_fetchers:
        test_fetcher(f, True, server)

    for f in non_exc_fetchers:
        test_fetcher(f, False, server)

from http.server import BaseHTTPRequestHandler, HTTPServer


class FetcherTestHandler(BaseHTTPRequestHandler):
    cases = {
        '/success': (200, None),
        '/301redirect': (301, '/success'),
        '/302redirect': (302, '/success'),
        '/303redirect': (303, '/success'),
        '/307redirect': (307, '/success'),
        '/notfound': (404, None),
        '/badreq': (400, None),
        '/forbidden': (403, None),
        '/error': (500, None),
        '/server_error': (503, None),
        }

    def log_request(self, *args):
        pass

    def do_GET(self):
        if self.path == '/closed':
            # *somehow* a ResourceWarning gets raised with an unclosed socket
            # when this is hit. It's virtually impossible to find out where.
            # Since ResourceWarnings are just warnings, ignore for now.
            pass
        else:
            try:
                http_code, location = self.cases[self.path]
            except KeyError:
                self.errorResponse('Bad path')
            else:
                extra_headers = [('Content-type', 'text/plain')]
                if location is not None:
                    host, port = self.server.server_address
                    base = ('http://%s:%s' % (host, port,))
                    location = base + location
                    extra_headers.append(('Location', location))
                self._respond(http_code, extra_headers, self.path)

    def do_POST(self):
        try:
            http_code, extra_headers = self.cases[self.path]
        except KeyError:
            self.errorResponse('Bad path')
        else:
            if http_code in [301, 302, 303, 307]:
                self.errorResponse()
            else:
                content_type = self.headers.get('content-type', 'text/plain')
                extra_headers.append(('Content-type', content_type))
                content_length = int(self.headers.get('Content-length', '-1'))
                body = self.rfile.read(content_length)
                self._respond(http_code, extra_headers, body)

    def errorResponse(self, message=None):
        req = [
            ('HTTP method', self.command),
            ('path', self.path),
            ]
        if message:
            req.append(('message', message))

        body_parts = ['Bad request:\r\n']
        for k, v in req:
            body_parts.append(' %s: %s\r\n' % (k, v))
        body = ''.join(body_parts)
        self._respond(400, [('Content-type', 'text/plain')], body)

    def _respond(self, http_code, extra_headers, body):
        self.send_response(http_code)
        for k, v in extra_headers:
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(bytes(body, encoding="utf-8"))

    # def finish(self):
    #     if not self.wfile.closed:
    #         self.wfile.flush()
    #         # self.wfile.close()
    #     # self.rfile.close()


def test():
    host = 'localhost'
    # When I use port 0 here, it works for the first fetch and the
    # next one gets connection refused.  Bummer.  So instead, pick a
    # port that's *probably* not in use.
    import os
    port = (os.getpid() % 31000) + 1024

    server = HTTPServer((host, port), FetcherTestHandler)

    import threading
    server_thread = threading.Thread(target=server.serve_forever)
    server_thread.setDaemon(True)
    server_thread.start()

    run_fetcher_tests(server)

    server.shutdown()


class FakeFetcher(object):
    sentinel = object()

    def fetch(self, *args, **kwargs):
        return self.sentinel


class DefaultFetcherTest(unittest.TestCase):
    def setUp(self):
        """reset the default fetcher to None"""
        fetchers.setDefaultFetcher(None)

    def tearDown(self):
        """reset the default fetcher to None"""
        fetchers.setDefaultFetcher(None)

    def test_getDefaultNotNone(self):
        """Make sure that None is never returned as a default fetcher"""
        self.assertTrue(fetchers.getDefaultFetcher() is not None)
        fetchers.setDefaultFetcher(None)
        self.assertTrue(fetchers.getDefaultFetcher() is not None)

    def test_setDefault(self):
        """Make sure the getDefaultFetcher returns the object set for
        setDefaultFetcher"""
        sentinel = object()
        fetchers.setDefaultFetcher(sentinel, wrap_exceptions=False)
        self.assertTrue(fetchers.getDefaultFetcher() is sentinel)

    def test_callFetch(self):
        """Make sure that fetchers.fetch() uses the default fetcher
        instance that was set."""
        fetchers.setDefaultFetcher(FakeFetcher())
        actual = fetchers.fetch('bad://url')
        self.assertTrue(actual is FakeFetcher.sentinel)

    def test_wrappedByDefault(self):
        """Make sure that the default fetcher instance wraps
        exceptions by default"""
        default_fetcher = fetchers.getDefaultFetcher()
        self.assertIsInstance(
            default_fetcher, fetchers.ExceptionWrappingFetcher)

        self.assertRaises(fetchers.HTTPFetchingError,
                          fetchers.fetch, 'http://invalid.janrain.com/')

    def test_notWrapped(self):
        """Make sure that if we set a non-wrapped fetcher as default,
        it will not wrap exceptions."""
        # A fetcher that will raise an exception when it encounters a
        # host that will not resolve
        fetcher = fetchers.Urllib2Fetcher()
        fetchers.setDefaultFetcher(fetcher, wrap_exceptions=False)

        self.assertFalse(isinstance(fetchers.getDefaultFetcher(),
                               fetchers.ExceptionWrappingFetcher))

        try:
            fetchers.fetch('http://invalid.janrain.com/')
        except fetchers.HTTPFetchingError:
            self.fail('Should not be wrapping exception')
        except Exception as exc:
            self.assertIsInstance(exc, urllib.error.URLError)
            pass
        else:
            self.fail('Should have raised an exception')


class Urllib2FetcherTests(unittest.TestCase):
    '''Make sure a few of the utility methods are also covered by tests.'''
    def setUp(self):
        self.fetcher = fetchers.Urllib2Fetcher()

    def test_disallowed(self):
        '''
        Test that the _allowedURL function only lets through the right things.
        '''
        for url in ["file://localhost/thing.txt", "ftp://server/path",
                    "sftp://server/path", "ssh://server/path"]:
            self.assertEqual(fetchers._allowedURL(url), False)

    def test_lowerCaseKeys(self):
        uppercased = {'Content-Type': None, 'HiPPyHiPPyShAKe': None}
        lowercased = {'content-type': None, 'hippyhippyshake': None}
        self.assertEqual(self.fetcher._lowerCaseKeys(uppercased), lowercased)

    def test_parseHeaderValue(self):
        headers_parsed = [
            ("text/html; charset=latin-1",
             ("text/html", {"charset": "latin-1"})),
            ("1; mode=block", ("1", {"mode": "block"})),
            ("foo; bar=baz; thing=quux",
             ("foo", {"bar": "baz", "thing": "quux"})),
        ]
        for s, p in headers_parsed:
            self.assertEqual(self.fetcher._parseHeaderValue(s), p)


def pyUnitTests():
    case1 = unittest.FunctionTestCase(test)
    loadTests = unittest.defaultTestLoader.loadTestsFromTestCase
    case2 = loadTests(DefaultFetcherTest)
    case3 = loadTests(Urllib2FetcherTests)
    return unittest.TestSuite([case1, case2, case3])