Repository: zalando/patroni
Pull request #2788 (2023-07-28 20:41:20)

Fix `replicatefrom` tag in `postgres2.yml`
The node `postgresql2` in the repository is apparently supposed to be a cascading standby.

However, if that is the case, there is a typo in the name of its upstream node.

This commit fixes that issue.

coveralls: ## Pull Request Test Coverage Report for [Build 5696136101](https://coveralls.io/builds/61651116)

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage remained the same at **99.859%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/61651116/badge)](https://coveralls.io/builds/61651116) |
| :-- | --: |
| Change from base [Build 5680056308](https://coveralls.io/builds/61614054): |  0.0% |
| Covered Lines: | 12741 |
| Relevant Lines: | 12759 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/features/environment.py b/features/environment.py
index ec503bb5..e3c21252 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -59,7 +59,7 @@ class AbstractController(abc.ABC):
                 break
             time.sleep(1)
         else:
-            assert False,\
+            assert False, \
                 "{0} instance is not available for queries after {1} seconds".format(self._name, max_wait_limit)
 
     def stop(self, kill=False, timeout=15, _=False):
diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py
index 6499a03f..6718cb01 100644
--- a/features/steps/basic_replication.py
+++ b/features/steps/basic_replication.py
@@ -21,7 +21,7 @@ def start_duplicate_patroni(context, name, port):
         context.pctl.start('dup-' + name, custom_config=config)
         assert False, "Process was expected to fail"
     except AssertionError as e:
-        assert 'is not running after being started' in str(e),\
+        assert 'is not running after being started' in str(e), \
             "No error was raised by duplicate start of {0} ".format(name)
 
 
@@ -88,14 +88,14 @@ def table_is_present_on(context, table_name, pg_name, max_replication_delay):
             break
         sleep(1)
     else:
-        assert False,\
+        assert False, \
             "Table {0} is not present on {1} after {2} seconds".format(table_name, pg_name, max_replication_delay)
 
 
 @then('{pg_name:w} role is the {pg_role:w} after {max_promotion_timeout:d} seconds')
 def check_role(context, pg_name, pg_role, max_promotion_timeout):
     max_promotion_timeout *= context.timeout_multiplier
-    assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)),\
+    assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)), \
         "{0} role didn't change to {1} after {2} seconds".format(pg_name, pg_role, max_promotion_timeout)
 
 
@@ -111,5 +111,5 @@ def replication_works(context, primary, replica, time_limit):
 @then('there is a "{message}" {level:w} in the {node} patroni log')
 def check_patroni_log(context, message, level, node):
     messsages_of_level = context.pctl.read_patroni_log(node, level)
-    assert any(message in line for line in messsages_of_level),\
+    assert any(message in line for line in messsages_of_level), \
         "There was no {0} {1} in the {2} patroni log".format(message, level, node)
diff --git a/features/steps/citus.py b/features/steps/citus.py
index 1af70a30..644219c7 100644
--- a/features/steps/citus.py
+++ b/features/steps/citus.py
@@ -125,5 +125,5 @@ def check_transaction(context, name):
 
 @step("a transaction finishes in {timeout:d} seconds")
 def check_transaction_timeout(context, timeout):
-    assert (datetime.now(tzutc) - context.xact_start).seconds > timeout,\
+    assert (datetime.now(tzutc) - context.xact_start).seconds > timeout, \
         "a transaction finished earlier than in {0} seconds".format(timeout)
diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py
index a745b192..2c76d32d 100644
--- a/features/steps/patroni_api.py
+++ b/features/steps/patroni_api.py
@@ -98,7 +98,7 @@ def do_run(context, cmd):
 @then('I receive a response {component:w} {data}')
 def check_response(context, component, data):
     if component == 'code':
-        assert context.status_code == int(data),\
+        assert context.status_code == int(data), \
             "status code {0} != {1}, response: {2}".format(context.status_code, data, context.response)
     elif component == 'returncode':
         assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code,
@@ -158,7 +158,7 @@ def check_http_response(context, url, value, timeout, negate=False):
             break
         time.sleep(1)
     else:
-        assert False,\
+        assert False, \
             "Value {0} is {1} present in response after {2} seconds".format(value, "not" if not negate else "", timeout)
 
 
diff --git a/patroni/config.py b/patroni/config.py
index 66ffd891..facf16fd 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -329,9 +329,19 @@ class Config(object):
 
     @staticmethod
     def _process_postgresql_parameters(parameters: Dict[str, Any], is_local: bool = False) -> Dict[str, Any]:
