class AdapterHTTPError(IOError):
    """An exception which must be raised by adapters when an HTTP response
    with a non-successful status code has been received.

    Base Geocoder class translates this exception to an instance of
    :class:`geopy.exc.GeocoderServiceError`.

    """

    def __init__(self, message, *, status_code, headers, text):
        """

        :param str message: Standard exception message.
        :param int status_code: HTTP status code.
        :param dict headers: HTTP response readers. A mapping object
            with lowercased or case-insensitive keys.

            .. versionadded:: 2.2
        :param str text: HTTP body text.
        """
        self.status_code = status_code
        self.headers = headers
        self.text = text
        super().__init__(message) class AioHTTPAdapter(BaseAsyncAdapter):
    """The adapter which uses `aiohttp`_ library.

    .. _aiohttp: https://docs.aiohttp.org/

    `aiohttp` supports keep-alives, persists Cookies, allows response
    compression and uses HTTP/1.1 [currently].

    ``aiohttp`` package must be installed in order to use this adapter.
    """

    is_available = aiohttp_available

    def __init__(self, *, proxies, ssl_context):
        if not aiohttp_available:
            raise ImportError(
                "`aiohttp` must be installed in order to use AioHTTPAdapter. "
                "If you have installed geopy via pip, you may use "
                "this command to install aiohttp: "
                '`pip install "geopy[aiohttp]"`.'
            )
        proxies = _normalize_proxies(proxies)
        super().__init__(proxies=proxies, ssl_context=ssl_context)

        self.proxies = proxies
        self.ssl_context = ssl_context

    @property
    def session(self):
        # Lazy session creation, which allows to avoid "unclosed socket"
        # warnings if a Geocoder instance is created without entering
        # async context and making any requests.
        session = self.__dict__.get("session")
        if session is None:
            session = aiohttp.ClientSession(
                trust_env=False,  # don't use system proxies
                raise_for_status=False
            )
            self.__dict__["session"] = session
        return session

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # Might issue a warning if loop is immediately closed:
        #   ResourceWarning: unclosed transport <_SelectorSocketTransport fd=10>
        # https://github.com/aio-libs/aiohttp/issues/1115#issuecomment-242278593
        # https://github.com/python/asyncio/issues/466
        await self.session.close()

    async def get_text(self, url, *, timeout, headers):
        with self._normalize_exceptions():
            async with self._request(url, timeout=timeout, headers=headers) as resp:
                await self._raise_for_status(resp)
                return await resp.text()

    async def get_json(self, url, *, timeout, headers):
        with self._normalize_exceptions():
            async with self._request(url, timeout=timeout, headers=headers) as resp:
                await self._raise_for_status(resp)
                try:
                    try:
                        return await resp.json()
                    except aiohttp.client_exceptions.ContentTypeError:
                        # `Attempt to decode JSON with unexpected mimetype:
                        # text/plain;charset=utf-8`
                        return json.loads(await resp.text())
                except ValueError:
                    raise GeocoderParseError(
                        "Could not deserialize using deserializer:\n%s"
                        % (await resp.text())
                    )

    async def _raise_for_status(self, resp):
        if resp.status >= 400:
            raise AdapterHTTPError(
                "Non-successful status code %s" % resp.status,
                status_code=resp.status,
                headers=resp.headers,
                text=await resp.text(),
            )

    def _request(self, url, *, timeout, headers):
        if self.proxies:
            scheme = urlparse(url).scheme
            proxy = self.proxies.get(scheme.lower())
        else:
            proxy = None

        # aiohttp accepts url as string or as yarl.URL.
        # A string url might be re-encoded by yarl, which might cause
        # a hashsum of params to change. Some geocoders use that
        # to authenticate their requests (such as Baidu SK).
        url = yarl.URL(url, encoded=True)  # `encoded` param disables url re-encoding
        return self.session.get(
            url, timeout=timeout, headers=headers, proxy=proxy, ssl=self.ssl_context
        )

    @contextlib.contextmanager
    def _normalize_exceptions(self):
        try:
            yield
        except (GeopyError, AdapterHTTPError, AssertionError):
            raise
        except Exception as error:
            message = str(error)
            if isinstance(error, asyncio.TimeoutError):
                raise GeocoderTimedOut("Service timed out")
            elif isinstance(error, SSLError):
                if "timed out" in message:
                    raise GeocoderTimedOut("Service timed out")
            elif isinstance(error, aiohttp.ClientConnectionError):
                raise GeocoderUnavailable(message)
            raise GeocoderServiceError(message) class BaseAsyncAdapter(BaseAdapter):
    """Base class for asynchronous adapters.

    See also: :ref:`Async Mode <async_mode>`.
    """

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass class RequestsAdapter(BaseSyncAdapter):
    """The adapter which uses `requests`_ library.

    .. _requests: https://requests.readthedocs.io

    `requests` supports keep-alives, retries, persists Cookies,
    allows response compression and uses HTTP/1.1 [currently].

    ``requests`` package must be installed in order to use this adapter.
    """

    is_available = requests_available

    def __init__(
        self,
        *,
        proxies,
        ssl_context,
        pool_connections=10,
        pool_maxsize=10,
        max_retries=2,
        pool_block=False
    ):
        if not requests_available:
            raise ImportError(
                "`requests` must be installed in order to use RequestsAdapter. "
                "If you have installed geopy via pip, you may use "
                "this command to install requests: "
                '`pip install "geopy[requests]"`.'
            )
        proxies = _normalize_proxies(proxies)
        super().__init__(proxies=proxies, ssl_context=ssl_context)

        self.session = requests.Session()
        self.session.trust_env = False  # don't use system proxies
        self.session.proxies = proxies

        self.session.mount(
            "http://",
            RequestsHTTPAdapter(
                pool_connections=pool_connections,
                pool_maxsize=pool_maxsize,
                max_retries=max_retries,
                pool_block=pool_block,
            ),
        )
        self.session.mount(
            "https://",
            RequestsHTTPWithSSLContextAdapter(
                ssl_context=ssl_context,
                pool_connections=pool_connections,
                pool_maxsize=pool_maxsize,
                max_retries=max_retries,
                pool_block=pool_block,
            ),
        )

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.session.close()

    def __del__(self):
        # Cleanup keepalive connections when Geocoder (and, thus, Adapter)
        # instances are getting garbage-collected.
        session = getattr(self, "session", None)
        if session is not None:
            try:
                session.close()
            except TypeError:
                # It's possible for the close method to try to fetch a
                # non-existent old_pool in urllib3 with a misleading state
                # ultimately due to stdlib queue/threading behaviour.
                # Since the error arises from a non-existent pool
                # (TypeError: 'NoneType' object is not callable)
                # it's safe to ignore this error
                pass

    def get_text(self, url, *, timeout, headers):
        resp = self._request(url, timeout=timeout, headers=headers)
        return resp.text

    def get_json(self, url, *, timeout, headers):
        resp = self._request(url, timeout=timeout, headers=headers)
        try:
            return resp.json()
        except ValueError:
            raise GeocoderParseError(
                "Could not deserialize using deserializer:\n%s" % resp.text
            )

    def _request(self, url, *, timeout, headers):
        try:
            resp = self.session.get(url, timeout=timeout, headers=headers)
        except Exception as error:
            message = str(error)
            if isinstance(error, SocketTimeout):
                raise GeocoderTimedOut("Service timed out")
            elif isinstance(error, SSLError):
                if "timed out" in message:
                    raise GeocoderTimedOut("Service timed out")
            elif isinstance(error, requests.ConnectionError):
                if "unauthorized" in message.lower():
                    raise GeocoderServiceError(message)
                else:
                    raise GeocoderUnavailable(message)
            elif isinstance(error, requests.Timeout):
                raise GeocoderTimedOut("Service timed out")
            raise GeocoderServiceError(message)
        else:
            if resp.status_code >= 400:
                raise AdapterHTTPError(
                    "Non-successful status code %s" % resp.status_code,
                    status_code=resp.status_code,
                    headers=resp.headers,
                    text=resp.text,
                )

        return resp class URLLibAdapter(BaseSyncAdapter):
    """The fallback adapter which uses urllib from the Python standard
    library, see :func:`urllib.request.urlopen`.

    urllib doesn't support keep-alives, request retries,
    doesn't persist Cookies and is HTTP/1.1 only.

    urllib was the only available option
    for making requests in geopy 1.x, so this adapter behaves the same
    as geopy 1.x in terms of HTTP requests.
    """

    def __init__(self, *, proxies, ssl_context):
        proxies = _normalize_proxies(proxies)
        super().__init__(proxies=proxies, ssl_context=ssl_context)

        # `ProxyHandler` should be present even when actually there're
        # no proxies. `build_opener` contains it anyway. By specifying
        # it here explicitly we can disable system proxies (i.e.
        # from HTTP_PROXY env var) by setting `proxies` to `{}`.
        # Otherwise, if we didn't specify ProxyHandler for empty
        # `proxies` here, the `build_opener` would have used one internally
        # which could have unwillingly picked up the system proxies.
        opener = build_opener(
            HTTPSHandler(context=ssl_context),
            ProxyHandler(proxies),
        )
        self.urlopen = opener.open

    def get_json(self, url, *, timeout, headers):
        text = self.get_text(url, timeout=timeout, headers=headers)
        try:
            return json.loads(text)
        except ValueError:
            raise GeocoderParseError(
                "Could not deserialize using deserializer:\n%s" % text
            )

    def get_text(self, url, *, timeout, headers):
        req = Request(url=url, headers=headers)
        try:
            page = self.urlopen(req, timeout=timeout)
        except Exception as error:
            message = str(error.args[0]) if len(error.args) else str(error)
            if isinstance(error, HTTPError):
                code = error.getcode()
                response_headers = {
                    name.lower(): value
                    for name, value in error.headers.items()
                }
                body = self._read_http_error_body(error)
                raise AdapterHTTPError(
                    message,
                    status_code=code,
                    headers=response_headers,
                    text=body,
                )
            elif isinstance(error, URLError):
                if "timed out" in message:
                    raise GeocoderTimedOut("Service timed out")
                elif "unreachable" in message:
                    raise GeocoderUnavailable("Service not available")
            elif isinstance(error, SocketTimeout):
                raise GeocoderTimedOut("Service timed out")
            elif isinstance(error, SSLError):
                if "timed out" in message:
                    raise GeocoderTimedOut("Service timed out")
            raise GeocoderServiceError(message)
        else:
            text = self._decode_page(page)
            status_code = page.getcode()
            if status_code >= 400:
                response_headers = {
                    name.lower(): value
                    for name, value in page.headers.items()
                }
                raise AdapterHTTPError(
                    "Non-successful status code %s" % status_code,
                    status_code=status_code,
                    headers=response_headers,
                    text=text,
                )

        return text

    def _read_http_error_body(self, error):
        try:
            return self._decode_page(error)
        except Exception:
            logger.debug(
                "Unable to fetch body for a non-successful HTTP response", exc_info=True
            )
            return None

    def _decode_page(self, page):
        encoding = page.headers.get_content_charset() or "utf-8"
        try:
            body_bytes = page.read()
        except Exception:
            raise GeocoderServiceError("Unable to read the response")

        try:
            return str(body_bytes, encoding=encoding)
        except ValueError:
            raise GeocoderParseError("Unable to decode the response bytes") class GeocoderAuthenticationFailure(GeocoderServiceError):
    """
    The remote geocoding service rejected the API key or account
    credentials this geocoder was instantiated with.
    """ class GeocoderParseError(GeocoderServiceError):
    """
    Geopy could not parse the service's response. This is probably due
    to a bug in geopy.
    """ class GeocoderServiceError(GeopyError):
    """
    There was an exception caused when calling the remote geocoding service,
    and no more specific exception could be raised by geopy. When calling
    geocoders' ``geocode`` or `reverse` methods, this is the most generic
    exception that can be raised, and any non-geopy exception will be caught
    and turned into this. The exception's message will be that of the
    original exception.
    """ class Geocoder:
    """
    Template object for geocoders.
    """

    def __init__(
            self,
            *,
            scheme=None,
            timeout=DEFAULT_SENTINEL,
            proxies=DEFAULT_SENTINEL,
            user_agent=None,
            ssl_context=DEFAULT_SENTINEL,
            adapter_factory=None
    ):
        self.scheme = scheme or options.default_scheme
        if self.scheme not in ('http', 'https'):
            raise ConfigurationError(
                'Supported schemes are `http` and `https`.'
            )
        self.timeout = (timeout if timeout is not DEFAULT_SENTINEL
                        else options.default_timeout)
        self.proxies = (proxies if proxies is not DEFAULT_SENTINEL
                        else options.default_proxies)
        self.headers = {'User-Agent': user_agent or options.default_user_agent}
        self.ssl_context = (ssl_context if ssl_context is not DEFAULT_SENTINEL
                            else options.default_ssl_context)

        if isinstance(self.proxies, str):
            self.proxies = {'http': self.proxies, 'https': self.proxies}

        if adapter_factory is None:
            adapter_factory = options.default_adapter_factory
        self.adapter = adapter_factory(
            proxies=self.proxies,
            ssl_context=self.ssl_context,
        )
        if isinstance(self.adapter, BaseSyncAdapter):
            self.__run_async = False
        elif isinstance(self.adapter, BaseAsyncAdapter):
            self.__run_async = True
        else:
            raise ConfigurationError(
                "Adapter %r must extend either BaseSyncAdapter or BaseAsyncAdapter"
                % (type(self.adapter),)
            )

    def __enter__(self):
        """Context manager for synchronous adapters. At exit all
        open connections will be closed.

        In synchronous mode context manager usage is not required,
        and connections will be automatically closed by garbage collection.
        """
        if self.__run_async:
            raise TypeError("`async with` must be used with async adapters")
        res = self.adapter.__enter__()
        assert res is self.adapter, "adapter's __enter__ must return `self`"
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.adapter.__exit__(exc_type, exc_val, exc_tb)

    async def __aenter__(self):
        """Context manager for asynchronous adapters. At exit all
        open connections will be closed.

        In asynchronous mode context manager usage is not required,
        however, it is strongly advised to avoid warnings about
        resources leaks.
        """
        if not self.__run_async:
            raise TypeError("`async with` cannot be used with sync adapters")
        res = await self.adapter.__aenter__()
        assert res is self.adapter, "adapter's __enter__ must return `self`"
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.adapter.__aexit__(exc_type, exc_val, exc_tb)

    def _coerce_point_to_string(self, point, output_format="%(lat)s,%(lon)s"):
        """
        Do the right thing on "point" input. For geocoders with reverse
        methods.
        """
        if not isinstance(point, Point):
            point = Point(point)

        # Altitude is silently dropped.
        #
        # Geocoding services (almost?) always consider only lat and lon
        # in queries, so altitude doesn't affect the request.
        # A non-zero altitude should not raise an exception
        # though, because PoIs are assumed to span the whole
        # altitude axis (i.e. not just the 0km plane).
        return output_format % dict(lat=point.latitude,
                                    lon=point.longitude)

    def _format_bounding_box(
        self, bbox, output_format="%(lat1)s,%(lon1)s,%(lat2)s,%(lon2)s"
    ):
        """
        Transform bounding box boundaries to a string matching
        `output_format` from the following formats:

            - [Point(lat1, lon1), Point(lat2, lon2)]
            - [[lat1, lon1], [lat2, lon2]]
            - ["lat1,lon1", "lat2,lon2"]

        It is guaranteed that lat1 <= lat2 and lon1 <= lon2.
        """
        if len(bbox) != 2:
            raise GeocoderQueryError("Unsupported format for a bounding box")
        p1, p2 = bbox
        p1, p2 = Point(p1), Point(p2)
        return output_format % dict(lat1=min(p1.latitude, p2.latitude),
                                    lon1=min(p1.longitude, p2.longitude),
                                    lat2=max(p1.latitude, p2.latitude),
                                    lon2=max(p1.longitude, p2.longitude))

    def _geocoder_exception_handler(self, error):
        """
        Geocoder-specific exceptions handler.
        Override if custom exceptions processing is needed.
        For example, raising an appropriate GeocoderQuotaExceeded on non-200
        response with a textual message in the body about the exceeded quota.

        Return `NONE_RESULT` to have the geocoding call return `None` (meaning
        empty result).
        """
        pass

    def _call_geocoder(
            self,
            url,
            callback,
            *,
            timeout=DEFAULT_SENTINEL,
            is_json=True,
            headers=None
    ):
        """
        For a generated query URL, get the results.
        """

        req_headers = self.headers.copy()
        if headers:
            req_headers.update(headers)

        timeout = (timeout if timeout is not DEFAULT_SENTINEL
                   else self.timeout)

        try:
            if is_json:
                result = self.adapter.get_json(url, timeout=timeout, headers=req_headers)
            else:
                result = self.adapter.get_text(url, timeout=timeout, headers=req_headers)
            if self.__run_async:
                async def fut():
                    try:
                        res = callback(await result)
                        if inspect.isawaitable(res):
                            res = await res
                        return res
                    except Exception as error:
                        res = self._adapter_error_handler(error)
                        if res is NONE_RESULT:
                            return None
                        raise

                return fut()
            else:
                return callback(result)
        except Exception as error:
            res = self._adapter_error_handler(error)
            if res is NONE_RESULT:
                return None
            raise

    def _adapter_error_handler(self, error):
        if isinstance(error, AdapterHTTPError):
            if error.text:
                logger.info(
                    'Received an HTTP error (%s): %s',
                    error.status_code,
                    error.text,
                    exc_info=False,
                )
            res = self._geocoder_exception_handler(error)
            if res is NONE_RESULT:
                return NONE_RESULT
            exc_cls = ERROR_CODE_MAP.get(error.status_code, GeocoderServiceError)
            if issubclass(exc_cls, GeocoderRateLimited):
                raise exc_cls(
                    str(error), retry_after=get_retry_after(error.headers)
                ) from error
            else:
                raise exc_cls(str(error)) from error
        else:
            res = self._geocoder_exception_handler(error)
            if res is NONE_RESULT:
                return NONE_RESULT

    # def geocode(self, query, *, exactly_one=True, timeout=DEFAULT_SENTINEL):
    #     raise NotImplementedError()

    # def reverse(self, query, *, exactly_one=True, timeout=DEFAULT_SENTINEL):
    #     raise NotImplementedError() class HttpServerThread(threading.Thread):
    spinup_timeout = 10

    def __init__(self, timeout=None):
        self.server_host = 'localhost'
        self.server_port = None  # randomly selected by OS
        self.timeout = timeout

        self.http_server = None
        self.socket_created_future = Future()

        super(HttpServerThread, self).__init__()
        self.daemon = True

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stop()
        self.join()

    def get_server_url(self):
        assert self.socket_created_future.result(self.spinup_timeout)
        return "http://%s:%s" % (self.server_host, self.server_port)

    def run(self):
        assert not self.http_server, ("This class is not reentrable. "
                                      "Please create a new instance.")

        class Server(SimpleHTTPServer.SimpleHTTPRequestHandler):
            timeout = self.timeout

            def do_GET(self):
                if self.path == "/":
                    self.send_response(200)
                    self.send_header('Connection', 'close')
                    self.end_headers()
                    self.wfile.write(b"Hello world")
                elif self.path == "/json":
                    self.send_response(200)
                    self.send_header('Content-type', 'application/json')
                    self.send_header('Connection', 'close')
                    self.end_headers()
                    self.wfile.write(b'{"hello":"world"}')
                elif self.path == "/json/plain":
                    self.send_response(200)
                    self.send_header('Content-type', 'text/plain;charset=utf-8')
                    self.send_header('Connection', 'close')
                    self.end_headers()
                    self.wfile.write(b'{"hello":"world"}')
                else:
                    self.send_response(404)
                    self.send_header('Connection', 'close')
                    self.send_header('X-test-header', 'hello')
                    self.end_headers()
                    self.wfile.write(b"Not found")
                self.connection.close()

        # ThreadingTCPServer offloads connections to separate threads, so
        # the serve_forever loop doesn't block until connection is closed
        # (unlike TCPServer). This allows to shutdown the serve_forever loop
        # even if there's an open connection.
        try:
            self.http_server = SocketServer.ThreadingTCPServer(
                (self.server_host, 0),
                Server
            )

            # don't hang if there're some open connections
            self.http_server.daemon_threads = True

            self.server_port = self.http_server.server_address[1]
        except Exception as e:
            self.socket_created_future.set_exception(e)
            raise
        else:
            self.socket_created_future.set_result(True)

        self.http_server.serve_forever()

    def stop(self):
        self.http_server.shutdown()  # stop serve_forever()
        self.http_server.server_close() class ProxyServerThread(threading.Thread):
    spinup_timeout = 10

    def __init__(self, timeout=None):
        self.proxy_host = 'localhost'
        self.proxy_port = None  # randomly selected by OS
        self.timeout = timeout

        self.proxy_server = None
        self.socket_created_future = Future()
        self.requests = []
        self.auth = None

        super().__init__()
        self.daemon = True

    def reset(self):
        self.requests.clear()
        self.auth = None

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stop()
        self.join()

    def set_auth(self, username, password):
        self.auth = "%s:%s" % (username, password)

    def get_proxy_url(self, with_scheme=True):
        assert self.socket_created_future.result(self.spinup_timeout)
        if self.auth:
            auth = "%s@" % self.auth
        else:
            auth = ""
        if with_scheme:
            scheme = "http://"
        else:
            scheme = ""
        return "%s%s%s:%s" % (scheme, auth, self.proxy_host, self.proxy_port)

    def run(self):
        assert not self.proxy_server, ("This class is not reentrable. "
                                       "Please create a new instance.")

        requests = self.requests
        proxy_thread = self

        class Proxy(SimpleHTTPServer.SimpleHTTPRequestHandler):
            timeout = self.timeout

            def check_auth(self):
                if proxy_thread.auth is not None:
                    auth_header = self.headers.get('Proxy-Authorization')
                    b64_auth = base64.standard_b64encode(
                        proxy_thread.auth.encode()
                    ).decode()
                    expected_auth = "Basic %s" % b64_auth
                    if auth_header != expected_auth:
                        self.send_response(401)
                        self.send_header('Connection', 'close')
                        self.end_headers()
                        self.wfile.write(
                            (
                                "not authenticated. Expected %r, received %r"
                                % (expected_auth, auth_header)
                            ).encode()
                        )
                        self.connection.close()
                        return False
                return True

            def do_GET(self):
                if not self.check_auth():
                    return
                requests.append(self.path)

                req = urlopen(self.path, timeout=self.timeout)
                self.send_response(req.getcode())
                content_type = req.info().get('content-type', None)
                if content_type:
                    self.send_header('Content-Type', content_type)
                self.send_header('Connection', 'close')
                self.end_headers()
                self.copyfile(req, self.wfile)
                self.connection.close()
                req.close()

            def do_CONNECT(self):
                if not self.check_auth():
                    return
                requests.append(self.path)

                # Make a raw TCP connection to the target server
                host, port = self.path.split(':')
                try:
                    addr = host, int(port)
                    other_connection = \
                        socket.create_connection(addr, timeout=self.timeout)
                except socket.error:
                    self.send_error(502, 'Bad gateway')
                    return

                # Respond that a tunnel has been created
                self.send_response(200)
                self.send_header('Connection', 'close')
                self.end_headers()
                pipe_sockets(self.connection,  # it closes sockets
                             other_connection, self.timeout)

        # ThreadingTCPServer offloads connections to separate threads, so
        # the serve_forever loop doesn't block until connection is closed
        # (unlike TCPServer). This allows to shutdown the serve_forever loop
        # even if there's an open connection.
        try:
            self.proxy_server = SocketServer.ThreadingTCPServer(
                (self.proxy_host, 0),
                Proxy
            )

            # don't hang if there're some open connections
            self.proxy_server.daemon_threads = True

            self.proxy_port = self.proxy_server.server_address[1]
        except Exception as e:
            self.socket_created_future.set_exception(e)
            raise
        else:
            self.socket_created_future.set_result(True)

        self.proxy_server.serve_forever()

    def stop(self):
        self.proxy_server.shutdown()  # stop serve_forever()
        self.proxy_server.server_close()
