diff --git a/governor.py b/governor.py index d12c7093..85cd28e0 100755 --- a/governor.py +++ b/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() diff --git a/helpers/api.py b/helpers/api.py new file mode 100644 index 00000000..c723e47b --- /dev/null +++ b/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 diff --git a/helpers/errors.py b/helpers/errors.py index dd1bc698..b86bd0f0 100644 --- a/helpers/errors.py +++ b/helpers/errors.py @@ -9,7 +9,3 @@ class EtcdError(Exception): class CurrentLeaderError(EtcdError): pass - - -class HealthiestMemberError(EtcdError): - pass diff --git a/helpers/etcd.py b/helpers/etcd.py index e808f8d4..6da6a3bc 100644 --- a/helpers/etcd.py +++ b/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) diff --git a/helpers/ha.py b/helpers/ha.py index 773c9b57..29979363 100644 --- a/helpers/ha.py +++ b/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') diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 720d0775..868e9bbf 100644 --- a/helpers/postgresql.py +++ b/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) diff --git a/postgres0.yml b/postgres0.yml index 5f4262f9..8757538f 100644 --- a/postgres0.yml +++ b/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 diff --git a/postgres1.yml b/postgres1.yml index 31a5dca3..a0afe20f 100644 --- a/postgres1.yml +++ b/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 diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 00000000..6b117b48 --- /dev/null +++ b/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) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index c358e89e..8f2280e6 100644 --- a/tests/test_etcd.py +++ b/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:rep-pass@127.0.0.1:5434/postgres","expiration":"2015-05-15T09:10:59.949384522Z","ttl":21,"modifiedIndex":20727,"createdIndex":20727},{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:rep-pass@127.0.0.1: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:rep-pass@127.0.0.1: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('')) diff --git a/tests/test_governor.py b/tests/test_governor.py index ed7ae32e..941d6d45 100644 --- a/tests/test_governor.py +++ b/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) diff --git a/tests/test_ha.py b/tests/test_ha.py index cabdf7d5..65d7bd6a 100644 --- a/tests/test_ha.py +++ b/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): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index e0a21645..dc6fb512 100644 --- a/tests/test_postgresql.py +++ b/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:rep-pass@127.0.0.1:5435/postgres', 28) me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', 28) other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1: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))