text: """Management API's for Couchbase Cluster"""
  
  
  import csv
  import json
  import requests
  import os
  import io
  import sys
  import time
  import urllib.request, urllib.parse, urllib.error
  
  N1QL_SERVICE = 'n1ql'
  INDEX_SERVICE = 'index'
  MGMT_SERVICE = 'mgmt'
  FTS_SERVICE = 'fts'
  EVENT_SERVICE = 'eventing'
  
  ERR_AUTH = 'unable to access the REST API - please check your username (-u) and password (-p)'
  ERR_INTERNAL = 'Internal server error, please retry your request'
  
  DEFAULT_REQUEST_TIMEOUT = 60
  
  try:
      from cb_version import VERSION
  except ImportError:
      VERSION = '0.0.0-0000'
  
  # Remove this once we can verify SSL certificates
  requests.packages.urllib3.disable_warnings()
  
  def request(f):
      def g(*args, **kwargs):
          cm = args[0]
          url = args[1]
          try:
              return f(*args, **kwargs)
          except requests.exceptions.ConnectionError as e:
              if '[SSL: CERTIFICATE_VERIFY_FAILED]' in str(e):
                  return None, ['Certificate verification failed.\n' +
                  'If you are using self-signed certificates you can re-run this command with\n' +
                  'the --no-ssl-verify flag. Note however that disabling ssl verification\n' +
                  'means that couchbase-cli will be vulnerable to man-in-the-middle attacks.\n\n' +
                  'For the most secure access to Couchbase make sure that you have X.509\n' +
                  'certificates set up in your cluster and use the --cacert flag to specify\n' +
                  'your client certificate.']
              elif str(e).startswith('[SSL]'):
                  return None, [f'Unable to connect with the given CA certificate: {str(e)}']
              return None, [f'Unable to connect to host at {cm.hostname}: {str(e)}']
          except requests.exceptions.ReadTimeout as e:
              return None, [f'Request to host `{url}` timed out after {cm.timeout} seconds']
      return g
  
  
  class ServiceNotAvailableException(Exception):
      """An exception raised when a service does not exist in the target cluster"""
  
      def __init__(self, service):
          Exception.__init__(self, f'Service {service} not available in target cluster')
  
  
  class ClusterManager(object):
      """A set of REST API's for managing a Couchbase cluster"""
  
      def __init__(self, hostname, username, password, sslFlag=False, verifyCert=True,
                   caCert=True, debug=False, timeout=DEFAULT_REQUEST_TIMEOUT, cert=None):
          hostname = hostname.replace("couchbase://", "http://", 1)
          hostname = hostname.replace("couchbases://", "https://", 1)
  
          self.hostname = hostname
          # verify argument on Request functions can take boolean or a path to a CA if
          # a path is not provide but the cert still needs to be verified it should use
          # the system provided CAs
          self.verifyCert = verifyCert
          self.caCert = caCert
          if not verifyCert:
              self.caCert = False
          # This is for client side certs which is currently not used.
          self.cert = cert
  
          parsed = urllib.parse.urlparse(hostname)
          if sslFlag:
              hostport = parsed.hostname.split(':')
              if parsed.scheme == 'http://':
                  if parsed.port == 8091:
                      self.hostname = f'https://{parsed.hostname}:18091'
                  else:
                      self.hostname = f'https://{parsed.hostname}:{parsed.port}'
  
              # Certificates and verification are not used when the ssl flag is
              # specified.
              self.verifyCert = False
              self.caCert = False
  
          self.username = username.encode('utf-8').decode('latin1')
          self.password = password.encode('utf-8').decode('latin1')
          self.timeout = timeout
          self.ssl = self.hostname.startswith("https://")
          self.debug = debug
          self.headers = requests.utils.default_headers()
          self.headers.update(
              {
                  'User-Agent': f'{os.path.basename(sys.argv[0])}  {VERSION}',
              }
          )
  
      def restore_index_metadata(self, bucket, index_defs):
          hosts, errors = self.get_hostnames_for_service(INDEX_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(INDEX_SERVICE)
  
          url = f'{hosts[0]}/restoreIndexMetadata?bucket={bucket}'
          return self._post_json(url, index_defs)
  
      def get_index_metadata(self, bucket):
          hosts, errors = self.get_hostnames_for_service(INDEX_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(INDEX_SERVICE)
  
          url = f'{hosts[0]}/getIndexMetadata?bucket={bucket}'
          return self._get(url)
  
      def restore_fts_index_metadata(self, index_defs):
          hosts, errors = self.get_hostnames_for_service(FTS_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(FTS_SERVICE)
  
          for index_def in index_defs:
              url = f'{hosts[0]}/api/index/{index_def["name"]}?prevIndexUUID=*'
              if "sourceUUID" in index_def:
                  del index_def["sourceUUID"]
              result, errors = self._put_json(url, index_def)
              if errors:
                  return None, errors
  
          return None, None
  
      def get_fts_index_alias(self):
          hosts, errors = self.get_hostnames_for_service(FTS_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(FTS_SERVICE)
  
          url = f'{hosts[0]}/api/index'
          result, errors = self._get(url)
          if errors:
              return None, errors
  
          index_defs = []
          if "indexDefs" in result and result["indexDefs"] is not None:
              for _, index in result['indexDefs']['indexDefs'].items():
                  if index['type'] == 'fulltext-alias':
                      index_defs.append(index)
  
          return index_defs, None
  
      def get_fts_index_metadata(self, bucket):
          hosts, errors = self.get_hostnames_for_service(FTS_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(FTS_SERVICE)
  
          url = f'{hosts[0]}/api/index'
          result, errors = self._get(url)
          if errors:
              return None, errors
  
  
          bucket_index_defs = []
          if "indexDefs" in result and result["indexDefs"] is not None:
              for _, index_def in result["indexDefs"]["indexDefs"].items():
                  if index_def["type"] == "fulltext-index" and index_def["sourceName"] == bucket:
                      bucket_index_defs.append(index_def)
          return bucket_index_defs, None
  
      def n1ql_query(self, stmt, args=None):
          """Sends a N1QL query
  
          Sends a N1QL query and returns the result of the query. Raises a
          ServiceNotAvailable exception if the target cluster is no running the n1ql
          service."""
  
          hosts, errors = self.get_hostnames_for_service(N1QL_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(N1QL_SERVICE)
  
          url = f'{hosts[0]}/query/service'
          body = {'statement': str(stmt)}
  
          if args:
              body['args'] = str(args)
  
          result, errors = self._post_form_encoded(url, body)
          if errors:
              return None, errors
  
          return result, None
  
      def is_cluster_initialized(self):
          data, errors = self.pools()
          if (errors and len(errors) == 1 and errors[0] == ERR_AUTH) or \
              (data and data['pools'] and len(data['pools']) > 0):
              return True, None
          return False, errors
  
      def is_enterprise(self):
          data, errors = self.pools()
          if errors:
              return None, errors
          return data["isEnterprise"], None
  
      def get_hostnames_for_service(self, service_name):
          """ Gets all hostnames that run a service
  
          Gets all hostnames for specified service and returns a list of strings
          in the form "http://hostname:port". If the ClusterManager is configured
          to use SSL/TLS then "https://" is prefixed to each name instead of
          "http://"."""
          url = f'{self.hostname}/pools/default/nodeServices'
          data, errors = self._get(url)
          if errors:
              return None, errors
  
          # this block of code will check if we are using internal or external address
          # first get the host being used to get the node services info
          used_host = urllib.parse.urlparse(self.hostname).hostname
          useAlt = False
          # next check if its external or internal
          for node in data['nodesExt']:
              if not 'hostname' in node and (used_host == '127.0.0.1' or used_host == 'localhost'):
                  useAlt = False
                  break
              if 'hostname' in node and used_host == node['hostname']:
                  useAlt = False
                  break
              if 'alternateAddresses' in node and node['alternateAddresses']['external']['hostname'] == used_host:
                  useAlt = True
                  break
  
          hosts = []
          for node in data['nodesExt']:
              node_host = '127.0.0.1'
              if 'hostname' in node:
                  node_host = node['hostname']
  
              # Check for Raw IPv6 address
              if ':' in node_host:
                  node_host = '[' + node_host + ']'
  
              http_prefix = 'http://'
              fts_port_name = 'fts'
              n1ql_port_name = 'n1ql'
              mgmt_port_name = 'mgmt'
              index_port_name = 'indexHttp'
              event_port_name = 'eventingAdminPort'
  
              if self.ssl:
                  http_prefix = 'https://'
                  n1ql_port_name = 'n1qlSSL'
                  mgmt_port_name = 'mgmtSSL'
                  event_port_name = 'eventingSSL'
                  index_port_name = 'indexHttps'
                  fts_port_name = 'ftsSSL'
  
              services = node['services']
  
              if useAlt and 'alternateAddresses' not in node:
                  continue
  
              if 'alternateAddresses' in node and useAlt:
                  alt_node_host = node['alternateAddresses']['external']['hostname']
                  # Check for Raw IPv6 address
                  if ':' in alt_node_host:
                      alt_node_host = '[' + alt_node_host + ']'
                  node_host = alt_node_host
                  services = node['alternateAddresses']['external']['ports']
  
              if service_name == MGMT_SERVICE and mgmt_port_name in services:
                  hosts.append(http_prefix + node_host + ':' + str(services[mgmt_port_name]))
  
              if service_name == N1QL_SERVICE and n1ql_port_name in services:
                  hosts.append(http_prefix + node_host + ':' + str(services[n1ql_port_name]))
  
              if service_name == INDEX_SERVICE and index_port_name in services:
                  hosts.append(http_prefix + node_host + ':' + str(services[index_port_name]))
  
              if service_name == FTS_SERVICE and fts_port_name in services:
                  hosts.append(http_prefix + node_host + ':' + str(services[fts_port_name]))
  
              if service_name == EVENT_SERVICE and event_port_name in services:
                  hosts.append(http_prefix + node_host + ':' + str(services[event_port_name]))
  
          return hosts, None
  
      def pools(self, pool=None):
          """ Retrieves information about Couchbase management pools
  
          Returns Couchbase pools data"""
          url = f'{self.hostname}/pools'
          if pool:
              url += '/' + pool
          return self._get(url)
  
      def set_admin_password(self, password):
          url = f'{self.hostname}/controller/resetAdminPassword'
          params = { "password": password }
  
          return self._post_form_encoded(url, params)
  
      def regenerate_admin_password(self):
          url = f'{self.hostname}/controller/resetAdminPassword?generate=1'
  
          return self._post_form_encoded(url, None)
  
      def rotate_master_pwd(self):
          url = f'{self.hostname}/node/controller/rotateDataKey'
          return self._post_form_encoded(url, None)
  
      def set_master_pwd(self, password):
          url = f'{self.hostname}/node/controller/changeMasterPassword'
          params = { "newPassword": password }
          return self._post_form_encoded(url, params)
  
      def user_change_passsword(self, new_password):
          url = f'{self.hostname}/controller/changePassword'
          params = {'password': new_password}
          return self._post_form_encoded(url, params)
  
      def set_pools_default(self, data_ramsize, index_ramsize, fts_ramsize, cbas_ramsize, eventing_ramsize, cluster_name):
          """ Sets Couchbase RAM Quotas for various services
  
          Options:
          data_ramsize - An integer denoting the size in MB, None skips the parameter
          index_ramsize - An integer denoting the size in MB, None skips the parameter
          fts_ramsize - An integer denoting the size in MB, None skips the parameter
          cbas_ramsize - An integer denoting the size in MB, None skips the parameter
          eventing_ramsize - An integer denoting the size in MB, None skips the parameter
          cluster_name - Sets a name for the cluster, None skips the parameter
          """
          url = f'{self.hostname}/pools/default'
          params = {}
          if data_ramsize:
              params["memoryQuota"] = data_ramsize
          if index_ramsize:
              params["indexMemoryQuota"] = index_ramsize
          if fts_ramsize:
              params["ftsMemoryQuota"] = fts_ramsize
          if cbas_ramsize:
              params["cbasMemoryQuota"] = cbas_ramsize
          if eventing_ramsize:
              params["eventingMemoryQuota"] = eventing_ramsize
          if cluster_name:
              params["clusterName"] = cluster_name
  
          return self._post_form_encoded(url, params)
  
      def setup_services(self, services):
          """ Sets the services on a node
  
          Options:
          services - A string containing a comma separated list of services
          """
          url = f'{self.hostname}/node/controller/setupServices'
          params = { "services": services }
  
          return self._post_form_encoded(url, params)
  
      def set_admin_credentials(self, username, password, port):
          """Sets the admin credentials and port for a cluster
  
          Options:
          username - The username for the cluster
          password - The password for the cluster
          port - The port number for the admin console to listen on. If set to
                 None then the port is kept the same as it currently is.
          """
          url = f'{self.hostname}/settings/web'
          params = {}
  
          if username:
              params["username"] = username
          if password:
              params["password"] = password
          if port:
              params["port"] = port
          else:
              params["port"] = "SAME"
  
          return self._post_form_encoded(url, params)
  
      def enable_notifications(self, enable):
          url = f'{self.hostname}/settings/stats'
          params = { "sendStats": "false"}
  
          if enable:
              params["sendStats"] = "true"
  
          return self._post_form_encoded(url, params)
  
      def get_server_groups(self):
          url = f'{self.hostname}/pools/default/serverGroups'
          return self._get(url)
  
      def get_server_group(self, groupName):
          groups, errors = self.get_server_groups()
          if errors:
              return None, errors
  
          if not groups or not groups["groups"] or groups["groups"] == 0:
              return None, ["No server groups found"]
  
          if groupName:
              for group in groups["groups"]:
                  if group["name"] == groupName:
                      return group, None
              return None, [f'Group `{groupName}` not found']
          else:
              return groups["groups"][0], None
  
      def add_server(self, add_server, groupName, username, password, services):
          group, errors = self.get_server_group(groupName)
          if errors:
              return None, errors
  
          url = f'{self.hostname}{group["addNodeURI"]}'
          params = { "hostname": add_server,
                     "user": username,
                     "password": password,
                     "services": services }
  
          return self._post_form_encoded(url, params)
  
      def readd_server(self, server):
          _, _, _, readd, _, errors = self._get_otps_names(readd_nodes=[server])
          if errors:
              return None, errors
  
          if len(readd) != 1:
              return None, ["Server not found %s" % server]
  
          url = f'{self.hostname}/controller/reAddNode'
          params = { "otpNode": readd[0] }
  
          return self._post_form_encoded(url, params)
  
      def get_tasks(self):
          url = f'{self.hostname}/pools/default/tasks'
          return self._get(url)
  
      def collect_logs_start(self, servers, redaction_level, salt, log_dir, tmp_dir, upload, upload_host, upload_proxy,
                             upload_customer, upload_ticket):
          url = f'{self.hostname}/controller/startLogsCollection'
          params = dict()
  
          if servers == "*":
              params["nodes"] = servers
          else:
              nodes = servers.split(",")
              known, _, _, readd, _, errors = self._get_otps_names(readd_nodes=nodes)
              if errors:
                  return None, errors
  
              if len(nodes) != len(readd):
                  return None, ["Servers list contains invalid servers"]
  
              params["nodes"] = ",".join(readd)
  
          if redaction_level:
              params["logRedactionLevel"] = redaction_level
          if log_dir:
              params["logDir"] = log_dir
          if tmp_dir:
              params["tmpDir"] = tmp_dir
          if salt:
              params["logRedactionSalt"] = salt
  
          if upload:
              if upload_host:
                  params["uploadHost"] = upload_host
              if upload_proxy:
                  params["uploadProxy"] = upload_proxy
              if upload_customer:
                  params["customer"] = upload_customer
              if upload_ticket:
                  params["ticket"] = upload_ticket
  
          return self._post_form_encoded(url, params)
  
      def collect_logs_stop(self):
          url = f'{self.hostname}/controller/cancelLogsCollection'
          return self._post_form_encoded(url, dict())
  
      def failover(self, servers_to_failover, force):
          _, _, failover, _, _, errors = self._get_otps_names(failover_nodes=servers_to_failover)
          if errors:
              return None, errors
  
  
          if len(failover) != len(servers_to_failover):
              if len(servers_to_failover) == 1:
                  return None, ["Server can't be failed over because it's not part of the cluster"]
              return None, ["Some nodes specified to be failed over are not part of the cluster"]
  
          params = {"otpNode": [server for server, _ in failover]}
  
          if force:
              params["allowUnsafe"] = "true"
              url = f'{self.hostname}/controller/failOver'
              return self._post_form_encoded(url, params)
          else:
              for server, server_status in failover:
                  if server_status != 'healthy':
                      return None, ["% can't be gracefully failed over because it is not healthy", server]
              url = f'{self.hostname}/controller/startGracefulFailover'
              return self._post_form_encoded(url, params)
  
      def recovery(self, server, recovery_type):
          _, _, _, readd, _, errors = self._get_otps_names(readd_nodes=[server])
          if errors:
              return None, errors
  
          if len(readd) != 1:
              return None, [f'Server not found {server}']
  
          url = f'{self.hostname}/controller/setRecoveryType'
          params = { "otpNode": readd[0],
                     "recoveryType": recovery_type }
  
          return self._post_form_encoded(url, params)
  
      def rebalance(self, remove_nodes):
          url = f'{self.hostname}/controller/rebalance'
          all, eject, _, _, _, errors = self._get_otps_names(eject_nodes=remove_nodes)
          if errors:
              return None, errors
  
          if len(eject) != len(remove_nodes):
              return None, ["Some nodes specified to be removed are not part of the cluster"]
  
          params = { "knownNodes": ','.join(all),
                     "ejectedNodes": ','.join(eject) }
  
          return self._post_form_encoded(url, params)
  
      def set_settings_rebalance_retry(self, enabled, wait_for, max_attempts):
          url = f'{self.hostname}/settings/retryRebalance'
  
          params = {'enabled': enabled}
          if wait_for:
              params['afterTimePeriod'] = wait_for
          if max_attempts:
              params['maxAttempts'] = max_attempts
  
          return self._post_form_encoded(url, params)
  
      def get_settings_rebalance_retry(self):
          url = f'{self.hostname}/settings/retryRebalance'
          return self._get(url)
  
      def cancel_rebalance_retry(self, rebalance_id):
          url = f'{self.hostname}/controller/cancelRebalanceRetry/{rebalance_id}'
          return self._post_form_encoded(url, {})
  
      def get_rebalance_info(self):
          url = f'{self.hostname}/pools/default/pendingRetryRebalance'
          return self._get(url)
  
      def rebalance_status(self):
          data, errors = self.get_tasks()
          if errors:
              return (None, None), errors
  
          rv = {
              "status": "unknown",
              "msg": "unknown state",
              "details": {}
          }
  
          for task in data:
              if task["type"] != "rebalance":
                  continue
  
              err_msg = None
              if "errorMessage" in task:
                  rv["status"] = "errored"
                  rv["msg"] = task['errorMessage']
                  break
              elif task["status"] == "running":
                  rv["status"] = task["status"]
                  rv["msg"] = "Rebalance is running"
                  rv["details"]["progress"] = task["progress"]
                  rv["details"]["refresh"] = task["recommendedRefreshPeriod"]
                  rv["details"]["totalBuckets"] = 0
                  rv["details"]["curBucket"] = 0
                  rv["details"]["curBucketName"] = ""
  
                  if "bucketsCount" in task["detailedProgress"]:
                      rv["details"]["totalBuckets"] = task["detailedProgress"]["bucketsCount"]
  
                  if "bucketNumber" in task["detailedProgress"]:
                      rv["details"]["curBucket"] = task["detailedProgress"]["bucketNumber"]
  
                  if "bucket" in task["detailedProgress"]:
                      rv["details"]["curBucketName"] = task["detailedProgress"]["bucket"]
  
                  acc = 0
                  if "perNode" in task["detailedProgress"]:
  
                      for _, node in task["detailedProgress"]["perNode"].items():
                          acc += node["ingoing"]["docsTotal"] - node["ingoing"]["docsTransferred"]
                          acc += node["outgoing"]["docsTotal"] - node["outgoing"]["docsTransferred"]
  
                  rv["details"]["docsRemaining"] = acc
              elif task["status"] == "notRunning":
                  rv["status"] = task["status"]
                  rv["msg"] = "Rebalance is not running"
                  if "statusIsStale" in task:
                      if task["statusIsStale"] or task["statusIsStale"] == "true":
                          rv["status"] = "stale"
                          rv["msg"] = "Current status is stale, please retry"
  
              break
  
          return rv, None
  
      # otpNode should only be printed out or handed back to ns_server
      # It should never be used to create a connection to a node
      def _get_otps_names(self, eject_nodes=[], failover_nodes=[], readd_nodes=[]):
          result, errors = self.pools('default')
          if errors:
              return None, None, None, None, None, errors
  
          all = list()
          eject = list()
          failover = list()
          readd = list()
          hostnames = list()
          for node in result["nodes"]:
              if "otpNode" not in node:
                  return [], [], [], [], [], ["Unable to get otp names"]
              all.append(node['otpNode'])
              hostnames.append(node['hostname'])
              if node['hostname'] in eject_nodes:
                  eject.append(node['otpNode'])
              if node['hostname'] in failover_nodes:
                  if node['clusterMembership'] != 'active':
                      return [], [], [], [], [], ["Can't failover a node that isn't in the cluster"]
                  else:
                      failover.append((node['otpNode'], node['status']))
              _, host = node['otpNode'].split('@')
              hostport = f'{host}:8091'
              if node['hostname'] in readd_nodes or hostport in readd_nodes:
                  readd.append(node['otpNode'])
  
          return all, eject, failover, readd, hostnames, None
  
      def create_bucket(self, name, bucket_type, storage_type, memory_quota,
                        eviction_policy, replicas, replica_indexes,
                        threads_number, conflict_resolution, flush_enabled,
                        max_ttl, compression_mode, sync, db_frag_perc, db_frag_size, view_frag_perc,
                        view_frag_size, from_hour, from_min, to_hour, to_min,
                        abort_outside, paralleldb_and_view_compact, purge_interval, timeout=60):
          url = f'{self.hostname}/pools/default/buckets'
  
          if name is None:
              return None, ["The bucket name is required when creating a bucket"]
          if bucket_type is None:
              return None, ["The bucket type is required when creating a bucket"]
          if memory_quota is None:
              return None, ["The bucket memory quota is required when creating a bucket"]
  
          params = {"name": name,
                    "bucketType": bucket_type,
                    "ramQuotaMB": memory_quota}
  
          if eviction_policy is not None:
              params["evictionPolicy"] = eviction_policy
          if replicas is not None:
              params["replicaNumber"] = replicas
          if replica_indexes is not None:
              params["replicaIndex"] = replica_indexes
          if threads_number is not None:
              params["threadsNumber"] = threads_number
          if conflict_resolution is not None:
              params["conflictResolutionType"] = conflict_resolution
          if flush_enabled is not None:
              params["flushEnabled"] = flush_enabled
          if max_ttl is not None:
              params["maxTTL"] = max_ttl
          if compression_mode is not None:
              params["compressionMode"] = compression_mode
          if storage_type is not None:
              params["storageBackend"] = storage_type
  
          if bucket_type == "couchbase":
              if (db_frag_perc is not None or db_frag_size is not None or view_frag_perc is not None or
                      view_frag_size is not None or from_hour is not None or from_min is not None or
                      to_hour is not None or to_min is not None or abort_outside is not None or
                      paralleldb_and_view_compact is not None or purge_interval is not None):
                  params["autoCompactionDefined"] = "true"
                  params["parallelDBAndViewCompaction"] = "false"
              if db_frag_perc is not None:
                  params["databaseFragmentationThreshold[percentage]"] = db_frag_perc
              if db_frag_size is not None:
                  params["databaseFragmentationThreshold[size]"] = db_frag_size * 1024 ** 2
              if view_frag_perc is not None:
                  params["viewFragmentationThreshold[percentage]"] = view_frag_perc
              if view_frag_size is not None:
                  params["viewFragmentationThreshold[size]"] = view_frag_size * 1024 ** 2
              if to_min is not None:
                  params["allowedTimePeriod[toMinute]"] = to_min
              if to_hour is not None:
                  params["allowedTimePeriod[toHour]"] = to_hour
              if from_min is not None:
                  params["allowedTimePeriod[fromMinute]"] = from_min
              if from_hour is not None:
                  params["allowedTimePeriod[fromHour]"] = from_hour
              if abort_outside is not None:
                  params["allowedTimePeriod[abortOutside]"] = abort_outside
              if paralleldb_and_view_compact is not None:
                  params["parallelDBAndViewCompaction"] = paralleldb_and_view_compact
  
          if bucket_type != "memcached" and purge_interval is not None:
                  params["purgeInterval"] = purge_interval
  
          result, errors = self._post_form_encoded(url, params)
          if errors:
              return None, errors
  
          if sync:
              all_node_ready = False
              start = time.time()
              while (time.time() - start) <= timeout and not all_node_ready:
                  buckets, errors = self.list_buckets()
                  if name not in buckets:
                      time.sleep(1)
                      continue
  
                  url = f'{self.hostname}/pools/default/buckets/{name}'
                  content, errors = self._get(url)
                  if errors:
                      return None, errors
  
                  all_node_ready = True
                  for node in content["nodes"]:
                      if node["status"] != "healthy":
                          all_node_ready = False
                          break
                  if not all_node_ready:
                      time.sleep(1)
  
              if not all_node_ready:
                  return None, [f'Bucket created, but not ready after {timeout} seconds']
  
          return result, None
  
      def edit_bucket(self, name, memory_quota, eviction_policy,
                      replicas, threads_number, flush_enabled, max_ttl,
                      compression_mode, remove_port, db_frag_perc, db_frag_size, view_frag_perc,
                      view_frag_size, from_hour, from_min, to_hour, to_min,
                      abort_outside, paralleldb_and_view_compact, purge_interval):
          url = f'{self.hostname}/pools/default/buckets/{name}'
  
          if name is None:
              return None, ["The bucket name is required when editing a bucket"]
  
          params = {}
          if memory_quota is not None:
              params["ramQuotaMB"] = memory_quota
          if eviction_policy is not None:
              params["evictionPolicy"] = eviction_policy
          if replicas is not None:
              params["replicaNumber"] = replicas
          if threads_number is not None:
              params["threadsNumber"] = threads_number
          if flush_enabled is not None:
              params["flushEnabled"] = flush_enabled
          if max_ttl is not None:
              params["maxTTL"] = max_ttl
          if compression_mode is not None:
              params["compressionMode"] = compression_mode
          if remove_port:
              params["proxyPort"] = "none"
          if (db_frag_perc is not None or db_frag_size is not None or view_frag_perc is not None or
                  view_frag_size is not None or from_hour is not None or from_min is not None or
                  to_hour is not None or to_min is not None or abort_outside is not None or
                  paralleldb_and_view_compact is not None or purge_interval is not None):
              params["autoCompactionDefined"] = "true"
              params["parallelDBAndViewCompaction"] = "false"
          if db_frag_perc is not None:
              params["databaseFragmentationThreshold[percentage]"] = db_frag_perc
          if db_frag_size is not None:
              params["databaseFragmentationThreshold[size]"] = db_frag_size * 1024 ** 2
          if view_frag_perc is not None:
              params["viewFragmentationThreshold[percentage]"] = view_frag_perc
          if view_frag_size is not None:
              params["viewFragmentationThreshold[size]"] = view_frag_size * 1024 ** 2
          if to_min is not None:
              params["allowedTimePeriod[toMinute]"] = to_min
          if to_hour is not None:
              params["allowedTimePeriod[toHour]"] = to_hour
          if from_min is not None:
              params["allowedTimePeriod[fromMinute]"] = from_min
          if from_hour is not None:
              params["allowedTimePeriod[fromHour]"] = from_hour
          if abort_outside is not None:
              params["allowedTimePeriod[abortOutside]"] = abort_outside
          if paralleldb_and_view_compact is not None:
              params["parallelDBAndViewCompaction"] = paralleldb_and_view_compact
          if purge_interval is not None:
              params["purgeInterval"] = purge_interval
  
          return self._post_form_encoded(url, params)
  
      def delete_bucket(self, name):
          url = f'{self.hostname}/pools/default/buckets/{name}'
          return self._delete(url, None)
  
      def flush_bucket(self, name):
          if name is None:
              return None, ["The bucket name is required when flushing a bucket"]
  
          url = f'{self.hostname}/pools/default/buckets/{name}/controller/doFlush'
          return self._post_form_encoded(url, None)
  
      def compact_bucket(self, name, data_only, view_only):
          if data_only and not view_only:
              url = f'{self.hostname}/pools/default/buckets/{name}/controller/compactDatabases'
              return self._post_form_encoded(url, None)
          elif view_only and not data_only:
              url = f'{self.hostname}/pools/default/buckets/{name}/ddocs'
              ddocs, errors = self._get(url)
              if errors:
                  return None, errors
  
              for row in ddocs["rows"]:
                  url = self.hostname + row["controllers"]["compact"]
                  _, errors = self._post_form_encoded(url, None)
                  if errors:
                      return None, errors
              return None, None
          elif not data_only and not view_only:
              url = f'{self.hostname}/pools/default/buckets/{name}/controller/compactBucket'
              return self._post_form_encoded(url, None)
          else:
              return None, ["Cannot compact data only and view only, pick one or neither"]
  
      def list_buckets(self, extended=False):
          url = f'{self.hostname}/pools/default/buckets'
          result, errors = self._get(url)
          if errors:
              return None, errors
  
          if extended:
              return result, errors
  
          names = list()
          for bucket in result:
              names.append(bucket["name"])
  
          return names, None
  
      def get_bucket(self, name):
          url = f'{self.hostname}/pools/default/buckets'
          result, errors = self._get(url)
          if errors:
              return None, errors
  
          for bucket in result:
              if bucket["name"] == name:
                  return bucket, None
  
          return None, ["Bucket not found"]
  
      def set_data_paths(self, data_path, index_path, cbas_path, eventing_path, java_home):
          url = f'{self.hostname}/nodes/self/controller/settings'
          params = dict()
  
          if data_path is not None:
              params["path"] = data_path
  
          if index_path is not None:
              params["index_path"] = index_path
  
          if cbas_path is not None:
              params["cbas_path"] = cbas_path
  
          if eventing_path is not None:
              params["eventing_path"] = eventing_path
  
          if java_home is not None:
              params["java_home"] = java_home
  
          return self._post_form_encoded(url, params)
  
      def node_info(self):
          url = f'{self.hostname}/nodes/self'
          return self._get(url)
  
      def get_babysitter_cookie(self):
          url = f'{self.hostname}/diag/eval'
          payload = \
              '{json, atom_to_binary(ns_server:get_babysitter_cookie(), latin1)}.'
          return self._post_form_encoded(url, payload)
  
      def set_hostname(self, hostname):
          url = f'{self.hostname}/node/controller/rename'
          params = { "hostname": hostname }
          return self._post_form_encoded(url, params)
  
      def stop_rebalance(self):
          params = {"allowUnsafe": "true"}
          url = f'{self.hostname}/controller/stopRebalance'
          return self._post_form_encoded(url, params)
  
      def create_server_group(self, name):
          url = f'{self.hostname}/pools/default/serverGroups'
          params = { "name": name }
          return self._post_form_encoded(url, params)
  
      def delete_server_group(self, name):
          uri, errors = self._get_server_group_uri(name)
          if errors:
              return None, errors
  
          url = self.hostname + uri
          params = { "name": name }
          return self._delete(url, params)
  
      def rename_server_group(self, name, new_name):
          uri, errors = self._get_server_group_uri(name)
          if errors:
              return None, errors
  
          url = self.hostname + uri
          params = { "name": new_name }
          return self._put(url, params)
  
      def move_servers_between_groups(self, servers, from_group, to_group):
          groups, errors = self.get_server_groups()
          if errors:
              return None, errors
  
          # Find the groups to move servers between
          move_from_group = None
          move_to_group = None
          for group in groups["groups"]:
              if from_group == group['name']:
                  move_from_group = group
              if to_group == group['name']:
                  move_to_group = group
  
          if move_from_group is None:
              return None, [f'Group to move servers from `{from_group}` not found']
          if move_to_group is None:
              return None, [f'Group to move servers to `{to_group}` not found']
  
          # Find the servers to move in the from group
          nodes_to_move = []
          for server in servers:
              found = False
              for node in move_from_group["nodes"]:
                  if server == node["hostname"]:
                      nodes_to_move.append(node)
                      move_from_group["nodes"].remove(node)
                      found = True
  
              if not found:
                  return None, [f"Can't move {server} because it doesn't exist in '{from_group}'"]
  
          # Move the servers to the to group
          for node in nodes_to_move:
              move_to_group["nodes"].append(node)
  
          url = self.hostname + groups["uri"]
          return self._put_json(url, groups)
  
  
      def _get_server_group_uri(self, name):
          groups, errors = self.get_server_groups()
          if errors:
              return errors
  
          for group in groups["groups"]:
              if name == group["name"]:
                  return group["uri"], None
          return None, [f'Group `{name}` not found']
  
      def delete_rbac_user(self, username, auth_domain):
          url = f'{self.hostname}/settings/rbac/users/{auth_domain}/{username}'
          return self._delete(url, None)
  
      def list_rbac_users(self):
          url = f'{self.hostname}/settings/rbac/users'
          return self._get(url)
  
      def my_roles(self):
          url = f'{self.hostname}/whoami'
          return self._get(url)
  
      def set_rbac_user(self, username, password, name, roles, auth_domain, groups):
          if auth_domain is None:
              return None, ["The authentication type is required"]
  
          if username is None:
              return None, ["The username is required"]
  
          url = f'{self.hostname}/settings/rbac/users/{auth_domain}/{username}'
  
          defaults, errors = self._get(url)
          if errors and errors[0] == '"Unknown user."':
              defaults = {}
          elif errors:
              return None, errors
  
          params = {}
          if name is not None:
              params['name'] = name
          elif 'name' in defaults:
              params['name'] = defaults['name']
          if password is not None:
              params['password'] = password
          if roles is not None:
              params['roles'] = roles;
          elif 'roles' in defaults:
              params['roles'] = self._format_user_roles(defaults['roles'])
          if groups is not None:
              params['groups'] = groups
          elif 'groups' in defaults:
              params['groups'] = ','.join(defaults['groups'])
  
          return self._put(url, params)
  
      def _format_user_roles(self, roles):
          directly_assigned = filter(lambda r: any(o for o in r['origins'] if o['type'] == 'user'), roles)
          return ",".join(self._format_role(r) for r in directly_assigned)
  
      def _format_role(self, role):
          return f'{role["role"]}' + (f'[{role["bucket_name"]}]' if 'bucket_name' in role else '')
  
      def get_user_group(self, group):
          if group is None:
              return None, ['group name is required']
  
          url = f'{self.hostname}/settings/rbac/groups/{group}'
          return self._get(url)
  
      def delete_user_group(self, group):
          if group is None:
              return None, ['group name is required']
  
          url = f'{self.hostname}/settings/rbac/groups/{group}'
          return self._delete(url, dict())
  
      def set_user_group(self, group, roles, description, ldap_ref):
          if group is None:
              return None, ['Group name is required']
  
          url = f'{self.hostname}/settings/rbac/groups/{group}'
  
          defaults, errors = self._get(url)
          if errors and errors[0] == '"Unknown group."':
              defaults = {}
          elif errors:
              return None, errors
  
          params = {}
          if roles is not None:
              params['roles'] = roles
          elif 'roles' in defaults:
              params['roles'] = ",".join(self._format_role(r) for r in defaults['roles'])
          if description is not None:
              params['description'] = description
          elif 'description' in defaults:
              params['description'] = defaults['description']
          if ldap_ref is not None:
              params['ldap_group_ref'] = ldap_ref
          elif 'ldap_group_ref' in defaults:
              params['ldap_group_ref'] = defaults['ldap_group_ref']
  
          return self._put(url, params)
  
      def list_user_groups(self):
          url = f'{self.hostname}/settings/rbac/groups'
          return self._get(url)
  
      def get_password_policy(self):
          url = f'{self.hostname}/settings/passwordPolicy'
          return self._get(url)
  
      def set_security_settings(self, disable_http_ui, cluster_encryption_level, tls_min_version,
                                honor_order, cipher_suites, disable_www_authenticate):
          url = f'{self.hostname}/settings/security'
          params = {}
  
          if disable_http_ui:
              params['disableUIOverHttp'] = disable_http_ui
          if cluster_encryption_level:
              params['clusterEncryptionLevel'] = cluster_encryption_level
          if tls_min_version:
              params['tlsMinVersion'] = tls_min_version
          if honor_order:
              params['honorCipherOrder'] = honor_order
          if cipher_suites:
              params['cipherSuites'] = cipher_suites
          if disable_www_authenticate:
              params['disableWWWAuthenticate'] = disable_www_authenticate
  
          return self._post_form_encoded(url, params)
  
      def get_security_settings(self):
          return self._get(f'{self.hostname}/settings/security')
  
      def set_password_policy(self, min_length, upper_case, lower_case, digit,
                              special_char):
          url = self.hostname + '/settings/passwordPolicy'
  
          params = dict()
          if min_length:
              params["minLength"] = min_length
          if upper_case:
              if upper_case == "1":
                  params["enforceUppercase"] = "true"
              else:
                  params["enforceUppercase"] ="false"
          if lower_case:
              if lower_case == "1":
                  params["enforceLowercase"] = "true"
              else:
                  params["enforceLowercase"] ="false"
          if digit:
              if digit == "1":
                  params["enforceDigits"] = "true"
              else:
                  params["enforceDigits"] = "false"
          if special_char:
              if special_char == "1":
                  params["enforceSpecialChars"] = "true"
              else:
                  params["enforceSpecialChars"] ="false"
  
          return self._post_form_encoded(url, params)
  
      def set_audit_settings(self, enabled, log_path, rotate_interval, rotate_size, disabled, disabled_users):
          url = f'{self.hostname}/settings/audit'
  
          params = dict()
          if enabled:
              params["auditdEnabled"] = enabled
          if log_path:
              params["logPath"] = log_path
          if rotate_interval:
              params["rotateInterval"] = rotate_interval
          if rotate_size:
              params["rotateSize"] = rotate_size
          if disabled:
              params["disabled"] = disabled
          if disabled_users:
              params["disabledUsers"] = disabled_users
  
          if "logPath" not in params and "auditdEnabled" in params and params["auditdEnabled"] == "true":
              return None, ["The audit log path must be specified when auditing is first set up"]
  
          return self._post_form_encoded(url, params)
  
      def get_audit_settings(self):
          return self._get(f'{self.hostname}/settings/audit')
  
      def get_id_descriptors(self):
          return self._get(f'{self.hostname}/settings/audit/descriptors')
  
      def set_autofailover_settings(self, enabled, timeout, failoverOfServerGroups, maxCount,
                                    failoverOnDataDiskIssuesEnabled, failoverOnDataDiskIssuesTimePeriod,
                                    canAbortRebalance):
          url = self.hostname + '/settings/autoFailover'
  
          params = dict()
          if enabled:
              params["enabled"] = enabled
          if timeout:
              params["timeout"] = timeout
          if failoverOfServerGroups:
              params["failoverServerGroup"] = failoverOfServerGroups
          if failoverOnDataDiskIssuesEnabled:
              params["failoverOnDataDiskIssues[enabled]"] = failoverOnDataDiskIssuesEnabled
          if maxCount:
              params["maxCount"] = maxCount
          if failoverOnDataDiskIssuesTimePeriod:
              params["failoverOnDataDiskIssues[timePeriod]"] = failoverOnDataDiskIssuesTimePeriod
          if canAbortRebalance:
              params["canAbortRebalance"] = canAbortRebalance
  
          return self._post_form_encoded(url, params)
  
      def set_dp_mode(self):
          url = f'{self.hostname}/settings/developerPreview'
          params = {'enabled': 'true'}
          return self._post_form_encoded(url, params)
  
      def set_autoreprovision_settings(self, enabled, max_nodes):
          url = self.hostname + '/settings/autoReprovision'
  
          params = dict()
          if enabled:
              params["enabled"] = enabled
          if max_nodes:
              params["maxNodes"] = max_nodes
  
          return self._post_form_encoded(url, params)
  
      def set_compaction_settings(self, dbFragPerc, dbFragSize, viewFragPerc, viewFragSize,
                                  fromHour, fromMin, toHour, toMin, abortOutside,
                                  parallelDBAndViewCompact, purgeInterval, gsiMode, gsiPerc,
                                  gsiInterval, gsiFromHour, gsiFromMin, gsiToHour, gsiToMin,
                                  enableGsiAbort):
          url = f'{self.hostname}/controller/setAutoCompaction'
          params = dict()
  
          if dbFragPerc is not None:
              params["databaseFragmentationThreshold[percentage]"] = dbFragPerc
          if dbFragSize is not None:
              params["databaseFragmentationThreshold[size]"] = dbFragSize
          if viewFragPerc is not None:
              params["viewFragmentationThreshold[percentage]"] = viewFragPerc
          if viewFragSize is not None:
              params["viewFragmentationThreshold[size]"] = viewFragSize
          if fromHour is not None:
              params["allowedTimePeriod[fromHour]"] = fromHour
          if fromMin is not None:
              params["allowedTimePeriod[fromMinute]"] = fromMin
          if toHour is not None:
              params["allowedTimePeriod[toHour]"] = toHour
          if toMin is not None:
              params["allowedTimePeriod[toMinute]"] = toMin
          if abortOutside is not None:
              params["allowedTimePeriod[abortOutside]"] = abortOutside
          if parallelDBAndViewCompact is not None:
              params["parallelDBAndViewCompaction"] = parallelDBAndViewCompact
          if purgeInterval is not None:
              params["purgeInterval"] = purgeInterval
          if gsiMode is not None:
              params["indexCompactionMode"] = gsiMode
          if gsiPerc is not None:
              params["indexFragmentationThreshold[percentage]"] = gsiPerc
          if gsiInterval is not None:
              params["indexCircularCompaction[daysOfWeek]"] = gsiInterval
          if gsiFromHour is not None:
              params["indexCircularCompaction[interval][fromHour]"] = gsiFromHour
          if gsiFromMin is not None:
              params["indexCircularCompaction[interval][fromMinute]"] = gsiFromMin
          if gsiToHour is not None:
              params["indexCircularCompaction[interval][toHour]"] = gsiToHour
          if gsiToMin is not None:
              params["indexCircularCompaction[interval][toMinute]"] = gsiToMin
          if enableGsiAbort is not None:
              params["indexCircularCompaction[interval][abortOutside]"] = enableGsiAbort
  
          return self._post_form_encoded(url, params)
  
      def set_index_settings(self, storageMode, maxRollbackPoints, stableSnapInterval,
                             memSnapInterval, threads, logLevel):
          """ Sets global index settings"""
          params = dict()
          if storageMode is not None:
              params["storageMode"] = storageMode
          if maxRollbackPoints is not None:
              params["maxRollbackPoints"] = maxRollbackPoints
          if stableSnapInterval is not None:
              params["stableSnapshotInterval"] = stableSnapInterval
          if memSnapInterval is not None:
              params["memorySnapshotInterval"] = memSnapInterval
          if threads is not None:
              params["indexerThreads"] = threads
          if logLevel is not None:
              params["logLevel"] = logLevel
  
          url = f'{self.hostname}/settings/indexes'
          return self._post_form_encoded(url, params)
  
      def set_alert_settings(self, enabled_email_alerts, email_recipients,
                             email_sender, email_user, email_pass, email_host,
                             email_port, email_encrypted, alerts):
          url = f'{self.hostname}/settings/alerts'
          params = dict()
  
          if enabled_email_alerts:
              params["enabled"] = enabled_email_alerts
          if email_recipients:
              params["recipients"] = email_recipients
          if email_sender:
              params["sender"] = email_sender
          if email_user:
              params["emailUser"] = email_user
          if email_pass:
              params["emailPass"] = email_pass
          if email_host:
              params["emailHost"] = email_host
          if email_port:
              params["emailPort"] = email_port
          if email_encrypted:
              params["emailEncrypt"] = email_encrypted
          if alerts is not None:
              params["alerts"] = alerts
  
          return self._post_form_encoded(url, params)
  
      def index_settings(self):
          """ Retrieves the index settings
  
              Returns a map of all global index settings"""
          url = f'{self.hostname}/settings/indexes'
          return self._get(url)
  
      def sasl_settings(self, enabled, read_only_admins, admins):
          """ Sets Sasl Settings
  
          enabled - The string "true" or "false"
          admins - A new line separated list or the string "asterisk"
          read_only_admins - A new line separated list or the string "asterisk"
          """
  
          url = f'{self.hostname}/settings/saslauthdAuth'
          params = { "enabled": enabled }
  
          if read_only_admins is not None:
              params["roAdmins"] = read_only_admins
          if admins is not None:
              params["admins"] = admins
  
          return self._post_form_encoded(url, params)
  
      def get_ldap(self):
          url = f'{self.hostname}/settings/ldap'
          return self._get(url)
  
      def ldap_settings(self, authentication_enabled, authorization_enabled, hosts, port, encryption, user_dn_mapping,
                      timeout, max_parallel, max_cache, cache_lifetime, query_dn, query_pass, group_query,
                      nested_groups, nested_groups_max_depth, server_ca_ver, ca):
          url = f'{self.hostname}/settings/ldap'
  
          params = {}
  
          if authentication_enabled is not None:
              params['authenticationEnabled'] = authentication_enabled
          if authorization_enabled is not None:
              params['authorizationEnabled'] = authorization_enabled
          if hosts is not None:
              params['hosts'] = hosts
          if port is not None:
              params['port'] = port
          if encryption is not None:
              params['encryption'] = encryption
          if user_dn_mapping is not None:
              params['userDNMapping'] = user_dn_mapping
          if timeout is not None:
              params['requestTimeout'] = timeout
          if max_parallel is not None:
              params['maxParallelConnections'] = max_parallel
          if max_cache is not None:
              params['maxCacheSize'] = max_cache
          if cache_lifetime is not None:
              params['cacheValueLifetime'] = cache_lifetime
          if query_dn is not None:
              params['bindDN'] = query_dn
          if query_pass is not None:
              params['bindPass'] = query_pass
          if group_query is not None:
              params['groupsQuery'] = group_query
          if nested_groups is not None:
              params['nestedGroupsEnabled'] = nested_groups
          if nested_groups_max_depth is not None:
              params['nestedGroupsMaxDepth'] = nested_groups_max_depth
          if server_ca_ver is not None:
              params['serverCertValidation'] = server_ca_ver
          if ca and server_ca_ver != 'false':
              params['cacert'] = ca
  
          if len(params) == 0:
              return None, ['Please provide at least one option to set']
  
          return self._post_form_encoded(url, params)
  
      def setRoles(self, userList, roleList, userNameList):
          # we take a comma-delimited list of roles that needs to go into a dictionary
          paramDict = {"roles" : roleList}
          userIds = []
          userNames = []
          userF = io.StringIO(userList)
          for idList in csv.reader(userF, delimiter=','):
              userIds.extend(idList)
  
          # did they specify user names?
          if userNameList != None:
              userNameF = io.StringIO(userNameList)
              for nameList in csv.reader(userNameF, delimiter=','):
                  userNames.extend(nameList)
              if len(userNames) != len(userIds):
                  return None, [f'Error: specified {len(userIds)} user ids and {len(userNames)} user names, must'
                                f' have the same number of each.']
  
          # did they specify user names?
          # but we need a separate REST call for each user in the comma-delimited user list
          for index in range(len(userIds)):
              user = userIds[index]
              paramDict["id"] = user
              if len(userNames) > 0:
                  paramDict["name"] = userNames[index]
              url = f'{self.hostname}/settings/rbac/users/{user}'
              data, errors = self._put(url,paramDict)
              if errors:
                  return data, errors
  
          return data, errors
  
      def deleteRoles(self,userList):
          # need a separate REST call for each user in the comma-delimited user list
          userF = io.StringIO(userList)
          reader = csv.reader(userF, delimiter=',')
          for users in reader:
              for user in users:
                  url = f'{self.hostname}/settings/rbac/users/{user}'
                  data, errors = self._delete(url, None)
                  if errors:
                      return data, errors
  
          return data, errors
  
          url = f'{self.hostname}/settings/rbac/users'
          data, errors = self._get(url)
  
          return data, errors
  
      def retrieve_cluster_certificate(self, extended=False):
          """ Retrieves the current cluster certificate
  
          Gets the current cluster certificate. If extended is set tot True then
          we return the extended certificate which contains the certificate type,
          certicicate key, expiration, subject, and warnings."""
          url = f'{self.hostname}/pools/default/certificate'
          if extended:
              url += '?extended=true'
          return self._get(url)
  
      def regenerate_cluster_certificate(self):
          """ Regenerates the cluster certificate
  
          Regenerates the cluster certificate and returns the new certificate."""
          url = f'{self.hostname}/controller/regenerateCertificate'
          return self._post_form_encoded(url, None)
  
      def upload_cluster_certificate(self, certificate):
          """ Uploads a new cluster certificate"""
          url = f'{self.hostname}/controller/uploadClusterCA'
          return self._post_form_encoded(url, certificate)
  
      def retrieve_node_certificate(self, node):
          """ Retrieves the current node certificate
  
          Returns the current node certificate"""
          url = f'{self.hostname}/pools/default/certificate/node/{node}'
          return self._get(url)
  
      def set_node_certificate(self):
          """Activates the current node certificate
  
          Grabs chain.pem and pkey.pem from the <data folder>/inbox/ directory and
          applies them to the node. chain.pem contains the chain encoded certificates
          starting from the node certificat and ending with the last intermediate
          certificate before cluster CA. pkey.pem contains the pem encoded private
          key for node certifiactes. Both files should exist on the server before
          this API is called."""
          url = f'{self.hostname}/node/controller/reloadCertificate'
          return self._post_form_encoded(url, None)
  
      def set_client_cert_auth(self, config):
          """Enable/disable the client cert auth"""
          url = f'{self.hostname}/settings/clientCertAuth'
          return self._post_json(url, config)
  
      def retrieve_client_cert_auth(self):
          url = f'{self.hostname}/settings/clientCertAuth'
          return self._get(url)
  
      def create_xdcr_reference(self, name, hostname, username, password, encrypted,
                                encryptionType, certificate, clientCertificate, clientKey):
          return self._set_xdcr_reference(False, name, hostname, username,
                                          password, encrypted, encryptionType,
                                          certificate, clientCertificate, clientKey)
  
      def edit_xdcr_reference(self, name, hostname, username, password, encrypted,
                              encryptionType, certificate, clientCertificate, clientKey):
          return self._set_xdcr_reference(True, name, hostname, username,
                                          password, encrypted, encryptionType,
                                          certificate, clientCertificate, clientKey)
  
      def _set_xdcr_reference(self, edit, name, hostname, username, password,
                              encrypted, encryptionType, certificate, clientCertificate, clientKey):
          url = f'{self.hostname}/pools/default/remoteClusters'
          params = {}
  
          if edit:
              url += f'/{urllib.parse.quote(name)}'
  
          if name is not None:
              params["name"] = name
          if hostname is not None:
              params["hostname"] = hostname
          if username is not None:
              params["username"] = username
          if password is not None:
              params["password"] = password
          if encrypted is not None:
              params["demandEncryption"] = encrypted
          if encryptionType is not None:
              params["encryptionType"] = encryptionType
          if certificate is not None:
              params["certificate"] = certificate
          if clientCertificate:
              params['clientCertificate'] = clientCertificate
          if clientKey:
              params['clientKey'] = clientKey
  
          return self._post_form_encoded(url, params)
  
      def delete_xdcr_reference(self, name):
          url = f'{self.hostname}/pools/default/remoteClusters/{urllib.parse.quote(name)}'
          return self._delete(url, None)
  
      def list_xdcr_references(self):
          url = f'{self.hostname}/pools/default/remoteClusters/'
          return self._get(url)
  
      def xdcr_replicator_settings(self, chk_interval, worker_batch_size,
                                   doc_batch_size, fail_interval, replication_thresh,
                                   src_nozzles, dst_nozzles, usage_limit, compression,
                                   log_level, stats_interval, replicator_id, filter, filter_skip, priority,
                                   reset_expiry, filter_del, filter_exp):
  
          url = f'{self.hostname}/settings/replications/{urllib.parse.quote_plus(replicator_id)}'
          params = self._get_xdcr_params(chk_interval, worker_batch_size, doc_batch_size,
                                         fail_interval, replication_thresh, src_nozzles,
                                         dst_nozzles, usage_limit, compression, log_level,
                                         stats_interval)
          if filter is not None:
              params['filterExpression'] = filter
              filter_numeric = "0"
              if filter_skip:
                  filter_numeric = "1"
              params['filterSkipRestream'] = filter_numeric
          if priority:
              params['priority'] = priority
          if reset_expiry:
              params['filterBypassExpiry'] = 'true' if reset_expiry == '1' else 'false'
          if filter_del:
              params['filterDeletion'] = 'true' if filter_del == '1' else 'false'
          if filter_exp:
              params['filterExpiration'] = 'true' if filter_exp == '1' else 'false'
  
          return self._post_form_encoded(url, params)
  
      def xdcr_global_settings(self, chk_interval, worker_batch_size, doc_batch_size,
                               fail_interval, replication_threshold, src_nozzles,
                               dst_nozzles, usage_limit, compression, log_level, stats_interval, max_proc):
          url = f'{self.hostname}/settings/replications'
          params = self._get_xdcr_params(chk_interval, worker_batch_size, doc_batch_size,
                                         fail_interval, replication_threshold, src_nozzles,
                                         dst_nozzles, usage_limit, compression, log_level, stats_interval)
          if max_proc:
              params['goMaxProcs'] = max_proc
  
          return self._post_form_encoded(url, params)
  
      def _get_xdcr_params(self, chk_interval, worker_batch_size, doc_batch_size,
                           fail_interval, replication_threshold, src_nozzles,
                           dst_nozzles, usage_limit, compression, log_level, stats_interval):
          params = {}
          if chk_interval is not None:
              params["checkpointInterval"] = chk_interval
          if worker_batch_size is not None:
              params["workerBatchSize"] = worker_batch_size
          if doc_batch_size is not None:
              params["docBatchSizeKb"] = doc_batch_size
          if fail_interval is not None:
              params["failureRestartInterval"] = fail_interval
          if replication_threshold is not None:
              params["optimisticReplicationThreshold"] = replication_threshold
          if src_nozzles is not None:
              params["sourceNozzlePerNode"] = src_nozzles
          if dst_nozzles is not None:
              params["targetNozzlePerNode"] = dst_nozzles
          if usage_limit is not None:
              params["networkUsageLimit"] = usage_limit
          if compression is not None:
              params["compressionType"] = compression
          if log_level is not None:
              params["logLevel"] = log_level
          if stats_interval is not None:
              params["statsInterval"] = stats_interval
          return params
  
      def  create_xdcr_replication(self, name, to_bucket, from_bucket, filter, rep_mode, compression,
                                   reset_expiry, filter_del, filter_exp):
          url = f'{self.hostname}/controller/createReplication'
          params = { "replicationType": "continuous" }
  
          if to_bucket is not None:
              params["toBucket"] = to_bucket
          if name is not None:
              params["toCluster"] = name
          if from_bucket is not None:
              params["fromBucket"] = from_bucket
          if rep_mode is not None:
              params["type"] = rep_mode
          if filter is not None:
              params["filterExpression"] = filter
          if compression is not None:
              params["compressionType"] = compression
          if reset_expiry:
              params['filterBypassExpiry'] = 'true' if reset_expiry == '1' else 'false'
          if filter_del:
              params['filterDeletion'] = 'true' if filter_del == '1' else 'false'
          if filter_exp:
              params['filterExpiration'] = 'true' if filter_exp == '1' else 'false'
  
          return self._post_form_encoded(url, params)
  
      def delete_xdcr_replicator(self, replicator_id):
          url =f'{self.hostname}/controller/cancelXDCR/{urllib.parse.quote_plus(replicator_id)}'
          return self._delete(url, None)
  
      def pause_xdcr_replication(self, replicator_id):
          url = f'{self.hostname}/settings/replications/{urllib.parse.quote_plus(replicator_id)}'
          params = { "pauseRequested": "true" }
          return self._post_form_encoded(url, params)
  
      def resume_xdcr_replication(self, replicator_id):
          url = f'{self.hostname}/settings/replications/{urllib.parse.quote_plus(replicator_id)}'
          params = { "pauseRequested": "false" }
          return self._post_form_encoded(url, params)
  
      def get_query_settings(self):
          url = f'{self.hostname}/settings/querySettings'
          return self._get(url)
  
      def post_query_settings(self, pipeline_batch, pipeline_cap, scan_cap, timeout, prepared_limit, completed_limit,
                              complete_threshold, log_level, max_parallelism, n1ql_feature_control):
          url = f'{self.hostname}/settings/querySettings'
          params = {}
          if pipeline_batch is not None:
              params['queryPipelineBatch'] = pipeline_batch
          if pipeline_cap is not None:
              params['queryPipelineCap'] = pipeline_cap
          if scan_cap is not None:
              params['queryScanCap'] = scan_cap
          if timeout is not None:
              params['queryTimeout'] = timeout
          if prepared_limit is not None:
              params['queryPreparedLimit'] = prepared_limit
          if completed_limit is not None:
              params['queryCompletedLimit'] = completed_limit
          if complete_threshold is not None:
              params['queryCompletedThreshold'] = complete_threshold
          if log_level is not None:
              params['queryLogLevel'] = log_level
          if max_parallelism is not None:
              params['queryMaxParallelism'] = max_parallelism
          if n1ql_feature_control is not None:
              params['queryN1QLFeatCtrl'] = n1ql_feature_control
  
          return self._post_form_encoded(url, params)
  
      def list_functions(self):
          hosts, errors = self.get_hostnames_for_service(EVENT_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(EVENT_SERVICE)
          url = f'{hosts[0]}/api/v1/functions'
          return self._get(url)
  
      def export_functions(self):
          hosts, errors = self.get_hostnames_for_service(EVENT_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(EVENT_SERVICE)
          url = f'{hosts[0]}/api/v1/export'
          return self._get(url)
  
      def import_functions(self, parms):
          hosts, errors = self.get_hostnames_for_service(EVENT_SERVICE)
          if errors:
              return None, errors
          if not hosts:
              raise ServiceNotAvailableException(EVENT_SERVICE)
          url = f'{hosts[0]}/api/v1/import'
          return self._post_json(url, parms)
  
      def delete_function(self, function):
          hosts, errors = self.get_hostnames_for_service(EVENT_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(EVENT_SERVICE)
          url = f'{hosts[0]}/api/v1/functions/{urllib.parse.quote_plus(function)}'
          return self._delete(url, None)
  
      def pause_resume_function(self, function_name, pause):
          hosts, errors = self.get_hostnames_for_service(EVENT_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(EVENT_SERVICE)
  
          url = f"{hosts[0]}/api/v1/functions/{urllib.parse.quote_plus(function_name)}/{'pause' if pause else 'resume'}"
          return self._post_json(url, None)
  
      def deploy_undeploy_function(self, function, deploy, boundary):
          hosts, errors = self.get_hostnames_for_service(EVENT_SERVICE)
          if errors:
              return None, errors
  
          if not hosts:
              raise ServiceNotAvailableException(EVENT_SERVICE)
  
          params = {}
          params["deployment_status"] = deploy
          params["processing_status"] = deploy
  
          if deploy:
              params["feed-boundary"] = boundary
  
          url = f'{hosts[0]}/api/v1/functions/{urllib.parse.quote_plus(function)}/settings'
          return self._post_json(url, params)
  
      def create_scope(self, bucket, scope):
          url = f'{self.hostname}/pools/default/buckets/{urllib.parse.quote_plus(bucket)}/collections'
          params = {"name": scope}
          return self._post_form_encoded(url, params)
  
      def drop_scope(self, bucket, scope):
          url = f'{self.hostname}/pools/default/buckets/{urllib.parse.quote_plus(bucket)}/collections/'\
              f'{urllib.parse.quote_plus(scope)}'
          return self._delete(url, None)
  
      def create_collection(self, bucket, scope, collection, max_ttl):
          url = f'{self.hostname}/pools/default/buckets/{urllib.parse.quote_plus(bucket)}/collections/' \
              f'{urllib.parse.quote_plus(scope)}'
          params = {"name": collection}
          if max_ttl:
              params["maxTTL"] = max_ttl
          return self._post_form_encoded(url, params)
  
      def drop_collection(self, bucket, scope, collection):
          url = f'{self.hostname}/pools/default/buckets/{urllib.parse.quote_plus(bucket)}/collections/'\
              f'{urllib.parse.quote_plus(scope)}/{urllib.parse.quote_plus(collection)}'
          return self._delete(url, None)
  
      def get_manifest(self, bucket):
          url = f'{self.hostname}/pools/default/buckets/{urllib.parse.quote_plus(bucket)}/collections'
          return self._get(url)
  
      def set_alternate_address(self, hostname, ports):
          url = f'{self.hostname}/node/controller/setupAlternateAddresses/external'
          params = {}
          if hostname:
              params['hostname'] = hostname
          if ports:
              for (name, value) in ports:
                  params[name] = value
  
          return self._put(url, params)
  
      def delete_alternate_address(self):
          url = f'{self.hostname}/node/controller/setupAlternateAddresses/external'
          return self._delete(url, {})
  
      def get_alternate_address(self):
          url = f'{self.hostname}/pools/default/nodeServices'
          node_service, error = self._get(url)
          if error:
              return None, error
          return node_service['nodesExt'], None
  
      def enable_external_listener(self, host=None, ipfamily=None, encryption=None):
          hostname = host if host else self.hostname
          url = f'{hostname}/node/controller/enableExternalListener'
          params = {}
          if ipfamily:
              params['afamily'] = ipfamily
          if encryption:
              params['nodeEncryption'] = encryption
          return self._post_form_encoded(url, params)
  
      def disable_external_listener(self, host=None, ipfamily=None, encryption=None):
          hostname = host if host else self.hostname
          url = f'{hostname}/node/controller/disableExternalListener'
          params = {}
          if ipfamily:
              params['afamily'] = ipfamily
          if encryption:
              params['nodeEncryption'] = encryption
          return self._post_form_encoded(url, params)
  
      def setup_net_config(self, host=None, ipfamily=None, encryption=None):
          hostname = host if host else self.hostname
          url = f'{hostname}/node/controller/setupNetConfig'
          params = {}
          if ipfamily:
              params['afamily'] = ipfamily
          if encryption:
              params['nodeEncryption'] = encryption
  
          return self._post_form_encoded(url, params)
  
      def node_get_address_family(self, host):
          node_data, err = self._get(f'{host}/pools/nodes')
          if err:
              return '', err
  
          for n in node_data['nodes']:
              if 'thisNode' in n and n['thisNode']:
                  if 'addressFamily' in n:
                      return n['addressFamily'], None
                  return '', [f'Node {host} must be version 6.5 or higher']
  
          return '', [f'Could not get data for {host}']
  
      def reset_cipher_suites(self):
          url = f'{self.hostname}/controller/resetCipherSuites'
          return self._post_form_encoded(url, None)
  
      # returns info about nodes even if cluster is not initialized yet
      def nodes_info(self):
          node_data, err = self.pools('nodes')
          if err and err[0] == '"unknown pool"':
              result, err = self.node_info()
              if err:
                  return None, err
              else:
                  return [result], None
          elif err:
              return None, err
  
          return node_data['nodes'], None
  
      # Low level methods for basic HTML operations
  
      @request
      def _get(self, url):
          if self.debug:
              print(f'GET {url}')
          response = requests.get(url, auth=(self.username, self.password), verify=self.caCert,
                                  cert=self.cert, timeout=self.timeout,
                                  headers=self.headers)
          return _handle_response(response, self.debug)
  
      @request
      def _post_form_encoded(self, url, params):
          if self.debug:
              if params is None:
                  params = {}
              print(f'POST {url} {urllib.parse.urlencode(params)}')
          response = requests.post(url, auth=(self.username, self.password), data=params,
                                   cert=self.cert, verify=self.caCert, timeout=self.timeout,
                                   headers=self.headers)
          return _handle_response(response, self.debug)
  
      @request
      def _post_json(self, url, params):
          if self.debug:
              if params is None:
                  params = {}
              print(f'POST {url} {json.dumps(params)}')
          response = requests.post(url, auth=(self.username, self.password), json=params,
                                   cert=self.cert, verify=self.caCert, timeout=self.timeout,
                                   headers=self.headers)
          return _handle_response(response, self.debug)
  
      @request
      def _put(self, url, params):
          if self.debug:
              if params is None:
                  params = {}
              print(f'PUT {url} {urllib.parse.urlencode(params)}')
          response = requests.put(url, params, auth=(self.username, self.password),
                                  cert=None, verify=self.caCert, timeout=self.timeout,
                                  headers=self.headers)
          return _handle_response(response, self.debug)
  
      @request
      def _put_json(self, url, params):
          if self.debug:
              if params is None:
                  params = {}
              print(f'PUT {url} {json.dumps(params)}')
          response = requests.put(url, auth=(self.username, self.password), json=params,
                                  cert=None, verify=self.caCert, timeout=self.timeout,
                                  headers = self.headers)
          return _handle_response(response, self.debug)
  
      @request
      def _delete(self, url, params):
          if self.debug:
              if params is None:
                  params = {}
              print(f'DELETE {url} {urllib.parse.urlencode(params)}')
          response = requests.delete(url, auth=(self.username, self.password), data=params,
                                     cert=None, verify=self.caCert, timeout=self.timeout,
                                     headers=self.headers)
          return _handle_response(response, self.debug)
  
  
  def _handle_response(response, debug):
      if debug:
          output = str(response.status_code)
          if response.headers:
              output += f', {response.headers}'
          if response.content:
              response.encoding = 'utf-8'
              output += f', {response.content}'
          print(output)
      if response.status_code in [200, 202]:
          if 'Content-Type' not in response.headers:
              return "", None
          if not response.content:
              return "", None
          if 'application/json' in response.headers['Content-Type']:
              return response.json(), None
          else:
              response.encoding = 'utf-8'
              return response.text, None
      elif response.status_code in [400, 404]:
          if 'application/json' in response.headers['Content-Type']:
              errors = response.json()
              if isinstance(errors, list):
                  return None, errors
              if "errors" in errors and isinstance(errors["errors"], list):
                  return None, errors["errors"]
              if isinstance(errors, dict):
                  if "errors" in errors and isinstance(errors["errors"], dict):
                      errors = errors["errors"]
                  rv = list()
                  for key, value in errors.items():
                      rv.append(key + " - " + str(value))
                  return None, rv
          return None, [response.text]
      elif response.status_code == 401:
          return None, [ERR_AUTH]
      elif response.status_code == 403:
          errors = response.json()
          return None, [errors["message"] + ": " + ", ".join(errors["permissions"])]
      # Error codes from Eventing service
      elif response.status_code  in [406, 422, 423]:
          errors = response.json()
          if "description" in errors:
              return None, [errors["description"]]
          return None, ['Received unexpected status %d' % response.status_code]
      # Error code from Eventing Service
      elif response.status_code == 207:
          errors = response.json()
          if isinstance(errors, list):
              rv = list()
              for error in errors:
                  if error['code'] == 20:
                      rv.append(error['info'])
              return None, rv
          else:
              return None, ['Received unexpected status %d' % response.status_code]
      elif response.status_code == 500:
          return None, [ERR_INTERNAL]
      else:
          return None, ['Received unexpected status %d' % response.status_code]
  