Merge branch 'feature/dynamic-configuration' of github.com:zalando/patroni into feature/environment-configuration

docs/ENVIRONMENT.rst

@@ -28,6 +28,10 @@ Etcd
 ----
 -  **PATRONI\_ETCD\_HOST**: the host:port for the etcd endpoint.
 
+Exhibitor
+---------
+-  **PATRONI\_EXHIBITOR\_HOSTS**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
+
 PostgreSQL
 ----------
 -  **PATRONI\_POSTGRESQL\_LISTEN**: IP address + port that Postgres listens to. Multiple comma-separated addresses are permitted, as long as the port component is appended after to the last one with a colon, i.e. ``listen: 127.0.0.1,127.0.0.2:5432``. Patroni will use the first address from this list to establish local connections to the PostgreSQL node.

docs/SETTINGS.rst

@@ -37,10 +37,16 @@ Consul
 ------
 -  **host**: the host:port for the Consul endpoint.
 
-etcd
+Etcd
 ----
 -  **host**: the host:port for the etcd endpoint.
 
+Exhibitor
+---------
+-  **hosts**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
+-  **poll\_interval**: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
+-  **port**: Exhibitor port.
+
 PostgreSQL
 ----------
 -  **authentication**:
@@ -80,7 +86,3 @@ REST API
 ZooKeeper
 ----------
 -  **hosts**: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...'].
--  **exhibitor**:  If you are running a ZooKeeper cluster under the Exhibitor supervisory, this section might interest you:
-        -  **hosts**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes.
-        -  **poll\_interval**: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor
-        -  **port**: Exhibitor port.

features/environment.py

@@ -143,14 +143,11 @@ class PatroniController(AbstractController):
             dcs_config = config.pop('etcd')
             dcs_config.pop('host')
 
-            if dcs == 'consul':
-                config[dcs] = dcs_config
-            else:
-                if dcs == 'exhibitor':
-                    dcs_config['exhibitor'] = {'hosts': ['127.0.0.1'], 'port': 8181}
-                else:
-                    dcs_config['hosts'] = ['127.0.0.1:2181']
-                config['zookeeper'] = dcs_config
+            if dcs == 'exhibitor':
+                dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181})
+            elif dcs == 'zookeeper':
+                dcs_config['hosts'] = ['127.0.0.1:2181']
+            config[dcs] = dcs_config
 
         with open(patroni_config_path, 'w') as f:
             yaml.safe_dump(config, f, default_flow_style=False)

patroni/ctl.py

@@ -26,7 +26,7 @@ CONFIG_DIR_PATH = click.get_app_dir('patroni')
 CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml')
 LOGLEVEL = 'WARNING'
 DCS_DEFAULTS = {'zookeeper': {'port': 2181, 'template': "zookeeper:\n hosts: ['{host}:{port}']"},
-                'exhibitor': {'port': 8181, 'template': "zookeeper:\n exhibitor:\n  hosts: [{host}]\n  port: {port}"},
+                'exhibitor': {'port': 8181, 'template': "exhibitor:\n hosts: [{host}]\n port: {port}"},
                 'consul': {'port': 8500, 'template': "consul:\n host: '{host}:{port}'"},
                 'etcd': {'port': 4001, 'template': "etcd:\n host: '{host}:{port}'"}}
 

patroni/dcs/exhibitor.py

