# coding: utf-8

"""
    Kubernetes

    No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)

    The version of the OpenAPI document: release-1.37
    Generated by OpenAPI Generator (https://openapi-generator.tech)

    Do not edit the class manually.
"""  # noqa: E501


import ipaddress
import io
import json
import re
import ssl
from urllib.parse import urlparse

import urllib3
from urllib3.util.retry import Retry

from kubernetes.utils.retry import (
    is_retry_after_response,
    on_retry_after_error,
    retry_after_backoff,
)
from kubernetes.utils.keepalive import tcp_keepalive_socket_options
from kubernetes.client.exceptions import ApiException, ApiValueError

SUPPORTED_SOCKS_PROXIES = {"socks5", "socks5h", "socks4", "socks4a"}
RESTResponseType = urllib3.HTTPResponse


def is_socks_proxy_url(url):
    if url is None:
        return False
    split_section = url.split("://")
    if len(split_section) < 2:
        return False
    else:
        return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES

def contenttype_matches(contenttype, maintype, subtype):
    """Matches the given contenttype against the given type and subtype

    :param contenttype: the content type to match
    :param maintype: the expected maintype
    :param subtype: the expected subtype
    :return: `true` when the given content type matches the given type and subtype,
        regardless of the presence of mime type parameters, otherwise returns `false`.
    :rtype: bool
    """
    pattern = '{type}/(?:[^+;]+\\+)?{subtype}(?:[ \t]*;.*)?'.format(
        type = re.escape(maintype),
        subtype = re.escape(subtype),
    )
    return re.fullmatch(pattern, contenttype, re.IGNORECASE) is not None

def should_bypass_proxies(url: str, no_proxy: str) -> bool:
    """Return whether ``url`` matches the comma-separated ``no_proxy`` rules."""
    parsed_url = urlparse(url)
    if not parsed_url.hostname:
        return True

    host = parsed_url.hostname.lower()
    host_and_port = parsed_url.netloc.lower()
    try:
        host_ip = ipaddress.ip_address(host)
    except ValueError:
        host_ip = None

    for entry in (entry.strip().lower() for entry in no_proxy.split(',')):
        if not entry:
            continue
        if entry == '*':
            return True

        if host_ip is not None:
            try:
                if host_ip in ipaddress.ip_network(entry, strict=False):
                    return True
            except ValueError:
                pass

        entry = entry.lstrip('.')
        if (
            host == entry
            or host.endswith('.' + entry)
            or host_and_port == entry
            or host_and_port.endswith('.' + entry)
        ):
            return True

    return False


class RESTResponse(io.IOBase):

    def __init__(self, resp) -> None:
        self.response = resp
        self.status = resp.status
        self.reason = resp.reason
        self.data = None

    def read(self):
        if self.data is None:
            self.data = self.response.data
        return self.data

    @property
    def headers(self):
        """Returns a dictionary of response headers."""
        return self.response.headers

    def getheaders(self):
        """Returns a dictionary of the response headers; use ``headers`` instead."""
        return self.response.headers

    def getheader(self, name, default=None):
        """Returns a given response header; use ``headers.get()`` instead."""
        return self.response.headers.get(name, default)