import os
import ssl
import pytest
import geopy.geocoders
    import requests  # noqa
    import aiohttp  # noqa
from unittest.mock import patch
from urllib.parse import urljoin
from urllib.request import getproxies, urlopen
from async_generator import async_generator, asynccontextmanager, yield_
from geopy.adapters import (
    AdapterHTTPError,
    AioHTTPAdapter,
    BaseAsyncAdapter,
    RequestsAdapter,
    URLLibAdapter,
)
from geopy.exc import (
    GeocoderAuthenticationFailure,
    GeocoderParseError,
    GeocoderServiceError,
)
from geopy.geocoders.base import Geocoder
from test.proxy_server import HttpServerThread, ProxyServerThread
    return http_server.get_server_url()


@pytest.fixture
def remote_website_http_json(remote_website_http):
    return urljoin(remote_website_http, "/json")


@pytest.fixture
def remote_website_http_json_plain(remote_website_http):
    return urljoin(remote_website_http, "/json/plain")


@pytest.fixture
def remote_website_http_404(remote_website_http):
    return urljoin(remote_website_http, "/404")


@pytest.fixture(params=AVAILABLE_ADAPTERS, autouse=True)
def adapter_factory(request):
    adapter_factory = request.param
    with patch.object(
        geopy.geocoders.options, "default_adapter_factory", adapter_factory
    ):
        yield adapter_factory