@@ -0,0 +1,75 @@
+import logging
+import random
+import requests
+import time
+
+from patroni.dcs.zookeeper import ZooKeeper
+from patroni.utils import sleep
+from requests.exceptions import RequestException
+
+logger = logging.getLogger(__name__)
+
+
+class ExhibitorEnsembleProvider(object):
+
+    TIMEOUT = 3.1
+
+    def __init__(self, hosts, port, uri_path='/exhibitor/v1/cluster/list', poll_interval=300):
+        self._exhibitor_port = port
+        self._uri_path = uri_path
+        self._poll_interval = poll_interval
+        self._exhibitors = hosts
+        self._master_exhibitors = hosts
+        self._zookeeper_hosts = ''
+        self._next_poll = None
+        while not self.poll():
+            logger.info('waiting on exhibitor')
+            sleep(5)
+
+    def poll(self):
+        if self._next_poll and self._next_poll > time.time():
+            return False
+
+        json = self._query_exhibitors(self._exhibitors)
+        if not json:
+            json = self._query_exhibitors(self._master_exhibitors)
+
+        if isinstance(json, dict) and 'servers' in json and 'port' in json:
+            self._next_poll = time.time() + self._poll_interval
+            zookeeper_hosts = ','.join([h + ':' + str(json['port']) for h in sorted(json['servers'])])
+            if self._zookeeper_hosts != zookeeper_hosts:
+                logger.info('ZooKeeper connection string has changed: %s => %s', self._zookeeper_hosts, zookeeper_hosts)
+                self._zookeeper_hosts = zookeeper_hosts
+                self._exhibitors = json['servers']
+                return True
+        return False
+
+    def _query_exhibitors(self, exhibitors):
+        random.shuffle(exhibitors)
+        for host in exhibitors:
+            uri = 'http://{0}:{1}{2}'.format(host, self._exhibitor_port, self._uri_path)
+            try:
+                response = requests.get(uri, timeout=self.TIMEOUT)
+                return response.json()
+            except RequestException:
+                pass
+        return None
+
+    @property
+    def zookeeper_hosts(self):
+        return self._zookeeper_hosts
+
+
+class Exhibitor(ZooKeeper):
+
+    def __init__(self, config):
+        interval = config.get('poll_interval', 300)
+        self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval)
+        config = config.copy()
+        config['hosts'] = self._ensemble_provider.zookeeper_hosts
+        super(Exhibitor, self).__init__(config)
+
+    def _load_cluster(self):
+        if self._ensemble_provider.poll():
+            self._client.set_hosts(self._ensemble_provider.zookeeper_hosts)
+        return super(Exhibitor, self)._load_cluster()

patroni/dcs/zookeeper.py

@@ -1,14 +1,9 @@
 import logging
-import random
-import requests
-import time
 
 from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member
 from patroni.exceptions import DCSError
-from patroni.utils import sleep
-from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)
 
@@ -17,56 +12,6 @@ class ZooKeeperError(DCSError):
     pass
 
 
-class ExhibitorEnsembleProvider(object):
-
-    TIMEOUT = 3.1
-
-    def __init__(self, hosts, port, uri_path='/exhibitor/v1/cluster/list', poll_interval=300):
-        self._exhibitor_port = port
-        self._uri_path = uri_path
-        self._poll_interval = poll_interval
-        self._exhibitors = hosts
-        self._master_exhibitors = hosts
-        self._zookeeper_hosts = ''
-        self._next_poll = None
-        while not self.poll():
-            logger.info('waiting on exhibitor')
-            sleep(5)
-
-    def poll(self):
-        if self._next_poll and self._next_poll > time.time():
-            return False
-
-        json = self._query_exhibitors(self._exhibitors)
-        if not json:
-            json = self._query_exhibitors(self._master_exhibitors)
-
-        if isinstance(json, dict) and 'servers' in json and 'port' in json:
-            self._next_poll = time.time() + self._poll_interval
-            zookeeper_hosts = ','.join([h + ':' + str(json['port']) for h in sorted(json['servers'])])
-            if self._zookeeper_hosts != zookeeper_hosts:
-                logger.info('ZooKeeper connection string has changed: %s => %s', self._zookeeper_hosts, zookeeper_hosts)
-                self._zookeeper_hosts = zookeeper_hosts
-                self._exhibitors = json['servers']
-                return True
-        return False
-
-    def _query_exhibitors(self, exhibitors):
-        random.shuffle(exhibitors)
-        for host in exhibitors:
-            uri = 'http://{0}:{1}{2}'.format(host, self._exhibitor_port, self._uri_path)
-            try:
-                response = requests.get(uri, timeout=self.TIMEOUT)
-                return response.json()
-            except RequestException:
-                pass
-        return None
-
-    @property
-    def zookeeper_hosts(self):
-        return self._zookeeper_hosts
-
-
 class ZooKeeper(AbstractDCS):
 
     def __init__(self, config):
@@ -76,16 +21,8 @@ class ZooKeeper(AbstractDCS):
         if isinstance(hosts, list):
             hosts = ','.join(hosts)
 
