Reword confirmation message

- Implement the dedicated class that represents a manual failover
  request
- Move manual failover/switchover prechecks to the class method and use
  for both ctl and api
- Use a single parse_schedule function in both ctl and api
- Implement has_members_eligible_to_promote Ha method
- Fix get_members + role='any' exception msg

docs/citus.rst

@@ -140,7 +140,7 @@ An example of ``patronictl switchover`` on the worker cluster::
     | work2-1 | 172.27.0.5 | Sync Standby | running |  1 |         0 |
     | work2-2 | 172.27.0.7 | Leader       | running |  1 |           |
     +---------+------------+--------------+---------+----+-----------+
-    Are you sure you want to switchover cluster demo, demoting current primary work2-2? [y/N]: y
+    Are you sure you want to perform a switchover in the cluster demo, demoting current primary work2-2? [y/N]: y
     2022-12-22 07:02:40.33003 Successfully switched over to "work2-1"
     + Citus cluster: demo (group: 2, 7179854924063375386) ------+
     | Member  | Host       | Role    | State   | TL | Lag in MB |

patroni/api.py

@@ -12,7 +12,6 @@ import json
 import logging
 import time
 import traceback
-import dateutil.parser
 import datetime
 import os
 import socket
@@ -28,11 +27,11 @@ from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, TYPE_CH
 
 from . import psycopg
 from .__main__ import Patroni
-from .dcs import Cluster
 from .exceptions import PostgresConnectionException, PostgresException
+from .manual_failover import ManualFailover
 from .postgresql.misc import postgres_version_to_int
 from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Retry, \
-    RetryFailedError, parse_int, split_host_port, tzutc, uri, cluster_as_json
+    RetryFailedError, parse_int, parse_schedule, split_host_port, tzutc, uri, cluster_as_json
 
 logger = logging.getLogger(__name__)
 
@@ -777,44 +776,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
             self.server.patroni.api_sigterm()
         self.write_response(202, 'shutdown scheduled')
 
-    @staticmethod
-    def parse_schedule(schedule: str,
-                       action: str) -> Tuple[Union[int, None], Union[str, None], Union[datetime.datetime, None]]:
-        """Parse the given *schedule* and validate it.
-
-        :param schedule: a string representing a timestamp, e.g. ``2023-04-14T20:27:00+00:00``.
-        :param action: the action to be scheduled (``restart``, ``switchover``, or ``failover``).
-
-        :returns: a tuple composed of 3 items:
-
-            * Suggested HTTP status code for a response:
-
-                * ``None``: if no issue was faced while parsing, leaving it up to the caller to decide the status; or
-                * ``400``: if no timezone information could be found in *schedule*; or
-                * ``422``: if *schedule* is invalid -- in the past or not parsable.
-
-            * An error message, if any error is faced, otherwise ``None``;
-            * Parsed *schedule*, if able to parse, otherwise ``None``.
-
-        """
-        error = None
-        scheduled_at = None
-        try:
-            scheduled_at = dateutil.parser.parse(schedule)
-            if scheduled_at.tzinfo is None:
-                error = 'Timezone information is mandatory for the scheduled {0}'.format(action)
-                status_code = 400
-            elif scheduled_at < datetime.datetime.now(tzutc):
-                error = 'Cannot schedule {0} in the past'.format(action)
-                status_code = 422
-            else:
-                status_code = None
-        except (ValueError, TypeError):
-            logger.exception('Invalid scheduled %s time: %s', action, schedule)
-            error = 'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601'
-            status_code = 422
-        return status_code, error, scheduled_at
-
     @check_access
     def do_POST_restart(self) -> None:
         """Handle a ``POST`` request to ``/restart`` path.
@@ -870,9 +831,9 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
         for k in request:
             if k == 'schedule':
-                (_, data, request[k]) = self.parse_schedule(request[k], "restart")
+                parse_result, request[k] = parse_schedule(request[k])
-                if _:
+                if parse_result:
-                    status_code = _
+                    data, status_code = parse_result.value[0], parse_result.value[1]
                     break
             elif k == 'role':
                 if request[k] not in ('master', 'primary', 'replica'):
@@ -1022,39 +983,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
                 logger.debug('Exception occurred during polling %s result: %s', action, e)
         return 503, action.title() + ' status unknown'
 
-    def is_failover_possible(self, cluster: Cluster, leader: Optional[str], candidate: Optional[str],
-                             action: str) -> Optional[str]:
-        """Checks whether there are nodes that could take over after demoting the primary.
-
-        :param cluster: the Patroni cluster.
-        :param leader: name of the current Patroni leader.
-        :param candidate: name of the Patroni node to be promoted.
-        :param action: the action to be performed (``switchover`` or ``failover``).
-
-        :returns: a string with the error message or ``None`` if good nodes are found.
-        """
-        is_synchronous_mode = self.server.patroni.config.get_global_config(cluster).is_synchronous_mode
-        if leader and (not cluster.leader or cluster.leader.name != leader):
-            return 'leader name does not match'
-        if candidate:
-            if action == 'switchover' and is_synchronous_mode and not cluster.sync.matches(candidate):
-                return 'candidate name does not match with sync_standby'
-            members = [m for m in cluster.members if m.name == candidate]
-            if not members:
-                return 'candidate does not exists'
-        elif is_synchronous_mode:
-            members = [m for m in cluster.members if cluster.sync.matches(m.name)]
-            if not members:
-                return action + ' is not possible: can not find sync_standby'
-        else:
-            members = [m for m in cluster.members if not cluster.leader or m.name != cluster.leader.name and m.api_url]
-            if not members:
-                return action + ' is not possible: cluster does not have members except leader'
-        for st in self.server.patroni.ha.fetch_nodes_statuses(members):
-            if st.failover_limitation() is None:
-                return None
-        return action + ' is not possible: no good candidates have been found'
-
     @check_access
     def do_POST_failover(self, action: str = 'failover') -> None:
         """Handle a ``POST`` request to ``/failover`` path.
