Refactor Patroni REST API communication (#1197)

* make it possible to use client certificates with REST API
* define a separate PatroniRequest class which handles all communication
* refactor patronictl to use the new class
* make Ha to use the new class instead of calling requests.get. The old call wasn't taking into account certificates and basic-auth

Close #898

docs/ENVIRONMENT.rst

@@ -117,6 +117,15 @@ REST API
 -  **PATRONI\_RESTAPI\_PASSWORD**: Basic-auth password to protect unsafe REST API endpoints.
 -  **PATRONI\_RESTAPI\_CERTFILE**: Specifies the file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL.
 -  **PATRONI\_RESTAPI\_KEYFILE**: Specifies the file with the secret key in the PEM format.
+-  **PATRONI\_RESTAPI\_CAFILE**: Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs.
+-  **PATRONI\_RESTAPI\_VERIFY\_CLIENT**: ``none``, ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. If ``verify_client`` is set to ``optional`` or ``required`` basic-auth is not checked.
+
+CTL
+---
+-  **PATRONI\_CTL\_INSECURE**: Allow connections to REST API without verifying SSL certs.
+-  **PATRONI\_CTL\_CACERT**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter.
+-  **PATRONI\_CTL\_CERTFILE**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter.
+-  **PATRONI\_CTL\_KEYFILE**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter.
 
 ZooKeeper
 ---------

docs/SETTINGS.rst

@@ -152,12 +152,12 @@ PostgreSQL
     -  **rewind**:
         -  **username**: name for the user for ``pg_rewind``; the user will be created during initialization of postgres 11+ and all necessary `permissions <https://www.postgresql.org/docs/11/app-pgrewind.html#id-1.9.5.8.8>`__ will be granted.
         -  **password**: password for the user for ``pg_rewind``; the user will be created during initialization.
--  **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.)
         -  **sslmode**: (optional) maps to the `sslmode <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLMODE>`__ connection parameter, which allows a client to specify the type of TLS negotiation mode with the server. For more information on how each mode works, please visit the `PostgreSQL documentation <https://www.postgresql.org/docs/current/libpq-ssl.html#LIBPQ-SSL-SSLMODE-STATEMENTS>`__. The default mode is ``prefer``.
         -  **sslkey**: (optional) maps to the `sslkey <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLKEY>`__ connection parameter, which specifies the location of the secret key used with the client's certificate.
         -  **sslcert**: (optional) maps to the `sslcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCERT>`__ connection parameter, which specifies the location of the client certificate.
         -  **sslrootcert**: (optional) maps to the `sslrootcert <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLROOTCERT>`__ connection parameter, which specifies the location of a file containing one ore more certificate authorities (CA) certificates that the client will use to verify a server's certificate.
         -  **sslcrl**: (optional) maps to the `sslcrl <https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNECT-SSLCRL>`__ connection parameter, which specifies the location of a file containing a certificate revocation list. A client will reject connecting to any server that has a certificate present in this list.
+-  **callbacks**: callback scripts to run on certain actions. Patroni will pass the action, role and cluster name. (See scripts/aws.py as an example of how to write them.)
         -  **on\_reload**: run this script when configuration reload is triggered.
         -  **on\_restart**: run this script when the postgres restarts (without changing role).
         -  **on\_role\_change**: run this script when the postgres is being promoted or demoted.
@@ -199,9 +199,10 @@ REST API
             -  **username**: Basic-auth username to protect unsafe REST API endpoints.
             -  **password**: Basic-auth password to protect unsafe REST API endpoints.
 
-        -  **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs.
         -  **certfile**: Specifies the file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL.
         -  **keyfile**: Specifies the file with the secret key in the PEM format.
+        -  **cafile**: Specifies the file with the CA_BUNDLE with certificates of trusted CAs to use while verifying client certs.
+        -  **verify\_client**: ``none``, ``optional`` or ``required``. When ``none`` REST API will not check client certificates. When ``required`` client certificates are required for all REST API calls. When ``optional`` client certificates are required for all unsafe REST API endpoints. If ``verify_client`` is set to ``optional`` or ``required`` basic-auth is not checked.
 
 .. _patronictl_settings:
 
@@ -209,8 +210,9 @@ CTL
 ---
 -  **Optional**:
     -  **insecure**: Allow connections to REST API without verifying SSL certs.
-    -  **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cacert" parameter.
-    -  **certfile**: Specifies the file with the certificate in the PEM format to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "certfile" parameter.
+    -  **cacert**: Specifies the file with the CA_BUNDLE file or directory with certificates of trusted CAs to use while verifying REST API SSL certs. If not provided patronictl will use the value provided for REST API "cafile" parameter.
+    -  **certfile**: Specifies the file with the client certificate in the PEM format. If not provided patronictl will use the value provided for REST API "certfile" parameter.
+    -  **keyfile**: Specifies the file with the client secret key in the PEM format. If not provided patronictl will use the value provided for REST API "keyfile" parameter.
 
 ZooKeeper
 ----------

patroni/__init__.py

@@ -16,6 +16,7 @@ class Patroni(object):
         from patroni.ha import Ha
         from patroni.log import PatroniLogger
         from patroni.postgresql import Postgresql
+        from patroni.request import PatroniRequest
         from patroni.version import __version__
         from patroni.watchdog import Watchdog
 
@@ -31,6 +32,7 @@ class Patroni(object):
 
         self.postgresql = Postgresql(self.config['postgresql'])
         self.api = RestApiServer(self, self.config['restapi'])
