Merge pull request #2 from CyberDem0n/restapi

Restapi

governor.py

@@ -7,6 +7,7 @@ import sys
 import time
 import yaml
 
+from helpers.api import RestApiServer
 from helpers.etcd import Etcd
 from helpers.postgresql import Postgresql
 from helpers.ha import Ha
@@ -34,9 +35,12 @@ class Governor:
         self.etcd = Etcd(config['etcd'])
         self.postgresql = Postgresql(config['postgresql'])
         self.ha = Ha(self.postgresql, self.etcd)
+        host, port = config['restapi']['listen'].split(':')
+        self.api = RestApiServer(self, config['restapi'])
 
-    def touch_member(self):
-        return self.etcd.touch_member(self.postgresql.name, self.postgresql.connection_string)
+    def touch_member(self, ttl=None):
+        connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string
+        return self.etcd.touch_member(self.postgresql.name, connection_string, ttl)
 
     def initialize(self):
         # wait for etcd to be available
@@ -64,6 +68,7 @@ class Governor:
             self.postgresql.load_replication_slots()
 
     def run(self):
+        self.api.start()
         while True:
             self.touch_member()
             logging.info(self.ha.run_cycle())
@@ -71,6 +76,10 @@ class Governor:
 
 
 def main():
+    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
+    signal.signal(signal.SIGTERM, sigterm_handler)
+    signal.signal(signal.SIGCHLD, sigchld_handler)
+
     if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
         print('Usage: {} config.yml'.format(sys.argv[0]))
         return
@@ -83,12 +92,10 @@ def main():
         governor.initialize()
         governor.run()
     finally:
+        governor.touch_member(300)  # schedule member removal
         governor.postgresql.stop()
         governor.etcd.delete_leader(governor.postgresql.name)
 
 
 if __name__ == '__main__':
-    logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
-    signal.signal(signal.SIGTERM, sigterm_handler)
-    signal.signal(signal.SIGCHLD, sigchld_handler)
     main()

helpers/api.py

@@ -0,0 +1,74 @@
+import json
+import logging
+import psycopg2
+import sys
+
+from threading import Thread
+
+if sys.hexversion >= 0x03000000:
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+else:
+    from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
+
+
+logger = logging.getLogger(__name__)
+
+
+class RestApiHandler(BaseHTTPRequestHandler):
+
+    def do_GET(self):
+        try:
+            response = self.get_postgresql_status()
+        except (psycopg2.OperationalError, psycopg2.InterfaceError):
+            logging.exception('get_postgresql_status')
+            response = {'running': False}
+
+        path = '/master' if self.path == '/' else self.path
+        status_code = 200 if response['running'] and response['role'] in path else 503
+
+        self.send_response(status_code)
+        self.send_header('Content-Type', 'application/json')
+        self.end_headers()
+        self.wfile.write(json.dumps(response).encode('utf-8'))
+
+    def get_postgresql_status(self):
+        if not self.server.governor.postgresql.is_running():
+            return {'running': False}
+        cursor = self.server._cursor()
+        cursor.execute("""SELECT to_char(pg_postmaster_start_time(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
+                                 pg_is_in_recovery(),
+                                 CASE WHEN pg_is_in_recovery()
+                                      THEN null
+                                      ELSE pg_current_xlog_location() END,
+                                 pg_last_xlog_receive_location(),
+                                 pg_last_xlog_replay_location(),
+                                 pg_is_in_recovery() AND pg_is_xlog_replay_paused()""")
+        row = cursor.fetchone()
+        return {
+            'running': True,
+            'postmaster_start_time': row[0],
+            'role': 'slave' if row[1] else 'master',
+            'xlog': ({
+                'received_location': row[3],
+                'replayed_location': row[4],
+                'paused': row[5]} if row[1] else {
+                'location': row[2]
+            })
+        }
+
+
+class RestApiServer(HTTPServer, Thread):
+
+    def __init__(self, governor, config):
+        self.connection_string = 'http://{}/governor'.format(config.get('connect_address', None) or config['listen'])
+        host, port = config['listen'].split(':')
+        HTTPServer.__init__(self, (host, int(port)), RestApiHandler)
+        Thread.__init__(self, target=self.serve_forever)
+        self.governor = governor
+        self._cursor_holder = None
+        self.daemon = True
+
+    def _cursor(self):
+        if not self._cursor_holder or self._cursor_holder.closed:
+            self._cursor_holder = self.governor.postgresql.connection().cursor()
+        return self._cursor_holder