@@ -1083,7 +1011,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
         :param action: the action to be performed (``switchover`` or ``failover``).
         """
         request = self._read_json_content()
-        (status_code, data) = (400, '')
         if not request:
             return
 
@@ -1096,33 +1023,15 @@ class RestApiHandler(BaseHTTPRequestHandler):
         logger.info("received %s request with leader=%s candidate=%s scheduled_at=%s",
                     action, leader, candidate, scheduled_at)
 
-        if action == 'failover' and not candidate:
+        manual_failover = ManualFailover(action, cluster, leader, candidate, scheduled_at,
-            data = 'Failover could be performed only to a specific candidate'
+                                         global_config.is_paused, global_config.is_synchronous_mode,
-        elif action == 'switchover' and not leader:
+                                         self.server.patroni)
-            data = 'Switchover could be performed only from a specific leader'
+        data, status_code = manual_failover.run_precheck().value
 
         if not data and scheduled_at:
-            if action == 'failover':
+            parse_result, scheduled_at = manual_failover.parse_scheduled()
-                data = "Failover can't be scheduled"
+            if parse_result:
-            elif global_config.is_paused:
+                data, status_code = parse_result.value[0], parse_result.value[1]
-                data = "Can't schedule switchover in the paused state"
-            else:
-                (status_code, data, scheduled_at) = self.parse_schedule(scheduled_at, action)
-
-        if not data and global_config.is_paused and not candidate:
-            data = 'Switchover is possible only to a specific candidate in a paused state'
-
-        if action == 'failover' and leader:
-            logger.warning('received failover request with leader specifed - performing switchover instead')
-            action = 'switchover'
-
-        if not data and leader == candidate:
-            data = 'Switchover target and source are the same'
-
-        if not data and not scheduled_at:
-            data = self.is_failover_possible(cluster, leader, candidate, action)
-            if data:
-                status_code = 412
 
         if not data:
             if self.server.patroni.dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at):
@@ -1136,12 +1045,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
             else:
                 data = 'failed to write failover key into DCS'
                 status_code = 503
-        # pyright thinks ``status_code`` can be ``None`` because ``parse_schedule`` call may return ``None``. However,
+
-        # if that's the case, ``status_code`` will be overwritten somewhere between ``parse_schedule`` and
+        status_code = status_code or 400
-        # ``write_response`` calls.
+        self.write_response(status_code, data.format(action=action, leader=leader, candidate=candidate,
-        if TYPE_CHECKING:  # pragma: no cover
+                                                     cluster_name=self.server.patroni.postgresql.scope))
-            assert isinstance(status_code, int)
-        self.write_response(status_code, data)
 
     def do_POST_switchover(self) -> None:
         """Handle a ``POST`` request to ``/switchover`` path.

patroni/ctl.py

@@ -16,8 +16,6 @@ import click
 import codecs
 import copy
 import datetime
-import dateutil.parser
-import dateutil.tz
 import difflib
 import io
 import json
@@ -49,8 +47,9 @@ except ImportError:  # pragma: no cover
 from .config import Config, get_global_config
 from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member
 from .exceptions import PatroniException
+from .manual_failover import ManualFailover
 from .postgresql.misc import postgres_version_to_int
-from .utils import cluster_as_json, patch_config, polling_loop
+from .utils import cluster_as_json, parse_schedule, patch_config, polling_loop
 from .request import PatroniRequest
 from .version import __version__
 