-        self.exhibitor = None
-        if 'exhibitor' in config:
-            exhibitor = config['exhibitor']
-            interval = exhibitor.get('poll_interval', 300)
-            self.exhibitor = ExhibitorEnsembleProvider(exhibitor['hosts'], exhibitor['port'], poll_interval=interval)
-            hosts = self.exhibitor.zookeeper_hosts
-
-        self._client = KazooClient(hosts=hosts, timeout=config['ttl'],
-                                   command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1},
-                                   connection_retry={'max_delay': 1, 'max_tries': -1})
+        self._client = KazooClient(hosts, timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1},
+                                   command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1})
         self._client.add_listener(self.session_listener)
 
         self._my_member_data = None
@@ -104,8 +41,7 @@ class ZooKeeper(AbstractDCS):
 
     def set_ttl(self, ttl):
         ttl = int(ttl * 1000)
-        # I know, it's weird to access private attributes, but there is
-        # no other way to change session_timeout without losing session
+        # I know, it's weird to access private attributes
         if self._client._session_timeout != ttl:
             self._client._session_timeout = ttl
             self._client.restart()
@@ -180,9 +116,6 @@ class ZooKeeper(AbstractDCS):
         self._cluster = Cluster(initialize, config, leader, self._last_leader_operation, members, failover)
 
     def _load_cluster(self):
-        if self.exhibitor and self.exhibitor.poll():
-            self._client.set_hosts(self.exhibitor.zookeeper_hosts)
-
         if self._fetch_cluster or self._cluster is None:
             try:
                 self._client.retry(self._inner_load_cluster)

patroni/postgresql.py

@@ -63,6 +63,7 @@ class Postgresql(object):
         self.config = config
         self.name = config['name']
         self.scope = config['scope']
+        self._database = config.get('database', 'postgres')
         self._data_dir = config['data_dir']
         self._pending_restart = False
         self._server_parameters = self.get_server_parameters(config)
@@ -74,13 +75,15 @@ class Postgresql(object):
 
         self._use_pg_rewind = config.get('use_pg_rewind', False)
         self._use_slots = config.get('use_slots', True)
+        self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
         self._major_version = self.get_major_version()
         self._schedule_load_slots = self.use_slots
 
         self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
         self.callback = config.get('callbacks') or {}
-        self._postgresql_conf = os.path.join(self._data_dir, 'postgresql.conf')
-        self._postgresql_base_conf_name = 'postgresql.base.conf'
+        config_base_name = config.get('config_base_name', 'postgresql')
+        self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf')
+        self._postgresql_base_conf_name = config_base_name + '.base.conf'
         self._postgresql_base_conf = os.path.join(self._data_dir, self._postgresql_base_conf_name)
         self._recovery_conf = os.path.join(self._data_dir, 'recovery.conf')
         self._configuration_to_save = (self._postgresql_conf, self._postgresql_base_conf,
@@ -112,10 +115,13 @@ class Postgresql(object):
     def use_slots(self):
         return self._use_slots and self._major_version >= 9.4
 
+    def _version_file_exists(self):
+        return not self.data_directory_empty() and os.path.isfile(self._version_file)
+
     def get_major_version(self):
-        if not self.data_directory_empty():
+        if self._version_file_exists():
             try:
-                with open(os.path.join(self._data_dir, 'PG_VERSION')) as f:
+                with open(self._version_file) as f:
                     return float(f.read())
             except Exception:
                 logger.exception('Failed to read PG_VERSION from %s', self._data_dir)
@@ -130,8 +136,8 @@ class Postgresql(object):
 
     def resolve_connection_addresses(self):
         self._local_address = self.get_local_address()
-        self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format(
-            connect_address=self._connect_address or self._local_address, **self._replication)
+        self.connection_string = 'postgres://{username}:{password}@{connect_address}/{database}'.format(
+            connect_address=self._connect_address or self._local_address, database=self._database, **self._replication)
 
     def reload_config(self, config):
         server_parameters = self.get_server_parameters(config)
@@ -244,7 +250,7 @@ class Postgresql(object):
 
     @property
     def _connect_kwargs(self):
-        r = parseurl('postgres://{0}/postgres'.format(self._local_address))
+        r = parseurl('postgres://{0}/{1}'.format(self._local_address, self._database))
         if 'username' in self._superuser:
             r['user'] = self._superuser['username']
         if 'password' in self._superuser:
@@ -428,7 +434,16 @@ class Postgresql(object):
         return not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
 
     def is_running(self):
-        return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null 2>&1', shell=True) == 0
+        if not (self._version_file_exists() and os.path.isfile(self._postmaster_pid)):
+            return False
+        try:
+            with open(self._postmaster_pid) as f:
+                pid = int(f.readline())
+                if pid < 0:
+                    pid = -pid
+                return pid > 0 and pid != os.getpid() and pid != os.getppid() and (os.kill(pid, 0) or True)
+        except Exception:
+            return False
 
     def call_nowait(self, cb_name):
         """ pick a callback command and call it without waiting for it to finish """
@@ -613,8 +628,9 @@ class Postgresql(object):
         r = parseurl(leader.conn_url)
         r.update(self._superuser)
         r['user'] = r.pop('username')
+        r['database'] = self._database
         env = self.write_pgpass(r)
-        pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
+        pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r)
         # first run a checkpoint on a promoted master in order
         # to make it store the new timeline ([email protected])
         self.checkpoint(r)