-        return {name: value for name, value in (parameters or {}).items()
-                if name not in ConfigHandler.CMDLINE_OPTIONS
-                or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)}
+        pg_params: Dict[str, Any] = {}
+
+        for name, value in (parameters or {}).items():
+            if name not in ConfigHandler.CMDLINE_OPTIONS:
+                pg_params[name] = value
+            elif not is_local:
+                if ConfigHandler.CMDLINE_OPTIONS[name][1](value):
+                    pg_params[name] = value
+                else:
+                    logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
+                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
+
+        return pg_params
 
     def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]:
         config = deepcopy(self.__DEFAULT_CONFIG)
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 73755a41..7b310d15 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -15,7 +15,7 @@ from urllib3.exceptions import HTTPError
 from urllib.parse import urlencode, urlparse, quote
 from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
     TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
 from ..exceptions import DCSError
 from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index 94e546ac..447c7dab 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -21,7 +21,7 @@ from urllib.parse import urlparse
 from urllib3 import Timeout
 from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
     TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
 from ..exceptions import DCSError
 from ..request import get as requests_get
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index 32d00359..e28da844 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -15,7 +15,7 @@ from urllib3.exceptions import ReadTimeoutError, ProtocolError
 from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
 
-from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\
+from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, \
     TimelineHistory, catch_return_false_exception, citus_group_re
 from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry
 from ..exceptions import DCSError, PatroniException
@@ -630,6 +630,16 @@ class PatroniEtcd3Client(Etcd3Client):
 
         return ret
 
+    def txn(self, compare: Dict[str, Any], success: Dict[str, Any],
+            failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
+        ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry)
+        # Here we abuse the fact that the `failure` is only set in the call from update_leader().
+        # In all other cases the txn() call failure may be an indicator of a stale cache,
+        # and therefore we want to restart watcher.
+        if not failure and not ret:
+            self._restart_watcher()
+        return ret
+
 
 class Etcd3(AbstractEtcd):
 
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index bda4bdc8..96097338 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -19,10 +19,10 @@ from urllib3.exceptions import HTTPError
 from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
     TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
 from ..exceptions import DCSError
-from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\
+from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \
     Retry, RetryFailedError, tzutc, uri, USER_AGENT
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
diff --git a/patroni/ha.py b/patroni/ha.py
index 95ca6e05..2445aa02 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -996,9 +996,22 @@ class Ha(object):
                 return ret
 
         if self.state_handler.is_leader():
-            # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
-            return not self.is_paused() or not self.cluster.initialize\
-                or self.state_handler.sysid == self.cluster.initialize
+            if self.is_paused():
+                # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
+                return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize
+
+            # We want to protect from the following scenario:
+            # 1. node1 is stressed so much that heart-beat isn't running regularly and the leader lock expires.
+            # 2. node2 promotes, gets heavy load and the situation described in 1 repeats.
+            # 3. Patroni on node1 comes back, notices that Postgres is running as primary but there is
+            #    no leader key and "happily" acquires the leader lock.
+            # That is, node1 discarded promotion of node2. To avoid it we want to detect timeline change.
+            my_timeline = self.state_handler.get_primary_timeline()
+            if my_timeline < self.cluster.timeline:
+                logger.warning('My timeline %s is behind last known cluster timeline %s',
+                               my_timeline, self.cluster.timeline)
+                return False
+            return True
 
         if self.is_paused():
             return False
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 9461714f..3a61a31d 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -15,7 +15,7 @@ from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
-from ..validator import IntValidator
+from ..validator import IntValidator, EnumValidator
 
 if TYPE_CHECKING:  # pragma: no cover
     from . import Postgresql
@@ -258,14 +258,14 @@ def _false_validator(value: Any) -> bool:
     return False
 
 
-def _wal_level_validator(value: Any) -> bool:
-    return str(value).lower() in ('hot_standby', 'replica', 'logical')
-
-
 def _bool_validator(value: Any) -> bool:
     return parse_bool(value) is not None
 
 
+def _bool_is_true_validator(value: Any) -> bool:
+    return parse_bool(value) is True
+
+
 class ConfigHandler(object):
 
     # List of parameters which must be always passed to postmaster as command line options