@asynccontextmanager
@async_generator
async def make_dummy_async_geocoder(**kwargs):
    geocoder = DummyGeocoder(**kwargs)
    run_async = isinstance(geocoder.adapter, BaseAsyncAdapter)
    if run_async:
        async with geocoder:
            await yield_(geocoder)
    else:
        orig_geocode = geocoder.geocode

        async def geocode(*args, **kwargs):
            return orig_geocode(*args, **kwargs)

        geocoder.geocode = geocode
        await yield_(geocoder)


@pytest.mark.parametrize("adapter_cls", NOT_AVAILABLE_ADAPTERS)
async def test_not_available_adapters_raise(adapter_cls):
    # Note: this test is uselessly parametrized with `adapter_factory`.
    with patch.object(
        geopy.geocoders.options, "default_adapter_factory", adapter_cls
    ):
        with pytest.raises(ImportError):
            async with make_dummy_async_geocoder():
                pass


async def test_geocoder_constructor_uses_https_proxy(
    timeout, proxy_server, proxy_url, remote_website_trusted_https
):
    base_http = urlopen(remote_website_trusted_https, timeout=timeout)
    base_html = base_http.read().decode()

    async with make_dummy_async_geocoder(
        proxies={"https": proxy_url}, timeout=timeout
    ) as geocoder_dummy:
        assert 0 == len(proxy_server.requests)
        assert base_html == await geocoder_dummy.geocode(remote_website_trusted_https)
        assert 1 == len(proxy_server.requests)