@@ -660,7 +676,7 @@ class Postgresql(object):
         for opt, val in sorted((options or {}).items()):
             cmd.extend(['-c', '{0}={1}'.format(opt, val)])
         # need a database name to connect
-        cmd.append('postgres')
+        cmd.append(self._database)
         p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
         if p:
             if command:

postgres0.yml

@@ -53,6 +53,7 @@ bootstrap:
       options:
         - createrole
         - createdb
+
 postgresql:
   listen: 127.0.0.1:5432
   connect_address: 127.0.0.1:5432

postgres1.yml

@@ -23,13 +23,13 @@ etcd:
 #  hosts:
 #    - 127.0.0.1:2181
 #    - 127.0.0.2:2181
-#  exhibitor:
-#    poll_interval: 300
-#    port: 8181
-#    hosts:
-#      - host1
-#      - host2
-#      - host3
+#exhibitor:
+#  poll_interval: 300
+#  port: 8181
+#  hosts:
+#    - host1
+#    - host2
+#    - host3
 postgresql:
   name: postgresql1
   scope: *scope

postgres2.yml

@@ -23,13 +23,13 @@ etcd:
 #  hosts:
 #    - 127.0.0.1:2181
 #    - 127.0.0.2:2181
-#  exhibitor:
-#    poll_interval: 300
-#    port: 8181
-#    hosts:
-#      - host1
-#      - host2
-#      - host3
+#exhibitor:
+#  poll_interval: 300
+#  port: 8181
+#  hosts:
+#    - host1
+#    - host2
+#    - host3
 postgresql:
   name: postgresql2
   scope: *scope

tests/test_ctl.py

@@ -70,7 +70,7 @@ class TestCtl(unittest.TestCase):
         assert parse_dcs('') == {'etcd': {'host': 'localhost:4001'}}
         assert parse_dcs('localhost:8500') == {'consul': {'host': 'localhost:8500'}}
         assert parse_dcs('zookeeper://localhost') == {'zookeeper': {'hosts': ['localhost:2181']}}
-        assert parse_dcs('exhibitor://dummy') == {'zookeeper': {'exhibitor': {'hosts': ['dummy'], 'port': 8181}}}
+        assert parse_dcs('exhibitor://dummy') == {'exhibitor': {'hosts': ['dummy'], 'port': 8181}}
         assert parse_dcs('consul://localhost') == {'consul': {'host': 'localhost:8500'}}
         self.assertRaises(PatroniCtlException, parse_dcs, 'invalid://test')
 

tests/test_exhibitor.py

@@ -0,0 +1,31 @@
+import unittest
+
+from mock import Mock, patch
+from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor
+from patroni.dcs.zookeeper import ZooKeeperError
+from test_etcd import SleepException, requests_get
+from test_zookeeper import MockKazooClient
+
+
+@patch('requests.get', requests_get)
+@patch('time.sleep', Mock(side_effect=SleepException))
+class TestExhibitorEnsembleProvider(unittest.TestCase):
+
+    def test_init(self):
+        self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181)
+
+    def test_poll(self):
+        self.assertFalse(ExhibitorEnsembleProvider(['exhibitor'], 8181).poll())
+
+
+class TestExhibitor(unittest.TestCase):
+
+    @patch('requests.get', requests_get)
+    @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
+    def setUp(self):
+        self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',
+                            'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
+
+    @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True))
+    def test_get_cluster(self):
+        self.assertRaises(ZooKeeperError, self.e.get_cluster)