helpers/errors.py

@@ -9,7 +9,3 @@ class EtcdError(Exception):
 
 class CurrentLeaderError(EtcdError):
     pass
-
-
-class HealthiestMemberError(EtcdError):
-    pass

helpers/etcd.py

@@ -2,16 +2,16 @@ import logging
 import requests
 import time
 
+from requests.exceptions import RequestException
 from collections import namedtuple
 from helpers.errors import CurrentLeaderError, EtcdError
 
 logger = logging.getLogger(__name__)
 
-
 Member = namedtuple('Member', 'hostname,address,ttl')
 
 
-class Cluster(namedtuple('Cluster', 'leader,last_leader_operation,members')):
+class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')):
 
     def is_unlocked(self):
         return not (self.leader and self.leader.hostname)
@@ -35,7 +35,7 @@ class Etcd:
                 response = requests.get(self.client_url(path))
                 if response.status_code == 200:
                     break
-            except Exception as e:
+            except RequestException as e:
                 logger.exception('get_client_path')
                 ex = e
 
@@ -54,15 +54,15 @@ class Etcd:
         try:
             response = requests.put(self.client_url(path), data=data)
             return response.status_code in [200, 201, 202, 204]
-        except:
+        except RequestException:
             logger.exception('PUT %s data=%s', path, data)
-            return False
+        raise EtcdError('Etcd is not responding properly')
 
     def delete_client_path(self, path):
         try:
             response = requests.delete(self.client_url(path))
             return response.status_code in [200, 202, 204]
-        except:
+        except RequestException:
             logger.exception('DELETE %s', path)
             return False
 
@@ -87,6 +87,8 @@ class Etcd:
         try:
             response, status_code = self.get_client_path('?recursive=true')
             if status_code == 200:
+                node = self.find_node(response['node'], '/initialize')
+                initialize = True if node else False
                 # get list of members
                 node = self.find_node(response['node'], '/members') or {'nodes': []}
                 members = [Member(n['key'].split('/')[-1], n['value'], n.get('ttl', None)) for n in node['nodes']]
@@ -108,11 +110,11 @@ class Etcd:
                             leader = m
                             break
                     if not leader:
-                        leader = Member(leader['value'], None, None)
+                        leader = Member(node['value'], None, None)
 
-                return Cluster(leader, last_leader_operation, members)
+                return Cluster(initialize, leader, last_leader_operation, members)
             elif status_code == 404:
-                return Cluster(None, None, [])
+                return Cluster(False, None, None, [])
         except:
             logger.exception('get_cluster')
 
@@ -122,27 +124,43 @@ class Etcd:
         try:
             cluster = self.get_cluster()
             return None if cluster.is_unlocked() else cluster.leader
-        except:
-            raise CurrentLeaderError("Etcd is not responding properly")
+        except EtcdError:
+            raise CurrentLeaderError('Etcd is not responding properly')
 
-    def touch_member(self, member, connection_string):
-        return self.put_client_path('/members/' + member, value=connection_string, ttl=self.member_ttl)
+    def touch_member(self, member, connection_string, ttl=None):
+        try:
+            return self.put_client_path('/members/' + member, value=connection_string, ttl=ttl or self.member_ttl)
+        except EtcdError:
+            return False
 
     def take_leader(self, value):
-        return self.put_client_path('/leader', value=value, ttl=self.ttl)
+        try:
+            return self.put_client_path('/leader', value=value, ttl=self.ttl)
+        except EtcdError:
+            return False
 
     def attempt_to_acquire_leader(self, value):
-        ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevExist=False)
-        ret or logger.info('Could not take out TTL lock')
-        return ret
+        try:
+            ret = self.put_client_path('/leader', value=value, ttl=self.ttl, prevExist=False)
+            ret or logger.info('Could not take out TTL lock')
+            return ret
+        except EtcdError:
+            return False
 
     def update_leader(self, state_handler):
-        ret = self.put_client_path('/leader', value=state_handler.name, ttl=self.ttl, prevValue=state_handler.name)
-        ret and self.put_client_path('/optime/leader', value=state_handler.last_operation())
-        return ret
+        if self.put_client_path('/leader', value=state_handler.name, ttl=self.ttl, prevValue=state_handler.name):
+            try:
+                self.put_client_path('/optime/leader', value=state_handler.last_operation())
+            except EtcdError:
+                pass
+            return True
+        return False
 
     def race(self, path, value):