@pytest.mark.parametrize("with_scheme", [False, True])
async def test_geocoder_http_proxy_auth_is_respected(
    timeout, proxy_server, remote_website_http, with_scheme
):
    proxy_server.set_auth("user", "test")
    base_http = urlopen(remote_website_http, timeout=timeout)
    base_html = base_http.read().decode()

    proxy_url = proxy_server.get_proxy_url(with_scheme=with_scheme)

    async with make_dummy_async_geocoder(
        proxies={"http": proxy_url}, timeout=timeout
    ) as geocoder_dummy:
        assert 0 == len(proxy_server.requests)
        assert base_html == await geocoder_dummy.geocode(remote_website_http)
        assert 1 == len(proxy_server.requests)


@pytest.mark.parametrize("with_scheme", [False, True])
async def test_geocoder_https_proxy_auth_is_respected(
    timeout, proxy_server, remote_website_trusted_https, with_scheme
):
    proxy_server.set_auth("user", "test")
    base_http = urlopen(remote_website_trusted_https, timeout=timeout)
    base_html = base_http.read().decode()

    proxy_url = proxy_server.get_proxy_url(with_scheme=with_scheme)

    async with make_dummy_async_geocoder(
        proxies={"https": proxy_url}, timeout=timeout
    ) as geocoder_dummy:
        assert 0 == len(proxy_server.requests)
        assert base_html == await geocoder_dummy.geocode(remote_website_trusted_https)
        assert 1 == len(proxy_server.requests)


async def test_geocoder_http_proxy_auth_error(
    timeout, proxy_server, proxy_url, remote_website_http
):
    # Set up proxy auth but query it without auth:
    proxy_server.set_auth("user", "test")

    async with make_dummy_async_geocoder(
        proxies={"http": proxy_url}, timeout=timeout
    ) as geocoder_dummy:
        # For HTTP targets we cannot distinguish between 401 from proxy
        # and 401 from geocoding service, thus the same error as for
        # 401 for geocoders is raised: GeocoderAuthenticationFailure
        with pytest.raises(GeocoderAuthenticationFailure) as exc_info:
            await geocoder_dummy.geocode(remote_website_http)

        assert exc_info.type is GeocoderAuthenticationFailure