tests/test_patroni.py

@@ -20,6 +20,7 @@ from test_postgresql import Postgresql, psycopg2_connect
 @patch.object(Postgresql, 'write_pg_hba', Mock())
 @patch.object(Postgresql, '_write_postgresql_conf', Mock())
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
+@patch.object(Postgresql, 'is_running', Mock(return_value=True))
 @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
 @patch.object(AsyncExecutor, 'run', Mock())
 @patch.object(etcd.Client, 'write', etcd_write)

tests/test_postgresql.py

@@ -160,6 +160,7 @@ class TestPostgresql(unittest.TestCase):
     @patch('psycopg2.connect', psycopg2_connect)
     @patch('os.rename', Mock())
     @patch.object(Postgresql, 'get_major_version', Mock(return_value=9.4))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def setUp(self):
         self.data_dir = 'data/test0'
         if not os.path.exists(self.data_dir):
@@ -197,9 +198,11 @@ class TestPostgresql(unittest.TestCase):
     def test_delete_trigger_file(self):
         self.p.delete_trigger_file()
 
-    def test_start(self):
+    @patch.object(Postgresql, 'is_running')
+    def test_start(self, mock_is_running):
+        mock_is_running.return_value = True
         self.assertTrue(self.p.start())
-        self.p.is_running = false
+        mock_is_running.return_value = False
         open(os.path.join(self.data_dir, 'postmaster.pid'), 'w').close()
         pg_conf = os.path.join(self.data_dir, 'postgresql.conf')
         open(pg_conf, 'w').close()
@@ -208,16 +211,16 @@ class TestPostgresql(unittest.TestCase):
             lines = f.readlines()
             self.assertTrue("f.oo = 'bar'\n" in lines)
 
-    def test_stop(self):
+    @patch.object(Postgresql, 'is_running')
+    def test_stop(self, mock_is_running):
+        mock_is_running.return_value = True
         self.assertTrue(self.p.stop())
         with patch('subprocess.call', Mock(return_value=1)):
+            mock_is_running.return_value = False
             self.assertTrue(self.p.stop())
-            self.p.is_running = Mock(return_value=True)
-            self.assertFalse(self.p.stop())
 
     def test_restart(self):
         self.p.start = false
-        self.p.is_running = false
         self.assertFalse(self.p.restart())
         self.assertEquals(self.p.state, 'restart failed (restarting)')
 
@@ -237,6 +240,7 @@ class TestPostgresql(unittest.TestCase):
     @patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
     @patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
     @patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_follow(self, mock_pg_rewind):
         self.p.follow(None, None)
         self.p.follow(self.leader, self.leader)
@@ -284,6 +288,7 @@ class TestPostgresql(unittest.TestCase):
         with patch('subprocess.call', Mock(side_effect=Exception("foo"))):
             self.assertEquals(self.p.create_replica(self.leader), 1)
 
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_sync_replication_slots(self):
         self.p.start()
         cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None)
@@ -312,9 +317,11 @@ class TestPostgresql(unittest.TestCase):
     def test_reload(self):
         self.assertTrue(self.p.reload())
 
-    def test_is_healthy(self):
+    @patch.object(Postgresql, 'is_running')
+    def test_is_healthy(self, mock_is_running):
+        mock_is_running.return_value = True
         self.assertTrue(self.p.is_healthy())
-        self.p.is_running = false
+        mock_is_running.return_value = False
         self.assertFalse(self.p.is_healthy())
 
     def test_promote(self):
@@ -325,6 +332,13 @@ class TestPostgresql(unittest.TestCase):
     def test_last_operation(self):
         self.assertEquals(self.p.last_operation(), '0')
 
