68 lines
2.5 KiB
Python
68 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import json
|
|
import unittest
|
|
from tempfile import NamedTemporaryFile
|
|
from pgbouncemgr.logger import *
|
|
from pgbouncemgr.state import *
|
|
from pgbouncemgr.state_store import *
|
|
|
|
|
|
def make_test_file_path(filename):
|
|
testfiles_path = os.path.dirname(os.path.realpath(__file__))
|
|
return os.path.join(testfiles_path, "testfiles", filename)
|
|
|
|
class StateStoreTests(unittest.TestCase):
|
|
def test_GivenNonExistingStateFile_OnLoad_StateStoreDoesNotLoadState(self):
|
|
state = State(Logger())
|
|
before = state.export()
|
|
StateStore("/non-existent/state.json", state)
|
|
after = state.export()
|
|
|
|
self.assertEqual(before, after)
|
|
|
|
def test_GivenExistingStateFile_ContainingInvalidJson_OnLoad_StateStoreDoesNotLoadState(self):
|
|
state = State(Logger())
|
|
before = state.export()
|
|
StateStore(make_test_file_path("invalid.json"), state)
|
|
after = state.export()
|
|
|
|
self.assertEqual(before, after)
|
|
|
|
def test_GivenExistingStateFile_OnLoad_StateStoreUpdatesState(self):
|
|
state = State(Logger())
|
|
|
|
# These are the fields as defined in the state.json file.
|
|
expected = state.export()
|
|
expected["system_id"] = "A"
|
|
expected["timeline_id"] = 42
|
|
expected["leader_node_id"] = None # because leader node id 'nodeA' does not exist.
|
|
expected["active_pgbouncer_config"] = None # because stored /my/config.ini does not exist.
|
|
|
|
StateStore(make_test_file_path("state.json"), state)
|
|
|
|
self.assertEqual(expected, state.export())
|
|
|
|
def test_GivenState_OnSave_StateStoreStoresState(self):
|
|
try:
|
|
state = State(Logger())
|
|
tmpfile = NamedTemporaryFile(delete=False)
|
|
StateStore(tmpfile.name, state).store()
|
|
|
|
with open(tmpfile.name, 'r') as stream:
|
|
stored = json.load(stream)
|
|
self.assertEqual(state.export(), stored)
|
|
except Exception as exception:
|
|
self.fail("Unexpected exception: %s" % str(exception))
|
|
finally:
|
|
if tmpfile and os.path.exists(tmpfile.name):
|
|
os.unlink(tmpfile.name)
|
|
|
|
def test_GivenError_OnSave_ExceptionIsRaised(self):
|
|
state = State(Logger())
|
|
with self.assertRaises(StateStoreException) as context:
|
|
StateStore("/tmp/path/that/does/not/exist/fofr/statefile", state).store()
|
|
self.assertIn("Storing state to file", str(context.exception))
|
|
self.assertIn("failed: FileNotFoundError", str(context.exception))
|