+        self.request = PatroniRequest(self.config, True)
         self.ha = Ha(self)
 
         self.tags = self.get_tags()
@@ -71,6 +73,8 @@ class Patroni(object):
             self.tags = self.get_tags()
             self.logger.reload_config(self.config.get('log', {}))
             self.watchdog.reload_config(self.config)
+            if sighup:
+                self.request.reload_config(self.config)
             self.api.reload_config(self.config['restapi'])
             self.postgresql.reload_config(self.config['postgresql'], sighup)
             self.dcs.reload_config(self.config)

patroni/api.py

@@ -7,6 +7,7 @@ import traceback
 import dateutil.parser
 import datetime
 import os
+import six
 import socket
 
 from patroni.postgresql import PostgresConnectionException
@@ -20,20 +21,6 @@ from threading import Thread
 logger = logging.getLogger(__name__)
 
 
-def check_auth(func):
-    """Decorator function to check authorization header.
-
-    Usage example:
-    @check_auth
-    def do_PUT_foo():
-        pass
-    """
-    def wrapper(handler, *args, **kwargs):
-        if handler.check_auth_header():
-            return func(handler, *args, **kwargs)
-    return wrapper
-
-
 class RestApiHandler(BaseHTTPRequestHandler):
 
     def _write_response(self, status_code, body, content_type='text/html', headers=None):
@@ -49,14 +36,20 @@ class RestApiHandler(BaseHTTPRequestHandler):
     def _write_json_response(self, status_code, response):
         self._write_response(status_code, json.dumps(response), content_type='application/json')
 
-    def send_auth_request(self, body):
-        headers = {'WWW-Authenticate': 'Basic realm="' + self.server.patroni.__class__.__name__ + '"'}
-        self._write_response(401, body, headers=headers)
+    def check_auth(func):
+        """Decorator function to check authorization header or client certificates
 
-    def check_auth_header(self):
-        auth_header = self.headers.get('Authorization')
-        status = self.server.check_auth_header(auth_header)
-        return not status or self.send_auth_request(status)
+        Usage example:
+        @check_auth
+        def do_PUT_foo():
+            pass
+        """
+
+        def wrapper(self, *args, **kwargs):
+            if self.server.check_auth(self):
+                return func(self, *args, **kwargs)
+
+        return wrapper
 
     def _write_status_response(self, status_code, response):
         patroni = self.server.patroni
@@ -488,8 +481,8 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
     def __init__(self, patroni, config):
         self.patroni = patroni
         self.__listen = None
-        self.__initialize(config)
-        self.__set_config_parameters(config)
+        self.__ssl_options = None
+        self.reload_config(config)
         self.daemon = True
 
     def query(self, sql, *params):
@@ -520,13 +513,16 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
             if not auth_header.startswith('Basic ') or not self.check_basic_auth_key(auth_header[6:]):
                 return 'not authenticated'
 
-    @staticmethod
-    def __get_ssl_options(config):
-        return {option: config[option] for option in ['certfile', 'keyfile'] if option in config}
+    def check_auth(self, rh):
+        if not hasattr(rh.request, 'getpeercert') or not rh.request.getpeercert():  # valid client cert isn't present
+            if self.__protocol == 'https' and self.__ssl_options.get('verify_client') in ('required', 'optional'):
+                return rh._write_response(403, 'client certificate required')
 
-    def __set_config_parameters(self, config):
-        self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None
-        self.connection_string = uri(self.__protocol, config.get('connect_address') or self.__listen, 'patroni')
+            reason = self.check_auth_header(rh.headers.get('Authorization'))
+            if reason:
+                headers = {'WWW-Authenticate': 'Basic realm="' + self.patroni.__class__.__name__ + '"'}
+                return rh._write_response(401, reason, headers=headers)
+        return True
 
     @staticmethod
     def __has_dual_stack():
@@ -545,7 +541,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
 
     def __httpserver_init(self, host, port):
         dual_stack = self.__has_dual_stack()
-        if host == '':
+        if host in ('', '*'):
             host = None
 
         info = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE)
@@ -555,43 +551,56 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
         self.address_family = info[0][0]
         HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler)
 
-    def __initialize(self, config):
+    def __initialize(self, listen, ssl_options):
         try:
-            host, port = split_host_port(config['listen'], None)
+            host, port = split_host_port(listen, None)
         except Exception:
             raise ValueError('Invalid "restapi" config: expected <HOST>:<PORT> for "listen", but got "{0}"'
-                             .format(config['listen']))
+                             .format(listen))
 
-        if self.__listen is not None:  # changing config in runtime
+        reloading_config = self.__listen is not None  # changing config in runtime
+        if reloading_config:
             self.shutdown()
 
-        self.__listen = config['listen']
-        self.__ssl_options = self.__get_ssl_options(config)
+        self.__listen = listen
+        self.__ssl_options = ssl_options
 
         self.__httpserver_init(host, port)
         Thread.__init__(self, target=self.serve_forever)
         self._set_fd_cloexec(self.socket)
 
-        self.__protocol = 'http'
-
         # wrap socket with ssl if 'certfile' is defined in a config.yaml
         # Sometime it's also needed to pass reference to a 'keyfile'.
-        if self.__ssl_options.get('certfile'):
+        self.__protocol = 'https' if ssl_options.get('certfile') else 'http'
+        if self.__protocol == 'https':
             import ssl
