managed-containerd/tests/test-monitor-backends.py

#!/usr/bin/env python3

import importlib.util
import json
import subprocess
import sys
import types
from pathlib import Path


# Match the existing daemon unit-test convention: psycopg2 is supplied by the
# production container image, while host-side tests use a minimal fake module.
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 base_config():
    return {
        "PGHOST": "127.0.0.1",
        "PGPORT": "5432",
        "PGUSER": "david",
        "PGDATABASE": "david",
        "PGPASSWORD_FILE": "/run/secrets/postgres_password",
        "PG_REPLICATION_PEERS": (
            "node1=192.0.2.1:5432,"
            "node2=192.0.2.2:5432"
        ),
        "PACEMAKER_NODE_NAME": "node1",
    }


print("===== DEFAULT BACKEND =====")

cfg = base_config()

assert plugin.monitor_backend(cfg) == "api"
plugin.validate_config(cfg)

print("default_api             accepted")


print()
print("===== EXPLICIT API =====")

cfg = base_config()
cfg["MONITOR_BACKEND"] = "api"

plugin.validate_config(cfg)
assert plugin.monitor_backend(cfg) == "api"

print("explicit_api            accepted")


print()
print("===== API DISPATCH =====")

cfg = base_config()
cfg["MONITOR_BACKEND"] = "api"

original_api = plugin.monitor_api
original_container = plugin.monitor_container

try:
    plugin.monitor_api = (
        lambda config, runtime_state: {
            "ha": {
                "ocf_rc": plugin.OCF_SUCCESS
            },
            "role": "standby",
        }
    )

    def must_not_run(*args, **kwargs):
        raise AssertionError(
            "container monitor executed for API backend"
        )

    plugin.monitor_container = must_not_run

    result = plugin.monitor(cfg, {})

    assert result["ha"]["ocf_rc"] == plugin.OCF_SUCCESS
    assert result["monitor_backend"] == "api"

finally:
    plugin.monitor_api = original_api
    plugin.monitor_container = original_container

print("api_dispatch            accepted")


print()
print("===== CONTAINER VALIDATION =====")

cfg = base_config()
cfg["MONITOR_BACKEND"] = "container"
cfg["CONTAINER_RUNTIME"] = "podman"
cfg["CONTAINER_NAME"] = "postgres"
cfg["PG_OS_USER"] = "postgres"

plugin.validate_config(cfg)

assert plugin.monitor_backend(cfg) == "container"
assert plugin.container_runtime(cfg) == "podman"

print("container_podman        accepted")


print()
print("===== DOCKER VALIDATION =====")

cfg["CONTAINER_RUNTIME"] = "docker"

plugin.validate_config(cfg)
assert plugin.container_runtime(cfg) == "docker"

print("container_docker        accepted")


print()
print("===== INVALID BACKEND =====")

bad = base_config()
bad["MONITOR_BACKEND"] = "invalid"

try:
    plugin.validate_config(bad)
except ValueError:
    print("invalid_backend         rejected")
else:
    raise AssertionError(
        "invalid backend was accepted"
    )


print()
print("===== MISSING RUNTIME =====")

bad = base_config()
bad["MONITOR_BACKEND"] = "container"
bad["CONTAINER_NAME"] = "postgres"

try:
    plugin.validate_config(bad)
except ValueError:
    print("missing_runtime         rejected")
else:
    raise AssertionError(
        "container backend without runtime was accepted"
    )


print()
print("===== INVALID RUNTIME =====")

bad["CONTAINER_RUNTIME"] = "lxc"

try:
    plugin.validate_config(bad)
except ValueError:
    print("invalid_runtime         rejected")
else:
    raise AssertionError(
        "invalid runtime was accepted"
    )


print()
print("===== DELEGATION WITHOUT CONTAINER IDENTITY =====")

cfg = base_config()
cfg["MONITOR_BACKEND"] = "container"
cfg["CONTAINER_RUNTIME"] = "podman"

plugin.validate_config(cfg)

print("delegation_config       accepted")


print()
print("===== DIRECT EXECUTOR STILL REQUIRES CONTAINER =====")

try:
    plugin.container_monitor_config(cfg)
except ValueError:
    print("direct_missing_container rejected")
else:
    raise AssertionError(
        "direct executor accepted missing container name"
    )


print()
print("===== CONTAINER DISPATCH DELEGATION =====")

original_container = plugin.monitor_container

try:
    def must_not_execute(*args, **kwargs):
        raise AssertionError(
            "direct container monitor executed during delegation"
        )

    plugin.monitor_container = must_not_execute

    result = plugin.monitor(cfg, {})

    assert result == {
        "delegated": True,
        "backend": "container",
        "container_runtime": "podman",
    }, result

finally:
    plugin.monitor_container = original_container

print("container_delegation    accepted")


print()
print("MONITOR_BACKEND_TEST_PASS=1")


print()
print("===== CONTAINER RUNTIME EXECUTION =====")

original_run = plugin.subprocess.run
calls = []

try:
    def fake_run(argv, **kwargs):
        calls.append((argv, kwargs))

        class Result:
            stdout = ""
            stderr = ""
            returncode = 0

        return Result()

    plugin.subprocess.run = fake_run

    podman_config = base_config()
    podman_config.update({
        "MONITOR_BACKEND": "container",
        "CONTAINER_RUNTIME": "podman",
        "CONTAINER_NAME": "postgres",
        "PG_OS_USER": "postgres",
    })

    plugin.container_exec(
        podman_config,
        "pg_isready",
        "-q",
    )

    assert calls[-1][0] == [
        "podman",
        "exec",
        "--user",
        "postgres",
        "postgres",
        "pg_isready",
        "-q",
    ], calls[-1][0]

    print("podman_exec_selection    accepted")

    docker_config = base_config()
    docker_config.update({
        "MONITOR_BACKEND": "container",
        "CONTAINER_RUNTIME": "docker",
        "CONTAINER_NAME": "postgres",
        "PG_OS_USER": "postgres",
    })

    plugin.container_exec(
        docker_config,
        "pg_isready",
        "-q",
    )

    assert calls[-1][0] == [
        "docker",
        "exec",
        "--user",
        "postgres",
        "postgres",
        "pg_isready",
        "-q",
    ], calls[-1][0]

    print("docker_exec_selection    accepted")

finally:
    plugin.subprocess.run = original_run


print()
print("CONTAINER_RUNTIME_SELECTION_TEST_PASS=1")