class RESTClientObject:

    def __init__(self, configuration) -> None:
        self.configuration = configuration

        # urllib3.PoolManager will pass all kw parameters to connectionpool
        # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75  # noqa: E501
        # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680  # noqa: E501
        # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html  # noqa: E501

        # cert_reqs
        if configuration.verify_ssl:
            cert_reqs = ssl.CERT_REQUIRED
        else:
            cert_reqs = ssl.CERT_NONE

        pool_args = {
            "cert_reqs": cert_reqs,
            "ca_certs": configuration.ssl_ca_cert,
            "cert_file": configuration.cert_file,
            "key_file": configuration.key_file,
            "ca_cert_data": configuration.ca_cert_data,
        }
        if configuration.assert_hostname is not None:
            pool_args['assert_hostname'] = (
                configuration.assert_hostname
            )

        if configuration.retries is not None:
            pool_args['retries'] = configuration.retries

        if configuration.tls_server_name:
            pool_args['server_hostname'] = configuration.tls_server_name


        if configuration.socket_options is not None:
            pool_args['socket_options'] = configuration.socket_options
        elif getattr(configuration, 'keep_alive', False):
            pool_args['socket_options'] = tcp_keepalive_socket_options()

        if configuration.connection_pool_maxsize is not None:
            pool_args['maxsize'] = configuration.connection_pool_maxsize

        # https pool manager
        self.pool_manager: urllib3.PoolManager

        if configuration.proxy and not should_bypass_proxies(
            configuration.host, configuration.no_proxy or ''
        ):
            if is_socks_proxy_url(configuration.proxy):
                from urllib3.contrib.socks import SOCKSProxyManager
                pool_args["proxy_url"] = configuration.proxy
                pool_args["headers"] = configuration.proxy_headers
                self.pool_manager = SOCKSProxyManager(**pool_args)
            else:
                pool_args["proxy_url"] = configuration.proxy
                pool_args["proxy_headers"] = configuration.proxy_headers
                if configuration.proxy_ssl_context is not None:
                    pool_args["proxy_ssl_context"] = configuration.proxy_ssl_context
                self.pool_manager = urllib3.ProxyManager(**pool_args)
        else:
            self.pool_manager = urllib3.PoolManager(**pool_args)

    def close(self) -> None:
        self.pool_manager.clear()

    def request(
        self,
        method,
        url,
        headers=None,
        body=None,
        post_params=None,
        _request_timeout=None
    ):
        """Perform requests.

        :param method: http request method
        :param url: http request url
        :param headers: http request headers
        :param body: request json body, for `application/json`
        :param post_params: request post parameters,
                            `application/x-www-form-urlencoded`
                            and `multipart/form-data`
        :param _request_timeout: timeout setting for this request. If one
                                 number provided, it will be total request
                                 timeout. It can also be a pair (tuple) of
                                 (connection, read) timeouts.
        """
        method = method.upper()
        assert method in [
            'GET',
            'HEAD',
            'DELETE',
            'POST',
            'PUT',
            'PATCH',
            'OPTIONS'
        ]

        if post_params and body:
            raise ApiValueError(
                "body parameter cannot be used with post_params parameter."
            )

        post_params = post_params or {}
        headers = headers or {}

        timeout = None
        if _request_timeout:
            if isinstance(_request_timeout, (int, float)):
                timeout = urllib3.Timeout(total=_request_timeout)
            elif (
                    isinstance(_request_timeout, tuple)
                    and len(_request_timeout) == 2
                ):
                timeout = urllib3.Timeout(
                    connect=_request_timeout[0],
                    read=_request_timeout[1]
                )

        client_go_read_retries = (
            method in ['GET', 'HEAD']
            and getattr(self.configuration, 'client_go_retries', False)
        )
        read_retries = None
        if client_go_read_retries:
            read_retries = self._urllib3_retries_without_status(
                getattr(self.configuration, 'retries', None))

        def read_request(check_retry_status=False):
            kwargs = {}
            if read_retries is not None:
                kwargs['retries'] = read_retries
            response = self.pool_manager.request(
                method,
                url,
                fields={},
                timeout=timeout,
                headers=headers,
                preload_content=False,
                **kwargs
            )
            if check_retry_status:
                self._raise_retry_after_response(response)
            return response

        try:
            # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
            if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:

                content_type = headers.get('Content-Type')
                # JSON Patch is a list of operations; object patches for built-in
                # APIs use strategic merge:
                # https://kubernetes.io/docs/reference/using-api/api-concepts/#updates-to-existing-resources
                if (
                    content_type == 'application/json-patch+json'
                    and not isinstance(body, list)
                ):
                    headers['Content-Type'] = content_type = \
                        'application/strategic-merge-patch+json'
                is_json = (
                    not content_type
                    or contenttype_matches(content_type, 'application', 'json')
                )
                # JSON is valid YAML 1.2, so structured YAML bodies can use
                # the existing JSON serializer:
                # https://yaml.org/spec/1.2.2/#13-relation-to-json
                is_structured_yaml = (
                    content_type
                    and (
                        contenttype_matches(content_type, 'application', 'yaml')
                        or contenttype_matches(content_type, 'text', 'yaml')
                        or contenttype_matches(content_type, 'text', 'x-yaml')
                    )
                    and not isinstance(body, (str, bytes))
                )
                if is_json or is_structured_yaml:
                    request_body = None
                    if body is not None:
                        request_body = json.dumps(body)
                    r = self.pool_manager.request(
                        method,
                        url,
                        body=request_body,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                elif contenttype_matches(content_type, 'application', 'x-www-form-urlencoded'):
                    r = self.pool_manager.request(
                        method,
                        url,
                        fields=post_params,
                        encode_multipart=False,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                elif contenttype_matches(content_type, 'multipart', 'form-data'):
                    # must del headers['Content-Type'], or the correct
                    # Content-Type which generated by urllib3 will be
                    # overwritten.
                    del headers['Content-Type']
                    # Ensures that dict objects are serialized
                    post_params = [(a, json.dumps(b)) if isinstance(b, dict) else (a,b) for a, b in post_params]
                    r = self.pool_manager.request(
                        method,
                        url,
                        fields=post_params,
                        encode_multipart=True,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                # Pass a `string` parameter directly in the body to support
                # other content types than JSON when `body` argument is
                # provided in serialized form.
                elif isinstance(body, str) or isinstance(body, bytes):
                    r = self.pool_manager.request(
                        method,
                        url,
                        body=body,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                elif content_type.startswith('text/') and isinstance(body, bool):
                    request_body = "true" if body else "false"
                    r = self.pool_manager.request(
                        method,
                        url,
                        body=request_body,
                        preload_content=False,
                        timeout=timeout,
                        headers=headers)
                else:
                    # Cannot generate the request from given parameters
                    msg = """Cannot prepare a request message for provided
                             arguments. Please check that your arguments match
                             declared content type."""
                    raise ApiException(status=0, reason=msg)
            # For `GET`, `HEAD`
            else:
                if client_go_read_retries:
                    backoff = retry_after_backoff(
                        getattr(self.configuration, 'retries', None),
                        getattr(self.configuration, 'client_go_retry_backoff', None),
                    )
                    r = on_retry_after_error(
                        backoff, self._is_read_retryable,
                        lambda: read_request(True))
                else:
                    r = read_request()
        except urllib3.exceptions.SSLError as e:
            msg = "\n".join([type(e).__name__, str(e)])
            raise ApiException(status=0, reason=msg)

        return RESTResponse(r)

    @classmethod
    def _is_read_retryable(cls, error):
        return is_retry_after_response(error)

    @staticmethod
    def _retry_after_error(response):
        error = ApiException(status=response.status, reason=response.reason)
        error.headers = response.getheaders()
        return error

    @classmethod
    def _raise_retry_after_response(cls, response):
        error = cls._retry_after_error(response)
        if not is_retry_after_response(error):
            return
        cls._read_and_close_retry_response(response)
        raise error

    @staticmethod
    def _read_and_close_retry_response(response):
        try:
            try:
                content_length = int(response.getheaders().get(
                    'Content-Length', '-1'))
            except (TypeError, ValueError):
                content_length = -1
            if content_length <= 2 << 10:
                response.read(2 << 10)
        finally:
            response.close()

    @staticmethod
    def _urllib3_retries_without_status(retries):
        if retries is False:
            return False
        if retries is None:
            retries = Retry.DEFAULT
        elif retries is True:
            retries = Retry.DEFAULT
        elif isinstance(retries, int):
            retries = Retry.from_int(retries)
        if isinstance(retries, Retry):
            return retries.new(
                status=0,
                status_forcelist=(),
                respect_retry_after_header=False,
            )
        return retries