-        return self.put_client_path(path, value=value, prevExist=False)
+        try:
+            return self.put_client_path(path, value=value, prevExist=False)
+        except EtcdError:
+            return False
 
     def delete_member(self, member):
         return self.delete_client_path('/members/' + member)

helpers/ha.py

@@ -1,6 +1,6 @@
 import logging
 
-from helpers.errors import EtcdError, HealthiestMemberError
+from helpers.errors import EtcdError
 from psycopg2 import InterfaceError, OperationalError
 
 logger = logging.getLogger(__name__)
@@ -94,5 +94,3 @@ class Ha:
                 return 'demoted self because etcd is not accessible and i was a leader'
         except (InterfaceError, OperationalError):
             logger.error('Error communicating with Postgresql.  Will try again')
-        except HealthiestMemberError:
-            logger.error('failed to determine healthiest member fromt etcd')

helpers/postgresql.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import psycopg2
+import subprocess
 import sys
 import time
 
@@ -17,15 +18,19 @@ logger = logging.getLogger(__name__)
 
 def parseurl(url):
     r = urlparse(url)
-    return {
+    ret = {
         'host': r.hostname,
         'port': r.port or 5432,
-        'user': r.username,
-        'password': r.password,
         'database': r.path[1:],
         'fallback_application_name': 'Governor',
-        'connect_timeout': 5,
+        'connect_timeout': 3,
+        'options': '-c statement_timeout=2000',
     }
+    if r.username:
+        ret['user'] = r.username
+    if r.password:
+        ret['password'] = r.password
+    return ret
 
 
 class Postgresql:
@@ -38,7 +43,7 @@ class Postgresql:
         self.replication = config['replication']
         self.recovery_conf = os.path.join(self.data_dir, 'recovery.conf')
         self.pid_path = os.path.join(self.data_dir, 'postmaster.pid')
-        self._pg_ctl = 'pg_ctl -w -D ' + self.data_dir
+        self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir]
 
         self.local_address = self.get_local_address()
         connect_address = config.get('connect_address', None) or self.local_address
@@ -55,7 +60,8 @@ class Postgresql:
 
     def connection(self):
         if not self._connection or self._connection.closed != 0:
-            self._connection = psycopg2.connect('postgres://{}/postgres'.format(self.local_address))
+            r = parseurl('postgres://{}/postgres'.format(self.local_address))
+            self._connection = psycopg2.connect(**r)
             self._connection.autocommit = True
         return self._connection
 
@@ -93,7 +99,7 @@ class Postgresql:
         return not os.path.exists(self.data_dir) or os.listdir(self.data_dir) == []
 
     def initialize(self):
-        ret = os.system(self._pg_ctl + ' initdb -o --encoding=UTF8') == 0
+        ret = subprocess.call(self._pg_ctl + ['initdb', '-o', '--encoding=UTF8']) == 0
         ret and self.write_pg_hba()
         return ret
 
@@ -107,8 +113,8 @@ class Postgresql:
 
         try:
             os.environ['PGPASSFILE'] = pgpass
-            return os.system('pg_basebackup -R -D {data_dir} --host={host} --port={port} -U {user}'.format(
-                data_dir=self.data_dir, **r)) == 0
+            return subprocess.call(['pg_basebackup', '-R', '-D', self.data_dir,
+                                    '--host=' + r['host'], '--port=' + str(r['port']), '-U', r['user']]) == 0
         finally:
             os.environ.pop('PGPASSFILE')
 
@@ -116,7 +122,7 @@ class Postgresql:
         return not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
 
     def is_running(self):
-        return os.system(self._pg_ctl + ' status > /dev/null') == 0
+        return subprocess.call(' '.join(self._pg_ctl) + ' status > /dev/null', shell=True) == 0
 
     def start(self):
         if self.is_running():
@@ -128,18 +134,18 @@ class Postgresql:
             os.remove(self.pid_path)
             logger.info('Removed %s', self.pid_path)
 
-        ret = os.system(self._pg_ctl + ' start -o "{}"'.format(self.server_options())) == 0
+        ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0
         ret and self.load_replication_slots()
         return ret
 
     def stop(self):
-        return os.system(self._pg_ctl + ' stop -m fast') != 0
+        return subprocess.call(self._pg_ctl + ['stop', '-m', 'fast']) != 0
 
     def reload(self):
-        return os.system(self._pg_ctl + ' reload') == 0
+        return subprocess.call(self._pg_ctl + ['reload']) == 0
 
     def restart(self):