@@ -286,8 +286,8 @@ class ConfigHandler(object):
         'listen_addresses': (None, _false_validator, 90100),
         'port': (None, _false_validator, 90100),
         'cluster_name': (None, _false_validator, 90500),
-        'wal_level': ('hot_standby', _wal_level_validator, 90100),
-        'hot_standby': ('on', _false_validator, 90100),
+        'wal_level': ('hot_standby', EnumValidator(('hot_standby', 'replica', 'logical')), 90100),
+        'hot_standby': ('on', _bool_is_true_validator, 90100),
         'max_connections': (100, IntValidator(min=25), 90100),
         'max_wal_senders': (10, IntValidator(min=3), 90100),
         'wal_keep_segments': (8, IntValidator(min=1), 90100),
@@ -297,7 +297,7 @@ class ConfigHandler(object):
         'track_commit_timestamp': ('off', _bool_validator, 90500),
         'max_replication_slots': (10, IntValidator(min=4), 90400),
         'max_worker_processes': (8, IntValidator(min=2), 90400),
-        'wal_log_hints': ('on', _false_validator, 90400)
+        'wal_log_hints': ('on', _bool_is_true_validator, 90400)
     })
 
     _RECOVERY_PARAMETERS = CaseInsensitiveSet(recovery_parameters.keys())
diff --git a/patroni/validator.py b/patroni/validator.py
index b68f9654..a8ada1f5 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -785,7 +785,7 @@ class IntValidator(object):
         self.base_unit = base_unit
         self.raise_assert = raise_assert
 
-    def __call__(self, value: Union[int, str]) -> bool:
+    def __call__(self, value: Any) -> bool:
         """Check if *value* is a valid integer and within the expected range.
 
         .. note::
@@ -821,7 +821,7 @@ class EnumValidator(object):
         self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values)
         self.raise_assert = raise_assert
 
-    def __call__(self, value: str) -> bool:
+    def __call__(self, value: Any) -> bool:
         """Check if provided *value* could be found within *allowed_values*.
 
         .. note::