@@ -645,7 +644,8 @@ def get_members(obj: Dict[str, Any], cluster: Cluster, cluster_name: str, member
     if member_names:
         member_names = list(set(member_names) & candidates)
         if not member_names:
-            raise PatroniCtlException('No {0} among provided members'.format(role))
+            raise PatroniCtlException(
+                'No{0} among provided members'.format('t a single cluster member' if role == 'any' else ' ' + role))
     elif action != 'reinitialize':
         member_names = list(candidates)
 
@@ -945,43 +945,6 @@ def check_response(response: urllib3.response.HTTPResponse, member_name: str,
     return True
 
 
-def parse_scheduled(scheduled: Optional[str]) -> Optional[datetime.datetime]:
-    """Parse a string *scheduled* timestamp as a :class:`~datetime.datetime` object.
-
-    :param scheduled: string representation of the timestamp. May also be ``now``.
-
-    :returns: the corresponding :class:`~datetime.datetime` object, if *scheduled* is not ``now``, otherwise ``None``.
-
-    :raises:
-        :class:`PatroniCtlException`: if unable to parse *scheduled* from :class:`str` to :class:`~datetime.datetime`.
-
-    :Example:
-
-        >>> parse_scheduled(None) is None
-        True
-
-        >>> parse_scheduled('now') is None
-        True
-
-        >>> parse_scheduled('2023-05-29T04:32:31')
-        datetime.datetime(2023, 5, 29, 4, 32, 31, tzinfo=tzlocal())
-
-        >>> parse_scheduled('2023-05-29T04:32:31-3')
-        datetime.datetime(2023, 5, 29, 4, 32, 31, tzinfo=tzoffset(None, -10800))
-    """
-    if scheduled is not None and (scheduled or 'now') != 'now':
-        try:
-            scheduled_at = dateutil.parser.parse(scheduled)
-            if scheduled_at.tzinfo is None:
-                scheduled_at = scheduled_at.replace(tzinfo=dateutil.tz.tzlocal())
-        except (ValueError, TypeError):
-            message = 'Unable to parse scheduled timestamp ({0}). It should be in an unambiguous format (e.g. ISO 8601)'
-            raise PatroniCtlException(message.format(scheduled))
-        return scheduled_at
-
-    return None
-
-
 @ctl.command('reload', help='Reload cluster member configuration')
 @click.argument('cluster_name')
 @click.argument('member_names', nargs=-1)
@@ -1061,16 +1024,20 @@ def restart(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member
             * *version* could not be parsed; or
             * a restart is attempted against a cluster that is in maintenance mode.
     """
+    action = 'restart'
     cluster = get_dcs(obj, cluster_name, group).get_cluster()
 
-    members = get_members(obj, cluster, cluster_name, member_names, role, force, 'restart', False, group=group)
+    members = get_members(obj, cluster, cluster_name, member_names, role, force, action, False, group=group)
     if scheduled is None and not force:
-        next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
+        next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M+00')
         scheduled = click.prompt('When should the restart take place (e.g. ' + next_hour + ') ',
                                  type=str, default='now')
+        scheduled = scheduled if scheduled != 'now' else None
 
-    scheduled_at = parse_scheduled(scheduled)
+    parse_result, scheduled_at = parse_schedule(scheduled)
-    confirm_members_action(members, force, 'restart', scheduled_at)
+    if parse_result:
+        raise PatroniCtlException(parse_result.value[0].format(action=action))
+    confirm_members_action(members, force, action, scheduled_at)
 
     if p_any:
         random.shuffle(members)
@@ -1212,6 +1179,9 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
     click.echo('Current cluster topology')
     output_members(obj, cluster, cluster_name, group=group)
 
+    # Define everything missing via interactive input or available cluster info (if force mode)
+
+    # Require Citus group
     if obj.get('citus') and group is None:
         if force:
             raise PatroniCtlException('For Citus clusters the --group must me specified')
@@ -1222,42 +1192,25 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
 
     global_config = get_global_config(cluster)
 
-    # leader has to be be defined for switchover only
+    # Leader is required for switchover only
-    if action == 'switchover':
+    if action == 'switchover' and leader is None:
         if cluster.leader is None or not cluster.leader.name:
             raise PatroniCtlException('This cluster has no leader')
-
+        if force:
-        if leader is None:
+            leader = cluster.leader.name
-            if force:
+        else:
-                leader = cluster.leader.name
+            prompt = 'Standby Leader' if global_config.is_standby_cluster else 'Primary'
-            else:
+            leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
-                prompt = 'Standby Leader' if global_config.is_standby_cluster else 'Primary'
-                leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
-
-        if cluster.leader.name != leader:
-            raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}')
-
-    # excluding members with nofailover tag
-    candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
-    # We sort the names for consistent output to the client
-    candidate_names.sort()
-
-    if not candidate_names:
-        raise PatroniCtlException('No candidates found to {0} to'.format(action))
 
     if candidate is None and not force:
+        # Check if there are any candidates available at all
+        candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
+        if not candidate_names:
+            raise PatroniCtlException('No candidates found to {0} to'.format(action))
+        candidate_names.sort()   # we sort the names for consistent output to the client
         candidate = click.prompt('Candidate ' + str(candidate_names), type=str, default='')
 
-    if action == 'failover' and not candidate:
+    # We allow manual failover to an aync node in the sync mode, so we better ask for the confirmation
-        raise PatroniCtlException('Failover could be performed only to a specific candidate')
-
-    if candidate == leader:
-        raise PatroniCtlException(action.title() + ' target and source are the same.')
-
-    if candidate and candidate not in candidate_names:
-        raise PatroniCtlException(
-            f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover')
-
     if all((not force,
             action == 'failover',
             global_config.is_synchronous_mode,
@@ -1266,21 +1219,45 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
         if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'):
             raise PatroniCtlException('Aborting ' + action)
 
+    if action == 'switchover' and scheduled is None and not force:
+        next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M+00')
+        scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ') ',
+                                 type=str, default='now')
+        scheduled = scheduled if scheduled != 'now' else None
+
+    # Now, when we collected all the possible info, run checks
+    manual_failover = ManualFailover(action, cluster, leader, candidate, scheduled,
+                                     global_config.is_paused, global_config.is_synchronous_mode)
+
+    result_text, _ = manual_failover.run_precheck().value
+    if result_text:
+        raise PatroniCtlException(result_text.format(action=action, leader=leader, candidate=candidate,
+                                                     cluster_name=cluster_name))
+
     scheduled_at_str = None
     scheduled_at = None
-
     if action == 'switchover':
-        if scheduled is None and not force:
+        parse_result, scheduled_at = manual_failover.parse_scheduled()
-            next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
+        if parse_result:
-            scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ' ) ',
+            raise PatroniCtlException(parse_result.value[0].format(action=action))
-                                     type=str, default='now')
-
-        scheduled_at = parse_scheduled(scheduled)
         if scheduled_at:
-            if global_config.is_paused:
-                raise PatroniCtlException("Can't schedule switchover in the paused state")
             scheduled_at_str = scheduled_at.isoformat()
 
+    # By now we have established that the leader exists and the candidate exists,
+    # so confirm the action that is about to be run
+    if not force:
+        demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else ''
+        if scheduled_at_str:
+            # only switchover can be scheduled
+            if not click.confirm(f'Are you sure you want to schedule a switchover in the cluster '
+                                 f'{cluster_name} at {scheduled_at_str}{demote_msg}?'):
+                # action as a var to catch a regression in the tests
+                raise PatroniCtlException('Aborting scheduled ' + action)
+        else:
+            if not click.confirm(f'Are you sure you want to perform a {action} in the cluster {cluster_name}{demote_msg}?'):
+                raise PatroniCtlException('Aborting ' + action)
+
+    # And finally the actual work
     failover_value = {'candidate': candidate}
     if action == 'switchover':
         failover_value['leader'] = leader
@@ -1289,19 +1266,6 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
 
     logging.debug(failover_value)
 
-    # By now we have established that the leader exists and the candidate exists
-    if not force:
-        demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else ''
-        if scheduled_at_str:
-            # only switchover can be scheduled
-            if not click.confirm(f'Are you sure you want to schedule switchover of cluster '
-                                 f'{cluster_name} at {scheduled_at_str}{demote_msg}?'):
-                # action as a var to catch a regression in the tests
-                raise PatroniCtlException('Aborting scheduled ' + action)
-        else:
-            if not click.confirm(f'Are you sure you want to {action} cluster {cluster_name}{demote_msg}?'):
-                raise PatroniCtlException('Aborting ' + action)
-
     r = None
     try:
         member = cluster.leader.member if cluster.leader else candidate and cluster.get_member(candidate, False)

patroni/dcs/__init__.py

@@ -451,7 +451,7 @@ class Leader(NamedTuple):
 
 
 class Failover(NamedTuple):
-    """Immutable object (namedtuple) representing configuration information required for failover/switchover capability.
+    """Immutable object (namedtuple) which represents failover key.
 
     :ivar version: version of the object.
     :ivar leader: name of the leader. If value isn't empty we treat it as a switchover from the specified node.
@@ -547,6 +547,13 @@ class Failover(NamedTuple):
         """
         return int(bool(self.leader)) + int(bool(self.candidate))
 
+    @property
+    def is_switchover(self) -> bool:
+        return bool(self.leader)
+
+    @property
+    def is_failover(self) -> bool:
+        return not self.is_switchover
 
 class ClusterConfig(NamedTuple):
     """Immutable object (namedtuple) which represents cluster configuration.

patroni/ha.py

@@ -234,7 +234,7 @@ class Ha(object):
         """
         if not self.cluster.failover:
             return 'failover'
-        return 'switchover' if self.cluster.failover.leader else 'manual failover'
+        return 'switchover' if self.cluster.failover.is_switchover else 'manual failover'
 
     def load_cluster_from_dcs(self) -> None:
         cluster = self.dcs.get_cluster()
@@ -922,6 +922,27 @@ class Ha(object):
         lag = (self.cluster.last_lsn or 0) - wal_position
         return lag > self.global_config.maximum_lag_on_failover
 
+    def has_members_eligible_to_promote(self, members: List[Member], reference_lsn: int = 0,
+                                        fast_path: bool = False) -> bool:
+        ret = False
+        cluster_timeline = self.cluster.timeline
+
+        for st in self.fetch_nodes_statuses(members):
+            not_allowed_reason = st.failover_limitation()
+            if not_allowed_reason:
+                logger.info('Member %s is %s', st.member.name, not_allowed_reason)
+            elif fast_path:
+                return True
+            elif reference_lsn and st.wal_position < reference_lsn or \
+                    not reference_lsn and self.is_lagging(st.wal_position):
+                logger.info('Member %s exceeds maximum replication lag', st.member.name)
+            elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
+                logger.info('Timeline %s of member %s is behind the cluster timeline %s',
+                            st.timeline, st.member.name, cluster_timeline)
+            else:
+                ret = True
+        return ret
+
     def _is_healthiest_node(self, members: Collection[Member], check_replication_lag: bool = True) -> bool:
         """This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
 
@@ -974,21 +995,7 @@ class Ha(object):
         elif not candidates:
             logger.warning('%s: candidates list is empty', action)
 
-        ret = False
+        return self.has_members_eligible_to_promote(candidates, cluster_lsn)
-        cluster_timeline = self.cluster.timeline
-        for st in self.fetch_nodes_statuses(candidates):
-            not_allowed_reason = st.failover_limitation()
-            if not_allowed_reason:
-                logger.info('Member %s is %s', st.member.name, not_allowed_reason)
-            elif cluster_lsn and st.wal_position < cluster_lsn or \
-                    not cluster_lsn and self.is_lagging(st.wal_position):
-                logger.info('Member %s exceeds maximum replication lag', st.member.name)
-            elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
-                logger.info('Timeline %s of member %s is behind the cluster timeline %s',
-                            st.timeline, st.member.name, cluster_timeline)
-            else:
-                ret = True
-        return ret
 
     def manual_failover_process_no_leader(self) -> Optional[bool]:
         """Handles manual failover/switchover when the old leader already stepped down.
@@ -1037,7 +1044,7 @@ class Ha(object):
             return False
 
         # try to pick some other members for switchover and check that they are healthy
-        if failover.leader:
+        if failover.is_switchover:
             if self.state_handler.name == failover.leader:  # I was the leader
                 # exclude desired member which is unhealthy if it was specified
                 if self.is_failover_possible(exclude_failover_candidate=bool(failover.candidate)):
@@ -1094,7 +1101,7 @@ class Ha(object):
 
         if self.cluster.failover:
             # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race
-            if self.cluster.failover.leader and self.sync_mode_is_active() \
+            if self.cluster.failover.is_switchover and self.sync_mode_is_active() \
                     and not self.cluster.sync.matches(self.state_handler.name, True):
                 return False
             return self.manual_failover_process_no_leader() or False
@@ -2015,7 +2022,7 @@ class Ha(object):
         def is_eligible(node: Member) -> bool:
             # in synchronous mode we allow failover (not switchover!) to async node
             if self.sync_mode_is_active() and not self.cluster.sync.matches(node.name)\
-                    and not (failover and not failover.leader):
+                    and not (failover and failover.is_failover):
                 return False
             # Don't spend time on "nofailover" nodes checking.
             # We also don't need nodes which we can't query with the api in the list.

patroni/manual_failover.py

@@ -0,0 +1,93 @@
+from enum import Enum
+from typing import Optional, Tuple, TYPE_CHECKING
+
+if TYPE_CHECKING:  # pragma: no cover
+    import datetime
+
+    from .dcs import Cluster
+    from .ha import Patroni
+    from .utils import ParseScheduleErrors
+
+from .utils import parse_schedule
+
+
+class ManualFailoverPrecheckStatus(Enum):
+    FAILOVER_NO_CANDIDATE = ('Failover could be performed only to a specific candidate', 400)
+    SWITCHOVER_NO_LEADER = ('Switchover could be performed only from a specific leader', 400)
+    SCHEDULED_FAILOVER = ("Failover can't be scheduled", 400)
+    SCHEDULED_SWITCHOVER_PAUSE = ("Can't schedule switchover in the paused state", 400)
+    SWITCHOVER_PAUSE_NO_CANDIDATE = ('Switchover is possible only to a specific candidate in a paused state', 400)
+    SWITCHOVER_TO_LEADER = ('Switchover target and source are the same', 400)
+
+    CLUSTER_NO_LEADER = ('Cluster {cluster_name} has no leader', 412)
+    LEADER_NOT_MEMBER = ('Member {leader} is not the leader of cluster {cluster_name}', 412)
+    CANDIDATE_NOT_SYNC_STANDBY = ('candidate name does not match with sync_standby', 412)
+    NO_SYNC_CANDIDATE = ('{action} is not possible: can not find sync_standby', 412)
+    ONLY_LEADER = ('{action} is not possible: cluster does not have members except leader', 412)
+    CANDIDATE_NOT_MEMEBER = ('Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover',
+                             412)
+    NO_GOOD_CANDIDATES = ('{action} is not possible: no good candidates have been found', 412)
+
+    CHECK_PASSED = ('', None)
+
+
+class ManualFailover(object):
+
+    def __init__(self, action: str, cluster: 'Cluster',
+                 leader: Optional[str], candidate: Optional[str], scheduled: Optional[str],
+                 paused: bool = False, sync_mode: bool = False, patroni_obj: Optional['Patroni'] = None) -> None:
+        self.action = action
+        self.cluster = cluster
+        self.leader = leader
+        self.candidate = candidate
+        self.scheduled = scheduled
+        self.paused = paused
+        self.sync_mode = sync_mode
+        self.patroni = patroni_obj
+
+    def parse_scheduled(self) -> Tuple[Optional['ParseScheduleErrors'], Optional['datetime.datetime']]:
+        return parse_schedule(self.scheduled)
+
+    def run_precheck(self) -> ManualFailoverPrecheckStatus:
+        if self.action == 'failover' and not self.candidate:
+            return ManualFailoverPrecheckStatus.FAILOVER_NO_CANDIDATE
+        elif self.action == 'switchover' and not self.leader:
+            return ManualFailoverPrecheckStatus.SWITCHOVER_NO_LEADER
+
+        if self.scheduled:
+            if self.action == 'failover':
+                return ManualFailoverPrecheckStatus.SCHEDULED_FAILOVER
+            elif self.paused:
+                return ManualFailoverPrecheckStatus.SCHEDULED_SWITCHOVER_PAUSE
+
+        if self.paused and not self.candidate:
+            return ManualFailoverPrecheckStatus.SWITCHOVER_PAUSE_NO_CANDIDATE
+
+        if self.leader == self.candidate:
+            return ManualFailoverPrecheckStatus.SWITCHOVER_TO_LEADER
+
+        if self.action == 'switchover':
+            if self.cluster.leader is None or not self.cluster.leader.name:
+                return ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER
+            if self.cluster.leader.name != self.leader:
+                return ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER
+
+        if self.candidate:
+            if self.action == 'switchover' and self.sync_mode and not self.cluster.sync.matches(self.candidate):
+                return ManualFailoverPrecheckStatus.CANDIDATE_NOT_SYNC_STANDBY
+            members = [m for m in self.cluster.members if m.name == self.candidate]
+            if not members:
+                return ManualFailoverPrecheckStatus.CANDIDATE_NOT_MEMEBER
+        elif self.sync_mode:
+            members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)]
+            if not members:
+                return ManualFailoverPrecheckStatus.NO_SYNC_CANDIDATE
+        else:
+            members = [m for m in self.cluster.members if not self.cluster.leader or m.name != self.cluster.leader.name and m.api_url]
+            if not members:
+                return ManualFailoverPrecheckStatus.ONLY_LEADER
+
+        if self.patroni and not self.patroni.ha.has_members_eligible_to_promote(members, fast_path=True):
+            return ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES
+
+        return ManualFailoverPrecheckStatus.CHECK_PASSED

patroni/utils.py

@@ -9,6 +9,8 @@
 :var DBL_RE: regular expression to match double precision numbers, signed or unsigned. Matches scientific notation too.
 :var WHITESPACE_RE: regular expression to match whitespace characters
 """
+import datetime
+import dateutil.parser
 import errno
 import logging
 import os
@@ -20,6 +22,7 @@ import subprocess
 import sys
 import tempfile
 import time
+from enum import Enum
 from shlex import split
 
 from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING
@@ -836,8 +839,9 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig']
         ret['pause'] = True
     if cluster.failover and cluster.failover.scheduled_at:
         ret['scheduled_switchover'] = {'at': cluster.failover.scheduled_at.isoformat()}
-        if cluster.failover.leader:
+        if TYPE_CHECKING:  # pragma: no cover
-            ret['scheduled_switchover']['from'] = cluster.failover.leader
+            assert cluster.failover.leader
+        ret['scheduled_switchover']['from'] = cluster.failover.leader
         if cluster.failover.candidate:
             ret['scheduled_switchover']['to'] = cluster.failover.candidate
     return ret
@@ -1061,3 +1065,23 @@ def get_major_version(bin_dir: Optional[str] = None, bin_name: str = 'postgres')
     if TYPE_CHECKING:  # pragma: no cover
         assert version is not None
     return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1)
+
+
+class ParseScheduleErrors(Enum):
+    NO_TIMEZONE = ('Timezone information is mandatory for the scheduled {action}', 400)
+    SCHEDULED_IN_PAST = ('Cannot schedule {action} in the past', 422)
+    PARSING_ERROR = ('Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601', 422)
+
+
+def parse_schedule(schedule: Optional[str]) -> Tuple[Optional[ParseScheduleErrors], Optional[datetime.datetime]]:
+    scheduled_at = None
+    if schedule is not None:
+        try:
+            scheduled_at = dateutil.parser.parse(schedule)
+            if scheduled_at.tzinfo is None:
+                return ParseScheduleErrors.NO_TIMEZONE, scheduled_at
+            elif scheduled_at < datetime.datetime.now(tzutc):
+                return ParseScheduleErrors.SCHEDULED_IN_PAST, scheduled_at
+        except (ValueError, TypeError):
+            return ParseScheduleErrors.PARSING_ERROR, scheduled_at
+    return None, scheduled_at

tests/test_api.py

@@ -13,11 +13,12 @@ from patroni.config import GlobalConfig
 from patroni.dcs import ClusterConfig, Member
 from patroni.exceptions import PostgresConnectionException
 from patroni.ha import _MemberStatus
+from patroni.manual_failover import ManualFailoverPrecheckStatus
 from patroni.psycopg import OperationalError
-from patroni.utils import RetryFailedError, tzutc
+from patroni.utils import ParseScheduleErrors, RetryFailedError, tzutc
 
 from . import MockConnect, psycopg_connect
-from .test_ha import get_cluster_initialized_without_leader
+from .test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader
 
 
 future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5)
@@ -140,6 +141,9 @@ class MockHa(object):
     def is_paused():
         return True
 
+    def has_members_eligible_to_promote(*args, **kwargs):
+        return True
+
 
 class MockLogger(object):
 
@@ -519,7 +523,7 @@ class TestRestApiHandler(unittest.TestCase):
         # Invalid content
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             MockRestApiServer(RestApiHandler, post + '7\n\n{"1":2}')
-            response_mock.assert_called_with(400, 'Switchover could be performed only from a specific leader')
+            response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SWITCHOVER_NO_LEADER.value[::-1])
 
         # Empty content
         request = post + '0\n\n'
@@ -527,25 +531,19 @@ class TestRestApiHandler(unittest.TestCase):
 
         # [Switchover without a candidate]
 
-        # Cluster with only a leader
+        cluster.leader.name = 'postgresql1'
-        with patch.object(RestApiHandler, 'write_response') as response_mock:
+        request = post + '25\n\n{"leader": "postgresql1"}'
-            cluster.leader.name = 'postgresql1'
-            request = post + '25\n\n{"leader": "postgresql1"}'
-            MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(
-                412, 'switchover is not possible: cluster does not have members except leader')
 
-        # Switchover in pause mode
+        # No candidate in pause mode
         with patch.object(RestApiHandler, 'write_response') as response_mock, \
              patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
             MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(
+            response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SWITCHOVER_PAUSE_NO_CANDIDATE.value[::-1])
-                400, 'Switchover is possible only to a specific candidate in a paused state')
 
         # No healthy nodes to promote in both sync and async mode
         for is_synchronous_mode, response in (
-                (True, 'switchover is not possible: can not find sync_standby'),
+                (True, ManualFailoverPrecheckStatus.NO_SYNC_CANDIDATE.value[0].format(action='switchover')),
-                (False, 'switchover is not possible: cluster does not have members except leader')):
+                (False, ManualFailoverPrecheckStatus.ONLY_LEADER.value[0].format(action='switchover'))):
             with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)), \
                  patch.object(RestApiHandler, 'write_response') as response_mock:
                 MockRestApiServer(RestApiHandler, request)
@@ -557,20 +555,25 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             request = post + '53\n\n{"leader": "postgresql2", "candidate": "postgresql2"}'
             MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(400, 'Switchover target and source are the same')
+            response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SWITCHOVER_TO_LEADER.value[::-1])
 
         # Current leader is different from the one specified
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             cluster.leader.name = 'postgresql2'
             request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
             MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(412, 'leader name does not match')
+            response_mock.assert_called_with(
+                ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER.value[1],
+                ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER.value[0].format(leader='postgresql1',
+                                                                               cluster_name='dummy'))
 
-        # Candidate to promote is not a member of the cluster
+        # Candidate to promote is not a sync standby/a member of the cluster
         cluster.leader.name = 'postgresql1'
         cluster.sync.matches.return_value = False
         for is_synchronous_mode, response in (
-                (True, 'candidate name does not match with sync_standby'), (False, 'candidate does not exists')):
+                (True, ManualFailoverPrecheckStatus.CANDIDATE_NOT_SYNC_STANDBY.value[0]),
+                (False, ManualFailoverPrecheckStatus.CANDIDATE_NOT_MEMEBER.value[0].format(candidate="postgresql2",
+                                                                                           cluster_name='dummy'))):
             with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)), \
                  patch.object(RestApiHandler, 'write_response') as response_mock:
                 MockRestApiServer(RestApiHandler, request)
@@ -579,9 +582,21 @@ class TestRestApiHandler(unittest.TestCase):
         cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
                            Member(0, 'postgresql2', 30, {'api_url': 'http'})]
 
+        # Cluster has no leader
+        cluster.leader.name = None
+        with patch.object(RestApiHandler, 'write_response') as response_mock:
+            request = post + '53\n\n{"leader": "postgresql1"}'
+            MockRestApiServer(RestApiHandler, request)
+            response_mock.assert_called_with(
+                ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER.value[1],
+                ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER.value[0].format(leader='leader', cluster_name='dummy'))
+
+        cluster.leader.name = 'postgresql1'
+
         # Failover key is empty in DCS
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             cluster.failover = None
+            request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
             MockRestApiServer(RestApiHandler, request)
             response_mock.assert_called_with(503, 'Switchover failed')
 
@@ -616,10 +631,12 @@ class TestRestApiHandler(unittest.TestCase):
         dcs.manual_failover.return_value = True
 
         # Candidate is not healthy to be promoted
-        with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])), \
+        with patch.object(MockHa, 'has_members_eligible_to_promote', Mock(return_value=False)), \
              patch.object(RestApiHandler, 'write_response') as response_mock:
             MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(412, 'switchover is not possible: no good candidates have been found')
+            response_mock.assert_called_with(
+                ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[1],
+                ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[0].format(action='switchover'))
 
         # [Scheduled switchover]
 
@@ -630,47 +647,58 @@ class TestRestApiHandler(unittest.TestCase):
             MockRestApiServer(RestApiHandler, request)
             response_mock.assert_called_with(202, 'Switchover scheduled')
 
-        # Schedule in paused mode
+        # Scheduled in pause mode
         with patch.object(RestApiHandler, 'write_response') as response_mock, \
              patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
             dcs.manual_failover.return_value = False
             MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(400, "Can't schedule switchover in the paused state")
+            response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SCHEDULED_SWITCHOVER_PAUSE.value[::-1])
 
         # No timezone specified
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' + \
                              ' "scheduled_at": "6016-02-15T18:13:30.568224"}'
             MockRestApiServer(RestApiHandler, request)
-            response_mock.assert_called_with(400, 'Timezone information is mandatory for the scheduled switchover')
+            response_mock.assert_called_with(
+                ParseScheduleErrors.NO_TIMEZONE.value[1],
+                ParseScheduleErrors.NO_TIMEZONE.value[0].format(action='switchover'))
 
         request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2", "scheduled_at": "'
 
         # Scheduled in the past
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             MockRestApiServer(RestApiHandler, request + '1016-02-15T18:13:30.568224+01:00"}')
-            response_mock.assert_called_with(422, 'Cannot schedule switchover in the past')
+            response_mock.assert_called_with(
+                ParseScheduleErrors.SCHEDULED_IN_PAST.value[1],
+                ParseScheduleErrors.SCHEDULED_IN_PAST.value[0].format(action='switchover'))
 
         # Invalid date
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             MockRestApiServer(RestApiHandler, request + '2010-02-29T18:13:30.568224+01:00"}')
-            response_mock.assert_called_with(
+            response_mock.assert_called_with(*ParseScheduleErrors.PARSING_ERROR.value[::-1])
-                422, 'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601')
 
-    def test_do_POST_failover(self):
+    @patch.object(MockPatroni, 'dcs')
+    def test_do_POST_failover(self, mock_dcs):
         post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: '
+        cluster = mock_dcs.get_cluster.return_value
 
         with patch.object(RestApiHandler, 'write_response') as response_mock:
-            MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}')
+            MockRestApiServer(RestApiHandler, post + '19\n\n{"leader":"leader"}')
-            response_mock.assert_called_once_with(400, 'Failover could be performed only to a specific candidate')
+            response_mock.assert_called_once_with(*ManualFailoverPrecheckStatus.FAILOVER_NO_CANDIDATE.value[::-1])
 
         with patch.object(RestApiHandler, 'write_response') as response_mock:
             MockRestApiServer(RestApiHandler, post + '37\n\n{"candidate":"2","scheduled_at": "1"}')
-            response_mock.assert_called_once_with(400, "Failover can't be scheduled")
+            response_mock.assert_called_once_with(*ManualFailoverPrecheckStatus.SCHEDULED_FAILOVER.value[::-1])
 
-        with patch.object(RestApiHandler, 'write_response') as response_mock:
+        # Candidate is not healthy to be promoted
-            MockRestApiServer(RestApiHandler, post + '30\n\n{"leader":"1","candidate":"2"}')
+        cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
-            response_mock.assert_called_once_with(412, 'leader name does not match')
+                           Member(0, 'postgresql2', 30, {'api_url': 'http'})]
+        with patch.object(MockHa, 'has_members_eligible_to_promote', Mock(return_value=False)), \
+             patch.object(RestApiHandler, 'write_response') as response_mock:
+            MockRestApiServer(RestApiHandler, post + '27\n\n{"candidate":"postgresql2"}')
+            response_mock.assert_called_with(
+                ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[1],
+                ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[0].format(action='failover'))
 
     @patch.object(MockHa, 'is_leader', Mock(return_value=True))
     def test_do_POST_citus(self):

tests/test_ctl.py

@@ -6,12 +6,14 @@ import unittest
 from click.testing import CliRunner
 from datetime import datetime, timedelta
 from mock import patch, Mock, PropertyMock
+from patroni.config import GlobalConfig
 from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \
     get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \
     format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Cluster, Failover
+from patroni.manual_failover import ManualFailoverPrecheckStatus
 from patroni.psycopg import OperationalError
-from patroni.utils import tzutc
+from patroni.utils import ParseScheduleErrors, tzutc
 from prettytable import PrettyTable, ALL
 from urllib3 import PoolManager
 
@@ -35,6 +37,10 @@ DEFAULT_CONFIG = {
 class TestCtl(unittest.TestCase):
     TEST_ROLES = ('master', 'primary', 'leader')
 
+    SCHEDULED_TS = '2055-01-01T12:00:00+01:00'
+    SCHEDULED_TS_NO_TZ = '2055-01-01T12:00:00'
+    SCHEDULED_TS_INVALID = '2055-02-30T12:00:00'
+
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     @patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
     def setUp(self):
@@ -116,70 +122,6 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
         self.assertEqual(result.exit_code, 0)
 
-        # Scheduled (confirm)
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
-                                    input='leader\nother\n2300-01-01T12:23:00\ny')
-        self.assertEqual(result.exit_code, 0)
-
-        # Scheduled (abort)
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
-                                    '--scheduled', '2015-01-01T12:00:00+01:00'], input='leader\nother\n\nN')
-        self.assertEqual(result.exit_code, 1)
-
-        # Scheduled with --force option
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
-                                    '--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
-        self.assertEqual(result.exit_code, 0)
-
-        # Scheduled in pause mode
-        with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
-            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
-                                              '--force', '--scheduled', '2015-01-01T12:00:00'])
-            self.assertEqual(result.exit_code, 1)
-            self.assertIn("Can't schedule switchover in the paused state", result.output)
-
-        # Target and source are equal
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
-        self.assertEqual(result.exit_code, 1)
-        self.assertIn('Switchover target and source are the same', result.output)
-
-        # Candidate is not a member of the cluster
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
-        self.assertEqual(result.exit_code, 1)
-        self.assertIn('Member Reality does not exist in cluster dummy or is tagged as nofailover', result.output)
-
-        # Invalid timestamp
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force', '--scheduled', 'invalid'])
-        self.assertEqual(result.exit_code, 1)
-        self.assertIn('Unable to parse scheduled timestamp', result.output)
-
-        # Invalid timestamp
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
-                                          '--force', '--scheduled', '2115-02-30T12:00:00+01:00'])
-        self.assertEqual(result.exit_code, 1)
-        self.assertIn('Unable to parse scheduled timestamp', result.output)
-
-        # Specifying wrong leader
-        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='dummy')
-        self.assertEqual(result.exit_code, 1)
-        self.assertIn('Member dummy is not the leader of cluster dummy', result.output)
-
-        # Errors while sending Patroni REST API request
-        with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
-            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
-                                        input='leader\nother\n2300-01-01T12:23:00\ny')
-            self.assertIn('falling back to DCS', result.output)
-
-        with patch.object(PoolManager, 'request') as mock_api_request:
-            mock_api_request.return_value.status = 500
-            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-            self.assertIn('Switchover failed', result.output)
-
-            mock_api_request.return_value.status = 501
-            mock_api_request.return_value.data = b'Server does not support this operation'
-            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
-            self.assertIn('Switchover failed', result.output)
-
         # No members available
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_only_leader
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
@@ -197,6 +139,132 @@ class TestCtl(unittest.TestCase):
         self.assertEqual(result.exit_code, 1)
         self.assertIn('For Citus clusters the --group must me specified', result.output)
 
+        # [Scheduled]
+
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+
+        # Scheduled (confirm)
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
+                                    input=f'leader\nother\n{self.SCHEDULED_TS}\ny')
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn(f'Are you sure you want to schedule a switchover in the cluster dummy '
+                      f'at {self.SCHEDULED_TS}, demoting current leader', result.output)
+
+        # Scheduled (abort)
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                    '--scheduled', self.SCHEDULED_TS], input='leader\nother\n\nN')
+        self.assertEqual(result.exit_code, 1)
+
+        # Scheduled with --force option
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                    '--force', '--scheduled', self.SCHEDULED_TS])
+        self.assertEqual(result.exit_code, 0)
+
+        # Scheduled in pause mode
+        with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                              '--force', '--scheduled', self.SCHEDULED_TS])
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(ManualFailoverPrecheckStatus.SCHEDULED_SWITCHOVER_PAUSE.value[0], result.output)
+
+        # Invalid timestamp with force
+        result = self.runner.invoke(ctl,['switchover', 'dummy', '--group', '0', '--force', '--scheduled',
+                                         self.SCHEDULED_TS_INVALID])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Unable to parse scheduled timestamp', result.output)
+
+        # Invalid timestamp
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                          '--force', '--scheduled', self.SCHEDULED_TS_INVALID])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Unable to parse scheduled timestamp', result.output)
+
+        # Invalid timestamp - no timezone
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
+                                          '--force', '--scheduled', self.SCHEDULED_TS_NO_TZ])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(ParseScheduleErrors.NO_TIMEZONE.value[0].format(action='switchover'), result.output)
+
+        # [Other erroneous combinations]
+
+        # No candidate in pause mode
+        with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\n\n\ny')
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(ManualFailoverPrecheckStatus.SWITCHOVER_PAUSE_NO_CANDIDATE.value[0], result.output)
+
+        # Target and source are equal
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(ManualFailoverPrecheckStatus.SWITCHOVER_TO_LEADER.value[0], result.output)
+
+        # Candidate is not a member of the cluster
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(ManualFailoverPrecheckStatus.CANDIDATE_NOT_MEMEBER.value[0].format(candidate='Reality',
+                                                                                         cluster_name='dummy'),
+                      result.output)
+
+        # Specifying wrong leader
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='dummy')
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(
+            ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER.value[0].format(leader='dummy',
+                                                                           cluster_name='dummy'),
+            result.output)
+
+        mock_get_dcs.return_value.get_cluster = Mock(
+            return_value=get_cluster_initialized_with_leader(sync=('leader', 'other')))
+
+        # Candidate is not a sync standby
+        with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=True)):
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\notherMember\n\ny')
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(ManualFailoverPrecheckStatus.CANDIDATE_NOT_SYNC_STANDBY.value[0], result.output)
+
+        # No healthy nodes to promote in sync mode
+        mock_get_dcs.return_value.get_cluster = Mock(return_value=get_cluster_initialized_with_leader(sync=('leader')))
+        with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=True)):
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(ManualFailoverPrecheckStatus.NO_SYNC_CANDIDATE.value[0].format(action='switchover'),
+                          result.output)
+
+        # No healthy nodes to promote in async mode
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_only_leader
+        with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=False)):
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn(ManualFailoverPrecheckStatus.ONLY_LEADER.value[0].format(action='switchover'),
+                          result.output)
+
+        # Cluster has no leader
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_without_leader
+        result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--leader', 'leader', '--force'])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(
+            ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER.value[0].format(leader='leader', cluster_name='dummy'),
+            result.output)
+
+        # [Errors while sending Patroni REST API request]
+
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+
+        with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
+                                        input=f'leader\nother\n{self.SCHEDULED_TS}\ny')
+            self.assertIn('falling back to DCS', result.output)
+
+        with patch.object(PoolManager, 'request') as mock_api_request:
+            mock_api_request.return_value.status = 500
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
+            self.assertIn('Switchover failed', result.output)
+
+            mock_api_request.return_value.status = 501
+            mock_api_request.return_value.data = b'Server does not support this operation'
+            result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
+            self.assertIn('Switchover failed', result.output)
+
     @patch('patroni.ctl.get_dcs')
     @patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
     @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse()))
@@ -206,8 +274,9 @@ class TestCtl(unittest.TestCase):
         # No candidate specified
         mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
         result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n')
-        self.assertIn('Failover could be performed only to a specific candidate', result.output)
+        self.assertIn(ManualFailoverPrecheckStatus.FAILOVER_NO_CANDIDATE.value[0], result.output)
 
+        # Failover to an async member in sync mode (confirm)
         cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
 
         # Temp test to check a fallback to switchover if leader is specified
@@ -320,12 +389,9 @@ class TestCtl(unittest.TestCase):
 
     @patch.object(PoolManager, 'request')
     @patch('patroni.ctl.get_dcs')
-    def test_restart_reinit(self, mock_get_dcs, mock_post):
+    def test_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
 
         result = self.runner.invoke(ctl, ['reinit', 'alpha'], input='y')
         assert result.exit_code == 1
@@ -334,67 +400,88 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['reinit', 'alpha', 'other'], input='y\ny')
         assert result.exit_code == 0
 
-        # Aborted restart
+    @patch.object(PoolManager, 'request')
+    @patch('patroni.ctl.get_dcs')
+    def test_restart(self, mock_get_dcs, mock_post):
+        mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
+        mock_post.return_value.status = 200
+
+        # Successful restart
+        result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\ny\n')
+        self.assertEqual(result.exit_code, 0)
+
+        # Aborted
         result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\nN')
-        assert result.exit_code == 1
+        self.assertEqual(result.exit_code, 1)
 
+        # With pending the flag
         result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force'])
-        assert result.exit_code == 0
+        self.assertEqual(result.exit_code, 0)
-
-        # Aborted scheduled restart
-        result = self.runner.invoke(ctl, ['restart', 'alpha', '--scheduled', '2019-10-01T14:30'], input='N')
-        assert result.exit_code == 1
 
         # Not a member
         result = self.runner.invoke(ctl, ['restart', 'alpha', 'dummy', '--any'], input='now\ny')
-        assert result.exit_code == 1
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Not a single cluster member among provided members', result.output)
+
+        # Not a member with the specified role
+        result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--role', 'primary'], input='now\ny')
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('No primary among provided members', result.output)
 
         # Wrong pg version
         result = self.runner.invoke(ctl, ['restart', 'alpha', '--any', '--pg-version', '9.1'], input='now\ny')
-        assert 'Error: Invalid PostgreSQL version format' in result.output
+        self.assertEqual(result.exit_code, 1)
-        assert result.exit_code == 1
+        self.assertIn('Error: Invalid PostgreSQL version format', result.output)
 
+        # Restart with timeout
         result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force', '--timeout', '10min'])
-        assert result.exit_code == 0
+        self.assertEqual(result.exit_code, 0)
 
-        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        # Scheduled restart
-        result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
-        assert 'Failed: flush scheduled restart' in result.output
 
+        # Aborted scheduled restart
+        result = self.runner.invoke(ctl, ['restart', 'alpha', '--scheduled', self.SCHEDULED_TS], input='N')
+        self.assertEqual(result.exit_code, 1)
+
+        # Error parsing scheduled flag value (no tz)
+        result = self.runner.invoke(ctl,
+                                    ['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS_NO_TZ])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn(ParseScheduleErrors.NO_TIMEZONE.value[0].format(action='restart'), result.output)
+
+        # Error parsing scheduled flag value (invalid date)
+        result = self.runner.invoke(ctl,
+                                    ['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS_INVALID])
+        self.assertEqual(result.exit_code, 1)
+        self.assertIn('Unable to parse scheduled timestamp', result.output)
+
+        # Successfully scheduled restart
+        result = self.runner.invoke(ctl, ['restart', 'alpha', '--scheduled', self.SCHEDULED_TS], input='Y')
+        self.assertEqual(result.exit_code, 0)
+        self.assertIn('Success: restart on member other', result.output)
+
+        # Not possible to schedule in pause mode
         with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
             result = self.runner.invoke(ctl,
-                                        ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
+                                        ['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS])
-            assert result.exit_code == 1
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn("Can't schedule restart in the paused state", result.output)
 
-        # force restart with restart already present
+        # Force restart with restart already scheduled
-        result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
+        result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS])
-        assert result.exit_code == 0
+        self.assertEqual(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
 
         # get restart with the non-200 return code
-        # normal restart, the schedule is actually parsed, but not validated in patronictl
+        ctl_args = ['restart', 'alpha', '--pg-version', '99.0', '--scheduled', self.SCHEDULED_TS]
-        mock_post.return_value.status = 204
+        for code, output in [
-        result = self.runner.invoke(ctl, ctl_args, input='y')
+            (204, 'Failed: restart for member other, status code=204'),
-        assert result.exit_code == 0
+            (202, 'Success: restart scheduled'),
-
+            (409, 'Failed: another restart is already')
-        # 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 = code
-        mock_post.return_value.status = 202
+            result = self.runner.invoke(ctl, ctl_args, input='y')
-        result = self.runner.invoke(ctl, ctl_args, input='y')
+            self.assertEqual(result.exit_code, 0)
-        assert 'Success: restart scheduled' in result.output
+            self.assertIn(output, 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):

tests/test_ha.py

@@ -1630,3 +1630,13 @@ class TestHa(PostgresInit):
             self.assertEqual(self.ha.patroni.request.call_args[1]['timeout'], 2)
             mock_logger.assert_called()
             self.assertTrue(mock_logger.call_args[0][0].startswith('Request to Citus coordinator'))
+
+    def test_has_members_eligible_to_promote(self):
+        self.ha.fetch_node_status = get_node_status()
+        members = [
+            Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres'}),
+            Member(0, 'test2', 1, {'api_url': 'http://127.0.0.1:8011/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres'}),
+        ]
+        with patch('patroni.ha.logger.info') as mock_logger:
+            self.assertTrue(self.ha.has_members_eligible_to_promote(members, fast_path=True))
+            mock_logger.assert_not_called()