+    @patch('os.path.isfile', Mock(return_value=True))
+    @patch('os.kill', Mock(side_effect=Exception))
+    @patch.object(builtins, 'open', mock_open(read_data='-999999999999999'))
+    @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
+    def test_is_running(self):
+        self.assertFalse(self.p.is_running())
+
     @patch('subprocess.Popen', Mock(side_effect=OSError))
     def test_call_nowait(self):
         self.assertFalse(self.p.call_nowait('on_start'))
@@ -332,6 +346,7 @@ class TestPostgresql(unittest.TestCase):
     def test_non_existing_callback(self):
         self.assertFalse(self.p.call_nowait('foobar'))
 
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_is_leader_exception(self):
         self.p.start()
         self.p.query = Mock(side_effect=psycopg2.OperationalError("not supported"))
@@ -343,11 +358,11 @@ class TestPostgresql(unittest.TestCase):
     @patch('os.rename', Mock())
     @patch('os.path.isdir', Mock(return_value=True))
     def test_move_data_directory(self):
-        self.p.is_running = false
         self.p.move_data_directory()
         with patch('os.rename', Mock(side_effect=OSError)):
             self.p.move_data_directory()
 
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_bootstrap(self):
         with patch('subprocess.call', Mock(return_value=1)):
             self.assertRaises(PostgresException, self.p.bootstrap, {})
@@ -478,6 +493,7 @@ class TestPostgresql(unittest.TestCase):
         self.p.config['foo'] = {'command': 'bar'}
         self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo'))
 
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_reload_config(self):
         parameters = self._PARAMETERS.copy()
         parameters.pop('f.oo')
@@ -490,6 +506,9 @@ class TestPostgresql(unittest.TestCase):
         parameters.pop('search_path')
         self.p.reload_config({'retry_timeout': 10, 'listen': '*:5433', 'parameters': parameters})
 
-    @patch.object(builtins, 'open', mock_open(read_data='9.4'))
+    @patch.object(Postgresql, '_version_file_exists', Mock(return_value=True))
     def test_get_major_version(self):
-        self.assertEquals(self.p.get_major_version(), 9.4)
+        with patch.object(builtins, 'open', mock_open(read_data='9.4')):
+            self.assertEquals(self.p.get_major_version(), 9.4)
+        with patch.object(builtins, 'open', Mock(side_effect=Exception)):
+            self.assertEquals(self.p.get_major_version(), 0.0)

tests/test_zookeeper.py

@@ -5,8 +5,7 @@ from kazoo.client import KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.protocol.states import ZnodeStat
 from mock import Mock, patch
-from patroni.dcs.zookeeper import Leader, ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError
-from test_etcd import SleepException, requests_get
+from patroni.dcs.zookeeper import Leader, ZooKeeper, ZooKeeperError
 
 
 class MockKazooClient(Mock):
@@ -14,6 +13,9 @@ class MockKazooClient(Mock):
     leader = False
     exists = True
 
+    def __init__(self, *args, **kwargs):
+        super(MockKazooClient, self).__init__()
+
     @property
     def client_id(self):
         return (-1, '')
@@ -90,21 +92,12 @@ class MockKazooClient(Mock):
             raise NoNodeError
 
 
-@patch('requests.get', requests_get)
-@patch('time.sleep', Mock(side_effect=SleepException))
-class TestExhibitorEnsembleProvider(unittest.TestCase):
-
-    def test_init(self):
-        self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181)
-
-
 class TestZooKeeper(unittest.TestCase):
 
-    @patch('requests.get', requests_get)
     @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
     def setUp(self):
-        self.zk = ZooKeeper({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181},
-                             'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
+        self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test',
+                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
 
     def test_session_listener(self):
         self.zk.session_listener(KazooState.SUSPENDED)
@@ -129,11 +122,12 @@ class TestZooKeeper(unittest.TestCase):
 
     def test_get_cluster(self):
         self.assertRaises(ZooKeeperError, self.zk.get_cluster)
-        self.zk.exhibitor.poll = lambda: True
         cluster = self.zk.get_cluster()
         self.assertIsInstance(cluster.leader, Leader)
         self.zk.touch_member('foo')
-        self.zk.delete_leader()
+
+    def test_delete_leader(self):
+        self.assertTrue(self.zk.delete_leader())
 
     def test_set_failover_value(self):
         self.zk.set_failover_value('')