managed-containerd/tests/test-promote-database.py

#!/usr/bin/env python3

import importlib.util
import sys
import time
import types
from pathlib import Path


fake_psycopg2 = types.ModuleType("psycopg2")
fake_psycopg2.Error = Exception
fake_psycopg2.OperationalError = OSError

fake_extras = types.ModuleType("psycopg2.extras")
fake_extras.RealDictCursor = object
fake_psycopg2.extras = fake_extras

sys.modules["psycopg2"] = fake_psycopg2
sys.modules["psycopg2.extras"] = fake_extras


spec = importlib.util.spec_from_file_location(
    "postgresql_plugin",
    Path("plugins/postgresql/plugin.py"),
)

plugin = importlib.util.module_from_spec(spec)
spec.loader.exec_module(plugin)


def standby_monitor(*, eligible=True):
    return {
        "ha": {
            "ocf_rc": plugin.OCF_SUCCESS,
            "health": "healthy",
            "ha_role": "unpromoted",
            "state": "streaming",
            "promotion_eligible": eligible,
        },
        "role": "standby",
        "in_recovery": True,
        "streaming": True,
    }


def peer_state(*primaries):
    return {
        "peers": [],
        "writable_primaries": list(primaries),
        "writable_primary_count": len(primaries),
    }


config = {
    "PACEMAKER_NODE_NAME": "david-eve",
    "PGPASSWORD_FILE": "/run/secrets/postgres_password",
    "PG_REPLICATION_PEERS": (
        "david-prodesk=192.168.1.4:5432,"
        "david-macbookpro=192.168.1.5:5432,"
        "david-eve=192.168.1.3:5432"
    ),
    "PGUSER": "david",
    "PGDATABASE": "david",
}


def fresh_state():
    return {
        "postgresql": {
            "notification": {
                "type": "pre",
                "operation": "promote",
                "promote_uname": "david-eve",
                "demote_uname": "david-prodesk",
                "received_at": time.time(),
            },
        },
    }


plugin.read_secret = lambda path: "test-password"
plugin.monitor = lambda config, runtime_state: standby_monitor()
plugin.probe_replication_peers = (
    lambda config, password: peer_state()
)


print("===== PROMOTE DATABASE ACTION =====")

runtime_state = fresh_state()

result = plugin.promote_database(
    config,
    runtime_state,
    {},
)

assert result["ocf_rc"] == plugin.OCF_SUCCESS
assert result["ready"] is True
assert result["mutation"] is False
assert result["reason"] == "promotion_preflight_ready"
assert result["target_node"] == "david-eve"
assert result["previous_primary_node"] == "david-prodesk"
assert result["writable_primary_count"] == 0
assert result["checks"]["local_target"] is True
assert result["checks"]["local_standby"] is True
assert result["checks"]["promotion_eligible"] is True
assert result["checks"]["no_reachable_writable_primary"] is True

print("valid_promotion_preflight accepted")


runtime_state = fresh_state()
runtime_state["postgresql"]["notification"]["promote_uname"] = (
    "david-macbookpro"
)

result = plugin.promote_database(
    config,
    runtime_state,
    {},
)

assert result["ocf_rc"] == 1
assert result["ready"] is False
assert result["reason"] == "local_node_not_target"

print("wrong_target              rejected")


runtime_state = fresh_state()
runtime_state["postgresql"]["notification"]["received_at"] = (
    time.time() - 121
)

result = plugin.promote_database(
    config,
    runtime_state,
    {},
)

assert result["ocf_rc"] == 1
assert result["reason"] == "notification_stale"

print("stale_notification        rejected")


result = plugin.promote_database(
    config,
    {},
    {},
)

assert result["ocf_rc"] == 1
assert result["reason"] == "notification_missing"

print("missing_notification      rejected")


runtime_state = fresh_state()

plugin.monitor = lambda config, runtime_state: {
    "ha": {
        "ocf_rc": plugin.OCF_RUNNING_PROMOTED,
        "promotion_eligible": True,
    },
    "role": "primary",
    "in_recovery": False,
}

result = plugin.promote_database(
    config,
    runtime_state,
    {},
)

assert result["ocf_rc"] == 1
assert result["ready"] is False
assert result["reason"] == "local_not_promotion_eligible"
assert result["checks"]["local_standby"] is False

print("local_primary             rejected")


plugin.monitor = lambda config, runtime_state: standby_monitor(
    eligible=False
)

result = plugin.promote_database(
    config,
    fresh_state(),
    {},
)

assert result["ocf_rc"] == 1
assert result["ready"] is False
assert result["checks"]["promotion_eligible"] is False

print("ineligible_standby        rejected")


plugin.monitor = lambda config, runtime_state: standby_monitor()
plugin.probe_replication_peers = (
    lambda config, password: peer_state("david-prodesk")
)

result = plugin.promote_database(
    config,
    fresh_state(),
    {},
)

assert result["ocf_rc"] == plugin.OCF_ERR_GENERIC
assert result["ready"] is False
assert result["mutation"] is False
assert result["reason"] == "reachable_writable_primary"
assert result["writable_primary_count"] == 1
assert result["writable_primaries"] == ["david-prodesk"]
assert result["checks"]["no_reachable_writable_primary"] is False

print("existing_primary          rejected")


print()
print("PROMOTE_DATABASE_TEST_PASS=1")