-        return os.system(self._pg_ctl + ' restart -m fast') == 0
+        return subprocess.call(self._pg_ctl + ['restart', '-m', 'fast']) == 0
 
     def server_options(self):
         options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port)
@@ -164,7 +170,8 @@ class Postgresql:
             if member.hostname == self.name:
                 continue
             try:
-                member_conn = psycopg2.connect(parseurl(member.address))
+                r = parseurl(member.address)
+                member_conn = psycopg2.connect(**r)
                 member_conn.autocommit = True
                 member_cursor = member_conn.cursor()
                 member_cursor.execute(
@@ -224,7 +231,7 @@ primary_conninfo = '{}'
         self.restart()
 
     def promote(self):
-        return os.system(self._pg_ctl + ' promote') == 0
+        return subprocess.call(self._pg_ctl + ['promote']) == 0
 
     def demote(self, leader):
         self.follow_the_leader(leader)

postgres0.yml

@@ -1,4 +1,7 @@
 loop_wait: 10
+restapi:
+  listen: 127.0.0.1:8008
+  connect_address: 127.0.0.1:8008
 etcd:
   scope: batman
   ttl: 30

postgres1.yml

@@ -1,4 +1,7 @@
 loop_wait: 10
+restapi:
+  listen: 127.0.0.1:8009
+  connect_address: 127.0.0.1:8009
 etcd:
   scope: batman
   ttl: 30

tests/test_api.py

@@ -0,0 +1,66 @@
+import psycopg2
+import sys
+import unittest
+
+from helpers.api import RestApiHandler, RestApiServer
+from test_postgresql import psycopg2_connect
+
+if sys.hexversion >= 0x03000000:
+    from io import BytesIO as IO
+else:
+    from StringIO import StringIO as IO
+
+
+def false(*args, **kwargs):
+    return False
+
+
+def throws(*args, **kwargs):
+    raise psycopg2.OperationalError()
+
+
+class MockPostgresql:
+
+    def connection(self):
+        return psycopg2_connect()
+
+    def is_running(self):
+        return True
+
+
+class MockGovernor:
+
+    def __init__(self):
+        self.postgresql = MockPostgresql()
+
+
+class MockRequest:
+
+    def __init__(self, path):
+        self.path = path
+
+    def makefile(self, *args, **kwargs):
+        return IO(self.path)
+
+
+class MockRestApiServer(RestApiServer):
+
+    def __init__(self, Handler, path, *args):
+        self.governor = MockGovernor()
+        if len(args) > 0:
+            self.governor.postgresql.is_running = args[0]
+        self._cursor_holder = None
+        Handler(MockRequest(path), ('0.0.0.0', 8080), self)
+
+
+class TestRestApiHandler(unittest.TestCase):
+
+    def __init__(self, method_name='runTest'):
+        super(TestRestApiHandler, self).__init__(method_name)
+
+    def test_do_GET(self):
+        MockRestApiServer(RestApiHandler, b'GET /')
+        MockRestApiServer(RestApiHandler, b'GET /', throws)
+
+    def test_get_postgresql_status(self):
+        MockRestApiServer(RestApiHandler, b'GET /', false)

tests/test_etcd.py

@@ -17,28 +17,39 @@ class MockResponse:
         return json.loads(self.content)
 
 
+class MockPostgresql:
+    name = ''
+
+    def last_operation(self):
+        return 0
+
+
 def requests_get(url, **kwargs):
     if url.startswith('http://local'):
-        raise Exception()
+        raise requests.exceptions.RequestException()
     response = MockResponse()
-    if url.startswith('http://remote'):
+    if url.startswith('http://remote') or url.startswith('http://127.0.0.1'):
         response.content = '{"action":"get","node":{"key":"/service/batman5","dir":true,"nodes":[{"key":"/service/batman5/initialize","value":"postgresql0","modifiedIndex":1582,"createdIndex":1582},{"key":"/service/batman5/leader","value":"postgresql1","expiration":"2015-05-15T09:11:00.037397538Z","ttl":21,"modifiedIndex":20728,"createdIndex":20434},{"key":"/service/batman5/optime","dir":true,"nodes":[{"key":"/service/batman5/optime/leader","value":"2164261704","modifiedIndex":20729,"createdIndex":20729}],"modifiedIndex":20437,"createdIndex":20437},{"key":"/service/batman5/members","dir":true,"nodes":[{"key":"/service/batman5/members/postgresql1","value":"postgres://replicator:[email protected]:5434/postgres","expiration":"2015-05-15T09:10:59.949384522Z","ttl":21,"modifiedIndex":20727,"createdIndex":20727},{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:[email protected]:5433/postgres","expiration":"2015-05-15T09:11:09.611860899Z","ttl":30,"modifiedIndex":20730,"createdIndex":20730}],"modifiedIndex":1581,"createdIndex":1581}],"modifiedIndex":1581,"createdIndex":1581}}'
     elif url.startswith('http://other'):
         response.status_code = 404
+    elif url.startswith('http://noleader'):
+        response.content = '{"action":"get","node":{"key":"/service/batman5","dir":true,"nodes":[{"key":"/service/batman5/initialize","value":"postgresql0","modifiedIndex":1582,"createdIndex":1582},{"key":"/service/batman5/leader","value":"postgresql1","expiration":"2015-05-15T09:11:00.037397538Z","ttl":21,"modifiedIndex":20728,"createdIndex":20434},{"key":"/service/batman5/optime","dir":true,"nodes":[{"key":"/service/batman5/optime/leader","value":"2164261704","modifiedIndex":20729,"createdIndex":20729}],"modifiedIndex":20437,"createdIndex":20437},{"key":"/service/batman5/members","dir":true,"nodes":[{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:[email protected]:5433/postgres","expiration":"2015-05-15T09:11:09.611860899Z","ttl":30,"modifiedIndex":20730,"createdIndex":20730}],"modifiedIndex":1581,"createdIndex":1581}],"modifiedIndex":1581,"createdIndex":1581}}'
     return response
 
 
 def requests_put(url, **kwargs):
-    if url.startswith('http://local'):
-        raise Exception()
+    if url.startswith('http://local') or '/optime/leader' in url:
+        raise requests.exceptions.RequestException()
     response = MockResponse()
     response.status_code = 201
+    if url.startswith('http://other'):
+        response.status_code = 404
     return response
 
 
 def requests_delete(url):
     if url.startswith('http://local'):
-        raise Exception()
+        raise requests.exceptions.RequestException()
     response = MockResponse()
     response.status_code = 204
     return response
@@ -65,7 +76,7 @@ class TestEtcd(unittest.TestCase):
         self.assertRaises(Exception, self.etcd.get_client_path, '', 2)
 
     def test_put_client_path(self):
-        self.assertFalse(self.etcd.put_client_path(''))
+        self.assertRaises(EtcdError, self.etcd.put_client_path, '')
 
     def test_delete_client_path(self):
         self.assertFalse(self.etcd.delete_client_path(''))
@@ -77,6 +88,29 @@ class TestEtcd(unittest.TestCase):
         self.assertIsInstance(cluster, Cluster)
         self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other')
         self.etcd.get_cluster()
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('other', 'noleader')
+        self.etcd.get_cluster()
 
     def test_current_leader(self):
         self.assertRaises(CurrentLeaderError, self.etcd.current_leader)
+
+    def test_touch_member(self):
+        self.assertFalse(self.etcd.touch_member('', ''))
+
+    def test_take_leader(self):
+        self.assertFalse(self.etcd.take_leader(''))
+
+    def test_attempt_to_acquire_leader(self):
+        self.assertFalse(self.etcd.attempt_to_acquire_leader(''))
+
+    def test_update_leader(self):
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('local', 'remote')
+        self.assertTrue(self.etcd.update_leader(MockPostgresql()))
+        self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other')
+        self.assertFalse(self.etcd.update_leader(MockPostgresql()))
+
+    def test_race(self):
+        self.assertFalse(self.etcd.race('', ''))
+
+    def test_delete_member(self):
+        self.assertFalse(self.etcd.delete_member(''))

tests/test_governor.py

@@ -1,16 +1,22 @@
 import os
 import psycopg2
-import unittest
 import requests
+import subprocess
 import sys
 import time
+import unittest
 import yaml
 
-from governor import Governor, main, sigchld_handler
+from governor import Governor, main, sigchld_handler, sigterm_handler
 from test_ha import true, false
-from test_postgresql import Postgresql, os_system, psycopg2_connect
+from test_postgresql import Postgresql, subprocess_call, psycopg2_connect
 from test_etcd import requests_get, requests_put, requests_delete
 
+if sys.hexversion >= 0x03000000:
+    import http.server as BaseHTTPServer
+else:
+    import BaseHTTPServer
+
 
 def nop(*args, **kwargs):
     pass
@@ -20,6 +26,10 @@ def os_waitpid(a, b):
     return (0, 0)
 
 
+def time_sleep(_):
+    raise Exception()
+
+
 class TestGovernor(unittest.TestCase):
 
     def __init__(self, method_name='runTest'):
@@ -28,25 +38,37 @@ class TestGovernor(unittest.TestCase):
         super(TestGovernor, self).__init__(method_name)
 
     def set_up(self):
-        os.system = os_system
+        self.touched = False
+        subprocess.call = subprocess_call
         psycopg2.connect = psycopg2_connect
         requests.get = requests_get
         requests.put = requests_put
         requests.delete = requests_delete
         time.sleep = nop
-        Governor.run = nop
         self.write_pg_hba = Postgresql.write_pg_hba
         self.write_recovery_conf = Postgresql.write_recovery_conf
         Postgresql.write_pg_hba = nop
         Postgresql.write_recovery_conf = nop
+        BaseHTTPServer.HTTPServer.__init__ = nop
 
     def tear_down(self):
         Postgresql.write_pg_hba = self.write_pg_hba
         Postgresql.write_recovery_conf = self.write_recovery_conf
 
+    def test_sigterm_handler(self):
+        self.assertRaises(SystemExit, sigterm_handler, None, None)
+
     def test_governor_main(self):
-        sys.argv = ['governor.py', 'postgres0.yml']
         main()
+        sys.argv = ['governor.py', 'postgres0.yml']
+        time.sleep = time_sleep
+        self.assertRaises(Exception, main)
+
+    def touch_member(self):
+        if not self.touched:
+            self.touched = True
+            return False
+        return True
 
     def test_governor_initialize(self):
         with open('postgres0.yml', 'r') as f:
@@ -60,7 +82,12 @@ class TestGovernor(unittest.TestCase):
             g.etcd.race = false
             g.initialize()
             g.postgresql.data_directory_empty = false
+            g.touch_member = self.touch_member
             g.initialize()
+            g.postgresql.data_directory_empty = true
+            time.sleep = time_sleep
+            g.postgresql.sync_from_leader = false
+            self.assertRaises(Exception, g.initialize)
 
     def test_sigchld_handler(self):
         sigchld_handler(None, None)

tests/test_ha.py

@@ -72,7 +72,8 @@ class TestHa(unittest.TestCase):
         self.p = MockPostgresql()
         self.e = Etcd({'ttl': 30, 'host': 'remotehost', 'scope': 'test'})
         self.ha = Ha(self.p, self.e)
-        self.ha.cluster = Cluster(None, None, [])
+        self.ha.load_cluster_from_etcd()
+        self.ha.cluster = Cluster(False, None, None, [])
         self.ha.load_cluster_from_etcd = nop
 
     def test_start_as_slave(self):

tests/test_postgresql.py

@@ -1,13 +1,14 @@
 import os
 import psycopg2
-import unittest
 import shutil
+import subprocess
+import unittest
 
 from helpers.etcd import Cluster, Member
 from helpers.postgresql import Postgresql
 
 
-def os_system(cmd):
+def subprocess_call(cmd, shell=False):
     return 0
 
 
@@ -89,7 +90,7 @@ class TestPostgresql(unittest.TestCase):
         super(TestPostgresql, self).__init__(method_name)
 
     def set_up(self):
-        os.system = os_system
+        subprocess.call = subprocess_call
         self.p = Postgresql({'name': 'test0', 'data_dir': 'data/test0', 'listen': '127.0.0.1, 127.0.0.2:5432', 'connect_address': '127.0.0.2:5432', 'replication': {
                             'username': 'replicator', 'password': 'rep-pass', 'network': '127.0.0.1/32'}, 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}})
         psycopg2.connect = psycopg2_connect
@@ -142,7 +143,7 @@ class TestPostgresql(unittest.TestCase):
         leader = Member('leader', 'postgres://replicator:[email protected]:5435/postgres', 28)
         me = Member('test0', 'postgres://replicator:[email protected]:5434/postgres', 28)
         other = Member('test1', 'postgres://replicator:[email protected]:5433/postgres', 28)
-        cluster = Cluster(leader, 0, [me, other, leader])
+        cluster = Cluster(True, leader, 0, [me, other, leader])
         self.assertTrue(self.p.is_healthiest_node(cluster))
         self.p.is_leader = false
         self.assertFalse(self.p.is_healthiest_node(cluster))