@@ -829,7 +829,7 @@ class EnumValidator(object):
         :param value: value to be checked.
         :returns: ``True`` if *value* could be found within *allowed_values*.
         """
-        ret = value in self.allowed_values
+        ret = isinstance(value, str) and value in self.allowed_values
 
         if self.raise_assert:
             assert_(ret)
diff --git a/postgres2.yml b/postgres2.yml
index c77e734a..581fa719 100644
--- a/postgres2.yml
+++ b/postgres2.yml
@@ -121,4 +121,4 @@ tags:
     nofailover: false
     noloadbalance: false
     clonefrom: false
-    replicatefrom: postgres1
+#    replicatefrom: postgresql1


Test patch:
diff --git a/tests/test_api.py b/tests/test_api.py
index f9d0b1c1..eb19ebca 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -226,7 +226,7 @@ class TestRestApiHandler(unittest.TestCase):
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary'))
         with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])):
             self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
-        with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\
+        with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)), \
                 patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /standby_leader')
 
@@ -559,7 +559,7 @@ class TestRestApiHandler(unittest.TestCase):
         request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' +\
                          ' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
         MockRestApiServer(RestApiHandler, request)
-        with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)),\
+        with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)), \
                 patch.object(MockPatroni, 'dcs') as d:
             d.manual_failover.return_value = False
             MockRestApiServer(RestApiHandler, request)
diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py
index 4c7c0fcc..9f98fecb 100644
--- a/tests/test_bootstrap.py
+++ b/tests/test_bootstrap.py
@@ -155,9 +155,9 @@ class TestBootstrap(BaseTestPostgresql):
 
         config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}}
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=False)),\
-                patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)),\
-                patch('multiprocessing.Process', Mock(side_effect=Exception)),\
+        with patch.object(Postgresql, 'is_running', Mock(return_value=False)), \
+                patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)), \
+                patch('multiprocessing.Process', Mock(side_effect=Exception)), \
                 patch('multiprocessing.get_context', Mock(side_effect=Exception), create=True):
             self.assertRaises(Exception, self.b.bootstrap, config)
         with open(os.path.join(self.p.data_dir, 'pg_hba.conf')) as f:
@@ -185,12 +185,12 @@ class TestBootstrap(BaseTestPostgresql):
         self.assertFalse(self.b.bootstrap(config))
 
         mock_cancellable_subprocess_call.return_value = 0
-        with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))),\
-                patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True),\
-                patch('os.path.isfile', Mock(return_value=True)),\
-                patch('os.unlink', Mock()),\
-                patch.object(ConfigHandler, 'save_configuration_files', Mock()),\
-                patch.object(ConfigHandler, 'restore_configuration_files', Mock()),\
+        with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))), \
+                patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True), \
+                patch('os.path.isfile', Mock(return_value=True)), \
+                patch('os.unlink', Mock()), \
+                patch.object(ConfigHandler, 'save_configuration_files', Mock()), \
+                patch.object(ConfigHandler, 'restore_configuration_files', Mock()), \
                 patch.object(ConfigHandler, 'write_recovery_conf', Mock()):
             with self.assertRaises(Exception) as e:
                 self.b.bootstrap(config)
diff --git a/tests/test_citus.py b/tests/test_citus.py
index 40d4df8a..7c2d63bb 100644
--- a/tests/test_citus.py
+++ b/tests/test_citus.py
@@ -52,7 +52,7 @@ class TestCitus(BaseTestPostgresql):
                                                'leader': 'leader', 'timeout': 30, 'cooldown': 10})
 
     def test_add_task(self):
-        with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
+        with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
                 patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)):
             self.c.add_task('', 1, None)
             mock_logger.assert_called_once()
@@ -107,7 +107,7 @@ class TestCitus(BaseTestPostgresql):
         self.c.process_tasks()
 
         self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres')
-        with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
+        with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
                 patch.object(CitusHandler, 'query', Mock(side_effect=Exception)):
             self.c.process_tasks()
             mock_logger.assert_called_once()
diff --git a/tests/test_config.py b/tests/test_config.py
index 57a717f5..3f7e4049 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -21,7 +21,8 @@ class TestConfig(unittest.TestCase):
         with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
             self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'}))
         self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': {
-            'parameters': {'cluster_name': 1, 'wal_keep_size': 1, 'track_commit_timestamp': 1, 'wal_level': 1}}}))
+            'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1,
+                           'track_commit_timestamp': 1, 'wal_level': 1}}}))
 
     def test_reload_local_configuration(self):
         os.environ.update({
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index b5add201..f9e07144 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -172,12 +172,12 @@ class TestClient(unittest.TestCase):
         self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
         self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
 
-        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
+        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \
                 patch.object(EtcdClient, '_load_machines_cache', Mock(side_effect=Exception)):
             self.client.http.request = Mock(side_effect=socket.error)
             self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
 
-        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
+        with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \
                 patch.object(EtcdClient, '_load_machines_cache', Mock(return_value=True)):
             self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
 
diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py
index a737f199..2e3ed59d 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -5,8 +5,9 @@ import urllib3
 
 from mock import Mock, PropertyMock, patch
 from patroni.dcs.etcd import DnsCachingResolver
-from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\
-    InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
+from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \
+    Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \
+    Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode
 from threading import Thread
 
 from . import SleepException, MockResponse
@@ -126,10 +127,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3):
         request = {'key': base64_encode('/patroni/test/leader')}
         mock_urlopen.return_value = MockResponse()
         mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}'
-        self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]})
         self.client.call_rpc('/kv/put', request)
         self.client.call_rpc('/kv/deleterange', request)
 
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test_txn(self, mock_urlopen):
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.content = '{"header":{"revision":"1"}}'
+        self.client.txn({'target': 'MOD', 'mod_revision': '1'},
+                        {'request_delete_range': {'key': base64_encode('/patroni/test/leader')}})
+
     @patch('time.time', Mock(side_effect=[1, 10.9, 100]))
     def test__wait_cache(self):
         with self.kv_cache.condition:
@@ -241,7 +248,7 @@ class TestEtcd3(BaseTestEtcd3):
             self.etcd3.update_leader(leader, '123', failsafe={'foo': 'bar'})
         self.etcd3._last_lease_refresh = 0
         self.etcd3.update_leader(leader, '124')
-        with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)),\
+        with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)), \
                 patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
             self.assertRaises(Etcd3Error, self.etcd3.update_leader, leader, '126')
         self.etcd3._lease = leader.session
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 7cdb00e1..ae6be3e2 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -307,7 +307,7 @@ class TestHa(PostgresInit):
         self.p.is_running = false
         self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
         self.assertEqual(self.ha.run_cycle(), 'doing crash recovery in a single user mode')
-        with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)),\
+        with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)), \
                 patch.object(Ha, 'check_timeline', Mock(return_value=False)):
             self.ha._async_executor.schedule('doing crash recovery in a single user mode')
             self.ha.state_handler.cancellable._process = Mock()
@@ -340,7 +340,7 @@ class TestHa(PostgresInit):
         self.ha._rewind.check_leader_is_not_in_recovery = true
         with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)):
             self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
-        with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)),\
+        with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)), \
                 patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
             self.p.follow = true
             self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
@@ -375,6 +375,12 @@ class TestHa(PostgresInit):
     def test_acquire_lock_as_primary(self):
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
 
+    def test_leader_race_stale_primary(self):
+        with patch.object(Postgresql, 'get_primary_timeline', Mock(return_value=1)), \
+                patch('patroni.ha.logger.warning') as mock_logger:
+            self.assertEqual(self.ha.run_cycle(), 'demoting self because i am not the healthiest node')
+            self.assertEqual(mock_logger.call_args[0][0], 'My timeline %s is behind last known cluster timeline %s')
+
     def test_promoted_by_acquiring_lock(self):
         self.ha.is_healthiest_node = true
         self.p.is_leader = false
@@ -608,7 +614,7 @@ class TestHa(PostgresInit):
         self.e.initialize = true
         self.ha.bootstrap()
         self.p.is_leader = true
-        with patch.object(Watchdog, 'activate', Mock(return_value=False)),\
+        with patch.object(Watchdog, 'activate', Mock(return_value=False)), \
                 patch('patroni.ha.logger.error') as mock_logger:
             self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
             self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)
@@ -669,9 +675,9 @@ class TestHa(PostgresInit):
 
             self.ha.update_lock = false
             self.p.set_role('primary')
-            with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)),\
+            with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)), \
                     patch('patroni.async_executor.CriticalTask.result',
-                          PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True),\
+                          PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True), \
                     patch('patroni.postgresql.Postgresql.terminate_starting_postmaster') as mock_terminate:
                 self.assertEqual(self.ha.run_cycle(), 'lost leader lock during restart')
                 mock_terminate.assert_called()
diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py
index f79539a7..662e22a8 100644
--- a/tests/test_kubernetes.py
+++ b/tests/test_kubernetes.py
@@ -8,8 +8,8 @@ import unittest
 import urllib3
 
 from mock import Mock, PropertyMock, mock_open, patch
-from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
-    K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
+from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \
+    K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \
     Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
 from threading import Thread
 from . import MockResponse, SleepException
@@ -86,8 +86,8 @@ class TestK8sConfig(unittest.TestCase):
             with patch('os.environ', env):
                 self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config)
 
-        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
-                patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])),\
+        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \
+                patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])), \
                 patch('builtins.open', Mock(side_effect=[
                     mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')(),
                     mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')()])):
@@ -98,8 +98,8 @@ class TestK8sConfig(unittest.TestCase):
             self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
 
     def test_refresh_token(self):
-        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
-                patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])),\
+        with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \
+                patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])), \
                 patch('builtins.open', Mock(side_effect=[
                     mock_open(read_data='cert')(), mock_open(read_data='a')(),
                     mock_open()(), mock_open(read_data='b')(), mock_open(read_data='c')()])):
@@ -138,10 +138,10 @@ class TestK8sConfig(unittest.TestCase):
 
         config["users"][0]["user"]["client-key-data"] = base64.b64encode(b'foobar').decode('utf-8')
         config["clusters"][0]["cluster"]["certificate-authority-data"] = base64.b64encode(b'foobar').decode('utf-8')
-        with patch('builtins.open', mock_open(read_data=json.dumps(config))),\
-                patch('os.write', Mock()), patch('os.close', Mock()),\
-                patch('os.remove') as mock_remove,\
-                patch('atexit.register') as mock_atexit,\
+        with patch('builtins.open', mock_open(read_data=json.dumps(config))), \
+                patch('os.write', Mock()), patch('os.close', Mock()), \
+                patch('os.remove') as mock_remove, \
+                patch('atexit.register') as mock_atexit, \
                 patch('tempfile.mkstemp') as mock_mkstemp:
             mock_mkstemp.side_effect = [(3, '1.tmp'), (4, '2.tmp')]
             k8s_config.load_kube_config()
diff --git a/tests/test_log.py b/tests/test_log.py
index 1a383908..ebdb4945 100644
--- a/tests/test_log.py
+++ b/tests/test_log.py
@@ -43,7 +43,7 @@ class TestPatroniLogger(unittest.TestCase):
         _LOG.exception('test')
         logger.start()
 
-        with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)),\
+        with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)), \
                 patch('_pytest.logging.LogCaptureHandler.emit', Mock()):
             logging.error('test')
 
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index c759cde0..036e225b 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -333,7 +333,7 @@ class TestPostgresql(BaseTestPostgresql):
 
         mock_read_auto = mock_open(read_data=read_data)
         mock_read_auto.return_value.__iter__ = lambda o: iter(o.readline, '')
-        with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])),\
+        with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])), \
                 patch('os.chmod', Mock()):
             self.p.config.write_postgresql_conf()
 
@@ -496,8 +496,8 @@ class TestPostgresql(BaseTestPostgresql):
             self.p.remove_data_directory()
         with patch('os.path.isfile', Mock(return_value=True)):
             self.p.remove_data_directory()
-        with patch('os.path.islink', Mock(side_effect=[False, False, True, True])),\
-                patch('os.listdir', Mock(return_value=['12345'])),\
+        with patch('os.path.islink', Mock(side_effect=[False, False, True, True])), \
+                patch('os.listdir', Mock(return_value=['12345'])), \
                 patch('os.path.realpath', Mock(side_effect=['../foo', '../foo_tsp'])):
             self.p.remove_data_directory()
 
diff --git a/tests/test_raft.py b/tests/test_raft.py
index ab68c2e6..1fe8733c 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -4,7 +4,7 @@ import tempfile
 import time
 
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\
+from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \
     Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
 from pysyncobj import SyncObjConf, FAIL_REASON
 
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index a4fafcff..22181266 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -65,14 +65,14 @@ class TestRewind(BaseTestPostgresql):
 
     def test_pg_rewind(self):
         r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''}
-        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)),\
+        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)), \
                 patch.object(CancellableSubprocess, 'call', Mock(return_value=None)):
             with patch('subprocess.check_output', Mock(return_value=b'boo')):
                 self.assertFalse(self.r.pg_rewind(r))
             with patch('subprocess.check_output', Mock(side_effect=Exception)):
                 self.assertFalse(self.r.pg_rewind(r))
 
-        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)),\
+        with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)), \
                 patch('subprocess.check_output', Mock(return_value=b'foo %f %p %r %% % %')):
             with patch.object(CancellableSubprocess, 'call', mock_cancellable_call):
                 self.assertFalse(self.r.pg_rewind(r))
@@ -91,7 +91,7 @@ class TestRewind(BaseTestPostgresql):
                                              'Latest checkpoint location': '0/'})):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\
+        with patch.object(Postgresql, 'is_running', Mock(return_value=True)), \
                 patch.object(MockCursor, 'fetchone',
                              Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
diff --git a/tests/test_slots.py b/tests/test_slots.py
index 3f21998f..6d3f17d3 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -43,12 +43,12 @@ class TestSlotsHandler(BaseTestPostgresql):
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
             self.s.sync_replication_slots(cluster, False)
         self.p.set_role('standby_leader')
-        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))),\
+        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
                 patch('patroni.postgresql.slots.logger.debug') as mock_debug:
             self.s.sync_replication_slots(cluster, False)
             mock_debug.assert_called_once()
         self.p.set_role('replica')
-        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)),\
+        with patch.object(Postgresql, 'is_leader', Mock(return_value=False)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
             self.s.sync_replication_slots(cluster, False, paused=True)
             mock_drop.assert_not_called()
@@ -96,8 +96,8 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
-                patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))),\
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
+                patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
@@ -119,10 +119,10 @@ class TestSlotsHandler(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_leader', Mock(return_value=False))
     def test_check_logical_slots_readiness(self):
         self.s.copy_logical_slots(self.cluster, ['ls'])
-        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
                 patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
             self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
-        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
+        with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
                 patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
             self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
@@ -144,7 +144,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.assertRaises(OSError, fsync_dir, 'foo')
 
     def test_slots_advance_thread(self):
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
             self.s.schedule_advance_slots({'foo': {'bar': 100}})
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index baf9ad10..c72fefe9 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -7,7 +7,7 @@ from kazoo.handlers.threading import SequentialThreadingHandler
 from kazoo.protocol.states import KeeperState, ZnodeStat
 from kazoo.retry import RetryFailedError
 from mock import Mock, PropertyMock, patch
-from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\
+from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient, \
     PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError