# -*- coding: utf-8 -*- import os import shutil import unittest from pgbouncemgr.manager import * CONFIG = os.path.join( os.path.dirname(os.path.realpath(__file__)), "testfiles", "basic.yaml") CONFIG_WITH_STATE_FILE = os.path.join( os.path.dirname(os.path.realpath(__file__)), "testfiles", "basic_with_state_file.yaml") STATE_FILE_TEMPLATE = os.path.join( os.path.dirname(os.path.realpath(__file__)), "testfiles", "state.json") STATE_FILE = "/tmp/pgbouncemgr.state" class ManagerTests(unittest.TestCase): def test_CreateManagerUsingCommandLineOptions(self): """Check if a Manager object can be created, making use of some command line arguments.""" mgr = Manager([ '-f', 'none', # no syslog '-d', # debug (and verbose) enabled '--config', CONFIG, # the config file to use '-f', 'none' # the syslog facility to use, 'none' disables syslog ]) # Check the logging setup. self.assertEqual(1, len(mgr.log)) console = mgr.log[0] self.assertEqual('ConsoleLog', type(console).__name__) self.assertTrue(console.verbose_enabled) self.assertTrue(console.debug_enabled) def test_GivenNoStateFile_WhenCreatingManager_DefaultsAreUsed(self): mgr = Manager(['--config', CONFIG]) # Check if the config file was read. # If yes, then the state should be filled with data for the # the configured nodes and all state should be in the initial # configuration. self.maxDiff = 5000 self.assertEqual({ 'system_id': None, 'timeline_id': None, 'active_pgbouncer_config': None, 'leader_error': None, 'leader_node_id': None, 'leader_status': 'LEADER_UNKNOWN', 'nodes': { 'nodeA': { 'config': { 'host': '1.2.3.4', 'port': 8888, 'pgbouncer_config': None }, 'node_id': 'nodeA', 'is_leader': False, 'system_id': None, 'timeline_id': None, 'status': 'NODE_UNKNOWN', 'error': None }, 'nodeB': { 'config': { 'host': '2.3.4.5', 'port': 7777, 'pgbouncer_config': None }, 'node_id': 'nodeB', 'is_leader': False, 'system_id': None, 'timeline_id': None, 'status': 'NODE_UNKNOWN', 'error': None } } }, mgr.state.export()) def test_GivenStateFile_WhenCreatingManager_StoredStateIsApplied(self): mgr = Manager(['--config', CONFIG_WITH_STATE_FILE]) # Create the state file as expected by the loaded config. shutil.copy(STATE_FILE_TEMPLATE, STATE_FILE) # Check if the config file was read. # If yes, then the state should be filled with data for the # the configured nodes and all state should be in the initial # configuration. self.maxDiff = 5000 self.assertEqual({ 'system_id': 'A', 'timeline_id': 42, 'active_pgbouncer_config': None, # because /myt/config.ini from the state does not exit 'leader_error': None, 'leader_node_id': 'nodeA', 'leader_status': 'LEADER_UNKNOWN', 'nodes': { 'nodeA': { 'config': { 'host': '1.2.3.4', 'port': 8888, 'pgbouncer_config': None }, 'node_id': 'nodeA', 'is_leader': True, 'system_id': None, 'timeline_id': None, 'status': 'NODE_UNKNOWN', 'error': None }, 'nodeB': { 'config': { 'host': '2.3.4.5', 'port': 7777, 'pgbouncer_config': None }, 'node_id': 'nodeB', 'is_leader': False, 'system_id': None, 'timeline_id': None, 'status': 'NODE_UNKNOWN', 'error': None } } }, mgr.state.export())