-            ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
-            ctx.load_cert_chain(**self.__ssl_options)
+            ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH, cafile=ssl_options.get('cafile'))
+            ctx.load_cert_chain(certfile=ssl_options['certfile'], keyfile=ssl_options.get('keyfile'))
+            verify_client = ssl_options.get('verify_client')
+            if verify_client:
+                modes = {'none': ssl.CERT_NONE, 'optional': ssl.CERT_OPTIONAL, 'required': ssl.CERT_REQUIRED}
+                if verify_client in modes:
+                    ctx.verify_mode = modes[verify_client]
+                else:
+                    logger.error('Bad value in the "restapi.verify_client": %s', verify_client)
             self.socket = ctx.wrap_socket(self.socket, server_side=True)
-            self.__protocol = 'https'
-        return True
+        if reloading_config:
+            self.start()
 
     def reload_config(self, config):
         if 'listen' not in config:  # changing config in runtime
             raise ValueError('Can not find "restapi.listen" config')
 
-        elif (self.__listen != config['listen'] or self.__ssl_options != self.__get_ssl_options(config)) \
-                and self.__initialize(config):
-            self.start()
-        self.__set_config_parameters(config)
+        ssl_options = {n: config[n] for n in ('certfile', 'keyfile', 'cafile') if n in config}
+
+        if isinstance(config.get('verify_client'), six.string_types):
+            ssl_options['verify_client'] = config['verify_client'].lower()
+
+        if self.__listen != config['listen'] or self.__ssl_options != ssl_options:
+            self.__initialize(config['listen'], ssl_options)
+
+        self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None
+        self.connection_string = uri(self.__protocol, config.get('connect_address') or self.__listen, 'patroni')
 
     @staticmethod
     def handle_error(request, client_address):

patroni/config.py

@@ -236,7 +236,8 @@ class Config(object):
                 if value:
                     ret[section][param] = value
 
-        _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile'])
+        _set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile', 'cafile', 'verify_client'])
+        _set_section_values('ctl', ['insecure', 'cacert', 'certfile', 'keyfile'])
         _set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir'])
         _set_section_values('log', ['level', 'format', 'dateformat', 'max_queue_size',
                                     'dir', 'file_size', 'file_num', 'loggers'])

patroni/ctl.py

@@ -2,7 +2,6 @@
 Patroni Control
 '''
 
-import base64
 import click
 import codecs
 import datetime
@@ -15,7 +14,6 @@ import json
 import logging
 import os
 import random
-import requests
 import subprocess
 import sys
 import tempfile
@@ -30,6 +28,7 @@ from patroni.dcs import get_dcs as _get_dcs
 from patroni.exceptions import PatroniException
 from patroni.postgresql import Postgresql
 from patroni.postgresql.misc import postgres_version_to_int
+from patroni.request import PatroniRequest
 from patroni.utils import patch_config, polling_loop
 from patroni.version import __version__
 from prettytable import PrettyTable
@@ -135,36 +134,12 @@ def get_dcs(config, scope):
         raise PatroniCtlException(str(e))
 
 
-def auth_header(config):
-    if config.get('restapi', {}).get('auth', ''):
-        return {'Authorization': 'Basic ' + base64.b64encode(config['restapi']['auth'].encode('utf-8')).decode('utf-8')}
-
-
-def request_patroni(member, request_type, endpoint, content=None, headers=None):
+def request_patroni(member, method='GET', endpoint=None, data=None):
     ctx = click.get_current_context()  # the current click context
-    headers = headers or {}
-    url_parts = urlparse(member.api_url)
-    logging.debug(url_parts)
-    if 'Content-Type' not in headers:
-        headers['Content-Type'] = 'application/json'
-
-    url = '{0}://{1}/{2}'.format(url_parts.scheme, url_parts.netloc, endpoint)
-
-    insecure = ctx.obj.get('ctl', {}).get('insecure', False)
-    # Get certfile if any from several configuration namespace
-    cert = ctx.obj.get('ctl', {}).get('cacert') or \
-        ctx.obj.get('restapi', {}).get('cacert') or \
-        ctx.obj.get('restapi', {}).get('certfile')
-    # In the case we specificaly disable SSL cert verification we don't want to have the warning
-    if insecure:
-        verify = False
-    elif cert:
-        verify = cert
-    else:
-        verify = True
-    return getattr(requests, request_type)(url, headers=headers,
-                                           data=json.dumps(content) if content else None, timeout=60,
-                                           verify=verify)
+    request_executor = ctx.obj.get('__request_patroni')
+    if not request_executor:
+        request_executor = ctx.obj['__request_patroni'] = PatroniRequest(ctx.obj)
+    return request_executor(member, method, endpoint, data)
 
 
 def print_output(columns, rows=None, alignment=None, fmt='pretty', header=True, delimiter='\t'):
@@ -457,9 +432,9 @@ def remove(obj, cluster_name, fmt):
 
 
 def check_response(response, member_name, action_name, silent_success=False):
-    if response.status_code >= 400:
+    if response.status >= 400:
         click.echo('Failed: {0} for member {1}, status code={2}, ({3})'.format(
-            action_name, member_name, response.status_code, response.text
+            action_name, member_name, response.status, response.data.decode('utf-8')
         ))
         return False
     elif not silent_success:
@@ -493,18 +468,17 @@ def reload(obj, cluster_name, member_names, force, role):
 
     members = get_members(cluster, cluster_name, member_names, role, force, 'reload')
 
-    content = {}
     for member in members:
-        r = request_patroni(member, 'post', 'reload', content, auth_header(obj))
-        if r.status_code == 200:
+        r = request_patroni(member, 'post', 'reload')
+        if r.status == 200:
             click.echo('No changes to apply on member {0}'.format(member.name))
-        elif r.status_code == 202:
+        elif r.status == 202:
             click.echo('Reload request received for member {0} and will be processed within {1} seconds'.format(
                 member.name, cluster.config.data.get('loop_wait'))
             )
         else:
             click.echo('Failed: reload for member {0}, status code={1}, ({2})'.format(
-                member.name, r.status_code, r.text)
+                member.name, r.status, r.data.decode('utf-8'))
             )
 
 
@@ -565,19 +539,19 @@ def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, vers
     for member in members:
         if 'schedule' in content:
             if force and member.data.get('scheduled_restart'):
-                r = request_patroni(member, 'delete', 'restart', headers=auth_header(obj))
+                r = request_patroni(member, 'delete', 'restart')
                 check_response(r, member.name, 'flush scheduled restart', True)
 
-        r = request_patroni(member, 'post', 'restart', content, auth_header(obj))
-        if r.status_code == 200:
+        r = request_patroni(member, 'post', 'restart', content)
+        if r.status == 200:
             click.echo('Success: restart on member {0}'.format(member.name))
-        elif r.status_code == 202:
+        elif r.status == 202:
             click.echo('Success: restart scheduled on member {0}'.format(member.name))
-        elif r.status_code == 409:
+        elif r.status == 409:
             click.echo('Failed: another restart is already scheduled on member {0}'.format(member.name))
         else:
             click.echo('Failed: restart for member {0}, status code={1}, ({2})'.format(
-                member.name, r.status_code, r.text)
+                member.name, r.status, r.data.decode('utf-8'))
             )
 
 
@@ -593,8 +567,8 @@ def reinit(obj, cluster_name, member_names, force):
     for member in members:
         body = {'force': force}
         while True:
-            r = request_patroni(member, 'post', 'reinitialize', body, auth_header(obj))
-            if not check_response(r, member.name, 'reinitialize') and r.text.endswith(' already in progress') \
+            r = request_patroni(member, 'post', 'reinitialize', body)
+            if not check_response(r, member.name, 'reinitialize') and r.data.endswith(b' already in progress') \
                     and not force and click.confirm('Do you want to cancel it and reinitialize anyway?'):
                 body['force'] = True
                 continue
@@ -682,19 +656,19 @@ def _do_failover_or_switchover(obj, action, cluster_name, master, candidate, for
     try:
         member = cluster.leader.member if cluster.leader else cluster.get_member(candidate, False)
 
-        r = request_patroni(member, 'post', action, failover_value, auth_header(obj))
+        r = request_patroni(member, 'post', action, failover_value)
 
         # probably old patroni, which doesn't support switchover yet
-        if r.status_code == 501 and action == 'switchover' and 'Server does not support this operation' in r.text:
-            r = request_patroni(member, 'post', 'failover', failover_value, auth_header(obj))
+        if r.status == 501 and action == 'switchover' and b'Server does not support this operation' in r.data:
+            r = request_patroni(member, 'post', 'failover', failover_value)
 
-        if r.status_code in (200, 202):
+        if r.status in (200, 202):
             logging.debug(r)
             cluster = dcs.get_cluster()
             logging.debug(cluster)
-            click.echo('{0} {1}'.format(timestamp(), r.text))
+            click.echo('{0} {1}'.format(timestamp(), r.data.decode('utf-8')))
         else:
-            click.echo('{0} failed, details: {1}, {2}'.format(action.title(), r.status_code, r.text))
+            click.echo('{0} failed, details: {1}, {2}'.format(action.title(), r.status, r.data.decode('utf-8')))
             return
     except Exception:
         logging.exception(r)
@@ -919,7 +893,7 @@ def flush(obj, cluster_name, member_names, force, role, target):
     for member in members:
         if target == 'restart':
             if member.data.get('scheduled_restart'):
-                r = request_patroni(member, 'delete', 'restart', None, auth_header(obj))
+                r = request_patroni(member, 'delete', 'restart')
                 check_response(r, member.name, 'flush scheduled restart')
             else:
                 click.echo('No scheduled restart for member {0}'.format(member.name))
@@ -956,20 +930,20 @@ def toggle_pause(config, cluster_name, paused, wait):
 
     for member in members:
         try:
-            r = request_patroni(member, 'patch', 'config', {'pause': paused or None}, auth_header(config))
+            r = request_patroni(member, 'patch', 'config', {'pause': paused or None})
         except Exception as err:
             logging.warning(str(err))
             logging.warning('Member %s is not accessible', member.name)
             continue
 
-        if r.status_code == 200:
+        if r.status == 200:
             if wait:
                 wait_until_pause_is_applied(dcs, paused, cluster)
             else:
                 click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed'))
         else:
             click.echo('Failed: {0} cluster management status code={1}, ({2})'.format(
-                       paused and 'pause' or 'resume', r.status_code, r.text))
+                       paused and 'pause' or 'resume', r.status, r.data.decode('utf-8')))
         break
     else:
         raise PatroniCtlException('Can not find accessible cluster member')
@@ -1233,8 +1207,8 @@ def version(obj, cluster_name, member_names):
         if m.api_url:
             if not member_names or m.name in member_names:
                 try:
-                    response = request_patroni(m, 'get', 'patroni')
-                    data = response.json()
+                    response = request_patroni(m)
+                    data = json.loads(response.data)
                     version = data.get('patroni', {}).get('version')
                     pg_version = data.get('server_version')
                     pg_version_str = " PostgreSQL {0}".format(format_pg_version(pg_version)) if pg_version else ""

patroni/ha.py

@@ -3,7 +3,6 @@ import functools
 import json
 import logging
 import psycopg2
-import requests
 import sys
 import time
 import uuid
@@ -558,16 +557,15 @@ class Ha(object):
                                                args=(self.dcs.loop_wait, on_success, self._leader_access_is_restricted))
             return promote_message
 
-    @staticmethod
-    def fetch_node_status(member):
+    def fetch_node_status(self, member):
         """This function perform http get request on member.api_url and fetches its status
         :returns: `_MemberStatus` object
         """
 
         try:
-            response = requests.get(member.api_url, timeout=2, verify=False)
-            logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content)
-            return _MemberStatus.from_api_response(member, response.json())
+            response = self.patroni.request(member, timeout=2, retries=0)
+            logger.info('Got response from %s %s: %s', member.name, member.api_url, response.data.decode('utf-8'))
+            return _MemberStatus.from_api_response(member, json.loads(response.data))
         except Exception as e:
             logger.warning("Request failed to %s: GET %s (%s)", member.name, member.api_url, e)
         return _MemberStatus.unknown(member)

patroni/request.py

@@ -0,0 +1,47 @@
+import json
+import urllib3
+
+from six.moves.urllib_parse import urlparse, urlunparse
+
+
+class PatroniRequest(object):
+
+    def __init__(self, config, insecure=False):
+        cert_reqs = 'CERT_NONE' if insecure or config.get('ctl', {}).get('insecure', False) else 'CERT_REQUIRED'
+        self._pool = urllib3.PoolManager(cert_reqs=cert_reqs)
+        self.reload_config(config)
+
+    @staticmethod
+    def _get_cfg_value(config, name):
+        return config.get('ctl', {}).get(name) or config.get('restapi', {}).get(name)
+
+    def _apply_pool_param(self, param, value):
+        if value:
+            self._pool.connection_pool_kw[param] = value
+        else:
+            self._pool.connection_pool_kw.pop(param, None)
+
+    def _apply_ssl_file_param(self, config, name):
+        value = self._get_cfg_value(config, name + 'file')
+        self._apply_pool_param(name + '_file', value)
+        return value
+
+    def reload_config(self, config):
+        self._pool.headers = urllib3.make_headers(basic_auth=self._get_cfg_value(config, 'auth'))
+
+        if self._apply_ssl_file_param(config, 'cert'):
+            self._apply_ssl_file_param(config, 'key')
+        else:
+            self._pool.connection_pool_kw.pop('key_file', None)
+
+        cacert = config.get('ctl', {}).get('cacert') or config.get('restapi', {}).get('cafile')
+        self._apply_pool_param('ca_certs', cacert)
+
+    def __call__(self, member, method='GET', endpoint=None, data=None, **kwargs):
+        url = member.api_url
+        if endpoint:
+            scheme, netloc, _, _, _, _ = urlparse(url)
+            url = urlunparse((scheme, netloc, endpoint, '', '', ''))
+        if data is not None:
+            kwargs['body'] = json.dumps(data)
+        return self._pool.request(method.upper(), url, **kwargs)

tests/test_api.py

@@ -145,7 +145,7 @@ class MockRestApiServer(RestApiServer):
         self.serve_forever = Mock()
         MockRestApiServer._BaseServer__is_shut_down = Mock()
         MockRestApiServer._BaseServer__shutdown_request = True
-        config = config or {'listen': '127.0.0.1:8008', 'auth': 'test:test', 'certfile': 'dumb'}
+        config = config or {'listen': '127.0.0.1:8008', 'auth': 'test:test', 'certfile': 'dumb', 'verify_client': 'a'}
         super(MockRestApiServer, self).__init__(MockPatroni(), config)
         Handler(MockRequest(request), ('0.0.0.0', 8080), self)
 
@@ -408,12 +408,18 @@ class TestRestApiServer(unittest.TestCase):
     def test_reload_config(self):
         bad_config = {'listen': 'foo'}
         self.assertRaises(ValueError, MockRestApiServer, None, '', bad_config)
-        srv = MockRestApiServer(lambda a1, a2, a3: None, '')
+        srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required'})
         self.assertRaises(ValueError, srv.reload_config, bad_config)
         self.assertRaises(ValueError, srv.reload_config, {})
         with patch.object(socket.socket, 'setsockopt', Mock(side_effect=socket.error)):
             srv.reload_config({'listen': ':8008'})
 
+    def test_check_auth(self):
+        srv = MockRestApiServer(Mock(), '', {'listen': '*:8008', 'certfile': 'a', 'verify_client': 'required'})
+        mock_rh = Mock()
+        mock_rh.request.getpeercert.return_value = None
+        self.assertIsNot(srv.check_auth(mock_rh), True)
+
     def test_handle_error(self):
         try:
             raise Exception()

tests/test_ctl.py

@@ -1,20 +1,21 @@
 import etcd
+import json
 import os
-import requests
 import sys
 import unittest
 
 from click.testing import CliRunner
 from datetime import datetime, timedelta
 from mock import patch, Mock
-from patroni.ctl import ctl, store_config, load_config, output_members, request_patroni, get_dcs, parse_dcs, \
+from patroni.ctl import ctl, store_config, load_config, output_members, get_dcs, parse_dcs, \
     get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException, apply_config_changes, \
     format_config_for_editing, show_diff, invoke_editor, format_pg_version, find_executable
 from patroni.dcs.etcd import Client, Failover
 from patroni.utils import tzutc
 from psycopg2 import OperationalError
+from urllib3 import PoolManager
 
-from . import MockConnect, MockCursor, MockResponse, psycopg2_connect, requests_get
+from . import MockConnect, MockCursor, MockResponse, psycopg2_connect
 from .test_etcd import etcd_read, socket_getaddrinfo
 from .test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \
     get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member
@@ -35,7 +36,7 @@ def test_rw_config():
 
 @patch('patroni.ctl.load_config',
        Mock(return_value={'scope': 'alpha', 'postgresql': {'data_dir': '.', 'parameters': {}, 'retry_timeout': 5},
-                          'restapi': {'auth': 'u:p', 'listen': ''}, 'etcd': {'host': 'localhost:2379'}}))
+                          'restapi': {'listen': '::', 'certfile': 'a'}, 'etcd': {'host': 'localhost:2379'}}))
 class TestCtl(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
@@ -75,7 +76,7 @@ class TestCtl(unittest.TestCase):
         self.assertIsNone(output_members(cluster, name='abc', fmt='tsv'))
 
     @patch('patroni.ctl.get_dcs')
-    @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse()))
+    @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
     def test_switchover(self, mock_get_dcs):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
@@ -125,18 +126,18 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='dummy')
         assert result.exit_code == 1
 
-        with patch('patroni.ctl.request_patroni', Mock(side_effect=Exception)):
+        with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
             # Non-responding patroni
             result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n2300-01-01T12:23:00\ny')
             assert 'falling back to DCS' in result.output
 
-        with patch('patroni.ctl.request_patroni') as mocked:
-            mocked.return_value.status_code = 500
+        with patch.object(PoolManager, 'request') as mocked:
+            mocked.return_value.status = 500
             result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
             assert 'Switchover failed' in result.output
 
-            mocked.return_value.status_code = 501
-            mocked.return_value.text = 'Server does not support this operation'
+            mocked.return_value.status = 501
+            mocked.return_value.data = b'Server does not support this operation'
             result = self.runner.invoke(ctl, ['switchover', 'dummy'], input='leader\nother\n\ny')
             assert 'Switchover failed' in result.output
 
@@ -151,7 +152,7 @@ class TestCtl(unittest.TestCase):
         assert result.exit_code == 1
 
     @patch('patroni.ctl.get_dcs')
-    @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse()))
+    @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
     def test_failover(self, mock_get_dcs):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
@@ -231,7 +232,7 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy'])
         assert result.exit_code == 1
 
-    @patch('requests.post')
+    @patch.object(PoolManager, 'request')
     @patch('patroni.ctl.get_dcs')
     def test_reload(self, mock_get_dcs, mock_post):
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
@@ -239,18 +240,19 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y')
         assert 'Failed: reload for member' in result.output
 
-        mock_post.return_value.status_code = 200
+        mock_post.return_value.status = 200
         result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y')
         assert 'No changes to apply on member' in result.output
 
-        mock_post.return_value.status_code = 202
+        mock_post.return_value.status = 202
         result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y')
         assert 'Reload request received for member' in result.output
 
-    @patch('requests.post', requests_get)
+    @patch.object(PoolManager, 'request')
     @patch('patroni.ctl.get_dcs')
-    def test_restart_reinit(self, mock_get_dcs):
+    def test_restart_reinit(self, mock_get_dcs, mock_post):
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+        mock_post.return_value.status = 503
         result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\ny\n')
         assert 'Failed: restart for' in result.output
         assert result.exit_code == 0
@@ -285,58 +287,49 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force', '--timeout', '10min'])
         assert result.exit_code == 0
 
-        with patch('requests.delete', Mock(return_value=MockResponse(500))):
-            # normal restart, the schedule is actually parsed, but not validated in patronictl
-            result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force',
-                                              '--scheduled', '2300-10-01T14:30'])
-            assert 'Failed: flush scheduled restart' in result.output
+        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
+        assert 'Failed: flush scheduled restart' in result.output
 
         with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)):
             result = self.runner.invoke(ctl,
                                         ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
             assert result.exit_code == 1
 
-        with patch('requests.post', Mock(return_value=MockResponse())):
-            # normal restart, the schedule is actually parsed, but not validated in patronictl
-            result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0.0',
-                                              '--scheduled', '2300-10-01T14:30'], input='y')
-            assert result.exit_code == 0
-
-        with patch('requests.post', Mock(return_value=MockResponse(204))):
-            # get restart with the non-200 return code
-            # normal restart, the schedule is actually parsed, but not validated in patronictl
-            result = self.runner.invoke(ctl, ['restart', 'alpha', '--pg-version', '42.0',
-                                              '--scheduled', '2300-10-01T14:30'], input='y')
-            assert result.exit_code == 0
-
         # force restart with restart already present
-        with patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse(204))):
-            result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force',
-                                              '--scheduled', '2300-10-01T14:30'])
-            assert result.exit_code == 0
+        result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
+        assert result.exit_code == 0
 
-        with patch('requests.post', Mock(return_value=MockResponse(202))):
-            # get restart with the non-200 return code
-            # normal restart, the schedule is actually parsed, but not validated in patronictl
-            result = self.runner.invoke(
-                ctl, ['restart', 'alpha', '--pg-version', '99.0.0', '--scheduled', '2300-10-01T14:30'], input='y'
-            )
-            assert 'Success: restart scheduled' in result.output
-            assert result.exit_code == 0
+        ctl_args = ['restart', 'alpha', '--pg-version', '99.0', '--scheduled', '2300-10-01T14:30']
+        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        mock_post.return_value.status = 200
+        result = self.runner.invoke(ctl, ctl_args, input='y')
+        assert result.exit_code == 0
 
-        with patch('requests.post', Mock(return_value=MockResponse(409))):
-            # get restart with the non-200 return code
-            # normal restart, the schedule is actually parsed, but not validated in patronictl
-            result = self.runner.invoke(
-                ctl, ['restart', 'alpha', '--pg-version', '99.0.0', '--scheduled', '2300-10-01T14:30'], input='y'
-            )
-            assert 'Failed: another restart is already' in result.output
-            assert result.exit_code == 0
+        # get restart with the non-200 return code
+        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        mock_post.return_value.status = 204
+        result = self.runner.invoke(ctl, ctl_args, input='y')
+        assert result.exit_code == 0
+
+        # get restart with the non-200 return code
+        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        mock_post.return_value.status = 202
+        result = self.runner.invoke(ctl, ctl_args, input='y')
+        assert 'Success: restart scheduled' in result.output
+        assert result.exit_code == 0
+
+        # get restart with the non-200 return code
+        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        mock_post.return_value.status = 409
+        result = self.runner.invoke(ctl, ctl_args, input='y')
+        assert 'Failed: another restart is already' in result.output
+        assert result.exit_code == 0
 
     @patch('patroni.ctl.get_dcs')
     def test_remove(self, mock_get_dcs):
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
-        result = self.runner.invoke(ctl, ['remove', 'alpha'], input='alpha\nslave')
+        result = self.runner.invoke(ctl, ['-k', 'remove', 'alpha'], input='alpha\nslave')
         assert 'Please confirm' in result.output
         assert 'You are about to remove all' in result.output
         # Not typing an exact confirmation
@@ -353,17 +346,6 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['remove', 'alpha'], input='alpha\nYes I am aware\nleader')
         assert result.exit_code == 0
 
-    @patch('requests.post', Mock(side_effect=requests.exceptions.ConnectionError('foo')))
-    @patch('click.get_current_context')
-    def test_request_patroni(self, mock_context):
-        member = get_cluster_initialized_with_leader().leader.member
-
-        mock_context.return_value.obj = {'ctl': {'cacert': 'cert.pem'}}
-        self.assertRaises(requests.exceptions.ConnectionError, request_patroni, member, 'post', 'dummy', {})
-
-        mock_context.return_value.obj = {'ctl': {'insecure': True}}
-        self.assertRaises(requests.exceptions.ConnectionError, request_patroni, member, 'post', 'dummy', {})
-
     def test_ctl(self):
         self.runner.invoke(ctl, ['list'])
 
@@ -437,7 +419,7 @@ class TestCtl(unittest.TestCase):
         assert 'Scheduled restart' in result.output
 
     @patch('patroni.ctl.get_dcs')
-    @patch('requests.delete', Mock(return_value=MockResponse()))
+    @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
     def test_flush(self, mock_get_dcs):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
@@ -447,58 +429,58 @@ class TestCtl(unittest.TestCase):
 
         result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force'])
         assert 'Success: flush scheduled restart' in result.output
-        with patch.object(requests, 'delete', return_value=MockResponse(404)):
+        with patch.object(PoolManager, 'request', return_value=MockResponse(404)):
             result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force'])
             assert 'Failed: flush scheduled restart' in result.output
 
+    @patch.object(PoolManager, 'request')
     @patch('patroni.ctl.get_dcs')
     @patch('patroni.ctl.polling_loop', Mock(return_value=[1]))
-    def test_pause_cluster(self, mock_get_dcs):
+    def test_pause_cluster(self, mock_get_dcs, mock_post):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
 
-        with patch('requests.patch', Mock(return_value=MockResponse(500))):
-            result = self.runner.invoke(ctl, ['pause', 'dummy'])
-            assert 'Failed' in result.output
+        mock_post.return_value.status = 500
+        result = self.runner.invoke(ctl, ['pause', 'dummy'])
+        assert 'Failed' in result.output
 
-        with patch('requests.patch', Mock(return_value=MockResponse(200))),\
-                patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)):
+        mock_post.return_value.status = 200
+        with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)):
             result = self.runner.invoke(ctl, ['pause', 'dummy'])
             assert 'Cluster is already paused' in result.output
 
-        with patch('requests.patch', Mock(return_value=MockResponse(200))):
-            result = self.runner.invoke(ctl, ['pause', 'dummy', '--wait'])
-            assert "'pause' request sent" in result.output
-            mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(),
-                                                                      get_cluster(None, None, [], None, None)])
-            self.runner.invoke(ctl, ['pause', 'dummy', '--wait'])
-            member = Member(1, 'other', 28, {})
-            mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(),
-                                                                      get_cluster(None, None, [member], None, None)])
-            self.runner.invoke(ctl, ['pause', 'dummy', '--wait'])
+        result = self.runner.invoke(ctl, ['pause', 'dummy', '--wait'])
+        assert "'pause' request sent" in result.output
+        mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(),
+                                                                  get_cluster(None, None, [], None, None)])
+        self.runner.invoke(ctl, ['pause', 'dummy', '--wait'])
+        member = Member(1, 'other', 28, {})
+        mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(),
+                                                                  get_cluster(None, None, [member], None, None)])
+        self.runner.invoke(ctl, ['pause', 'dummy', '--wait'])
 
+    @patch.object(PoolManager, 'request')
     @patch('patroni.ctl.get_dcs')
-    def test_resume_cluster(self, mock_get_dcs):
+    def test_resume_cluster(self, mock_get_dcs, mock_post):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
 
+        mock_post.return_value.status = 200
+        with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=False)):
+            result = self.runner.invoke(ctl, ['resume', 'dummy'])
+            assert 'Cluster is not paused' in result.output
+
         with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)):
-            with patch('requests.patch', Mock(return_value=MockResponse(200))):
-                result = self.runner.invoke(ctl, ['resume', 'dummy'])
-                assert 'Success' in result.output
+            result = self.runner.invoke(ctl, ['resume', 'dummy'])
+            assert 'Success' in result.output
 
-            with patch('requests.patch', Mock(return_value=MockResponse(500))):
-                result = self.runner.invoke(ctl, ['resume', 'dummy'])
-                assert 'Failed' in result.output
+            mock_post.return_value.status = 500
+            result = self.runner.invoke(ctl, ['resume', 'dummy'])
+            assert 'Failed' in result.output
 
-            with patch('requests.patch', Mock(return_value=MockResponse(200))),\
-                    patch('patroni.dcs.Cluster.is_paused', Mock(return_value=False)):
-                result = self.runner.invoke(ctl, ['resume', 'dummy'])
-                assert 'Cluster is not paused' in result.output
-
-            with patch('requests.patch', Mock(side_effect=Exception)):
-                result = self.runner.invoke(ctl, ['resume', 'dummy'])
-                assert 'Can not find accessible cluster member' in result.output
+            mock_post.side_effect = Exception
+            result = self.runner.invoke(ctl, ['resume', 'dummy'])
+            assert 'Can not find accessible cluster member' in result.output
 
     def test_apply_config_changes(self):
         config = {"postgresql": {"parameters": {"work_mem": "4MB"}, "use_pg_rewind": True}, "ttl": 30}
@@ -574,13 +556,13 @@ class TestCtl(unittest.TestCase):
     def test_version(self, mock_get_dcs):
         mock_get_dcs.return_value = self.e
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
-        with patch('patroni.ctl.request_patroni') as mocked:
+        with patch.object(PoolManager, 'request') as mocked:
             result = self.runner.invoke(ctl, ['version'])
             assert 'patronictl version' in result.output
-            mocked.return_value.json = lambda: {'patroni': {'version': '1.2.3'}, 'server_version': 100001}
+            mocked.return_value.data = json.dumps({'patroni': {'version': '1.2.3'}, 'server_version': 100001})
             result = self.runner.invoke(ctl, ['version', 'dummy'])
             assert '1.2.3' in result.output
-        with patch('requests.get', Mock(side_effect=Exception)):
+        with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
             result = self.runner.invoke(ctl, ['version', 'dummy'])
             assert 'failed to get version' in result.output
 

tests/test_ha.py

@@ -137,6 +137,7 @@ zookeeper:
         self.scheduled_restart = {'schedule': future_restart_time,
                                   'postmaster_start_time': str(postmaster_start_time)}
         self.watchdog = Watchdog(self.config)
+        self.request = lambda member, **kwargs: requests_get(member.api_url, **kwargs)
 
 
 def run_async(self, func, args=()):
@@ -464,7 +465,6 @@ class TestHa(PostgresInit):
                     self.assertEqual(self.ha.run_cycle(), 'lost leader lock during restart')
                     mock_terminate.assert_called()
 
-    @patch('requests.get', requests_get)
     def test_manual_failover_from_leader(self):
         self.ha.fetch_node_status = get_node_status()
         self.ha.has_lock = true
@@ -512,7 +512,6 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
         self.assertEqual('no action.  i am the leader with the lock', self.ha.run_cycle())
 
-    @patch('requests.get', requests_get)
     def test_manual_failover_from_leader_in_pause(self):
         self.ha.has_lock = true
         self.ha.is_paused = true
@@ -522,7 +521,6 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, self.p.name, '', None))
         self.assertEqual('PAUSE: no action.  i am the leader with the lock', self.ha.run_cycle())
 
-    @patch('requests.get', requests_get)
     def test_manual_failover_from_leader_in_synchronous_mode(self):
         self.p.is_leader = true
         self.ha.has_lock = true
@@ -535,7 +533,6 @@ class TestHa(PostgresInit):
         self.ha.is_failover_possible = true
         self.assertEqual('manual failover: demoting myself', self.ha.run_cycle())
 
-    @patch('requests.get', requests_get)
     def test_manual_failover_process_no_leader(self):
         self.p.is_leader = false
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None))
@@ -585,7 +582,6 @@ class TestHa(PostgresInit):
         self.ha.is_paused = true
         self.assertFalse(self.ha.is_healthiest_node())
 
-    @patch('requests.get', requests_get)
     def test__is_healthiest_node(self):
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('postgresql1', self.p.name))
         self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -607,7 +603,6 @@ class TestHa(PostgresInit):
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.patroni.nofailover = False
 
-    @patch('requests.get', requests_get)
     def test_fetch_node_status(self):
         member = Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni'})
         self.ha.fetch_node_status(member)
@@ -696,7 +691,6 @@ class TestHa(PostgresInit):
         with patch.object(Leader, 'conn_url', PropertyMock(return_value='')):
             self.assertEqual(self.ha.run_cycle(), 'continue following the old known standby leader')
 
-    @patch('requests.get', requests_get)
     def test_process_unhealthy_standby_cluster_as_standby_leader(self):
         self.p.is_leader = false
         self.p.name = 'leader'
@@ -1027,7 +1021,6 @@ class TestHa(PostgresInit):
             self.assertEqual(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
 
     @patch('sys.exit', return_value=1)
-    @patch('requests.get', requests_get)
     def test_abort_join(self, exit_mock):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
         self.p.is_leader = false