140 lines
2.6 KiB
Python
Executable File
140 lines
2.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"proxy_client",
|
|
Path(
|
|
"ocf/loxthelion/"
|
|
"managed-container.d/"
|
|
"proxy_client.py"
|
|
),
|
|
)
|
|
|
|
proxy = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(proxy)
|
|
|
|
|
|
print("===== NORMAL API MONITOR RESULT =====")
|
|
|
|
normal = {
|
|
"ha": {
|
|
"ocf_rc": 0,
|
|
"health": "healthy",
|
|
"ha_role": "unpromoted",
|
|
"state": "streaming",
|
|
"promotion_eligible": True,
|
|
},
|
|
"role": "standby",
|
|
"streaming": True,
|
|
"wal_receiver": {
|
|
"slot_name": "ha_pg_node1",
|
|
"sender_host": "192.0.2.10",
|
|
},
|
|
}
|
|
|
|
formatted = proxy.format_monitor_result(normal)
|
|
|
|
expected = (
|
|
"0|healthy|unpromoted|streaming|true|"
|
|
"standby|true|ha_pg_node1|192.0.2.10"
|
|
)
|
|
|
|
assert formatted == expected, formatted
|
|
|
|
print("api_monitor_format accepted")
|
|
|
|
|
|
print()
|
|
print("===== PODMAN DELEGATION =====")
|
|
|
|
delegated = {
|
|
"delegated": True,
|
|
"backend": "container",
|
|
"container_runtime": "podman",
|
|
}
|
|
|
|
formatted = proxy.format_monitor_result(delegated)
|
|
|
|
assert formatted == "delegate|container|podman", formatted
|
|
|
|
print("podman_delegation accepted")
|
|
|
|
|
|
print()
|
|
print("===== DOCKER DELEGATION =====")
|
|
|
|
delegated = {
|
|
"delegated": True,
|
|
"backend": "container",
|
|
"container_runtime": "docker",
|
|
}
|
|
|
|
formatted = proxy.format_monitor_result(delegated)
|
|
|
|
assert formatted == "delegate|container|docker", formatted
|
|
|
|
print("docker_delegation accepted")
|
|
|
|
|
|
print()
|
|
print("===== INVALID DELEGATED BACKEND =====")
|
|
|
|
bad = {
|
|
"delegated": True,
|
|
"backend": "api",
|
|
"container_runtime": "podman",
|
|
}
|
|
|
|
try:
|
|
proxy.format_monitor_result(bad)
|
|
except proxy.ProtocolFailure:
|
|
print("invalid_backend rejected")
|
|
else:
|
|
raise AssertionError(
|
|
"delegation with non-container backend was accepted"
|
|
)
|
|
|
|
|
|
print()
|
|
print("===== INVALID DELEGATED RUNTIME =====")
|
|
|
|
bad = {
|
|
"delegated": True,
|
|
"backend": "container",
|
|
"container_runtime": "lxc",
|
|
}
|
|
|
|
try:
|
|
proxy.format_monitor_result(bad)
|
|
except proxy.ProtocolFailure:
|
|
print("invalid_runtime rejected")
|
|
else:
|
|
raise AssertionError(
|
|
"delegation with invalid runtime was accepted"
|
|
)
|
|
|
|
|
|
print()
|
|
print("===== MALFORMED DELEGATION =====")
|
|
|
|
bad = {
|
|
"delegated": True,
|
|
"backend": "container",
|
|
}
|
|
|
|
try:
|
|
proxy.format_monitor_result(bad)
|
|
except proxy.ProtocolFailure:
|
|
print("missing_runtime rejected")
|
|
else:
|
|
raise AssertionError(
|
|
"delegation without runtime was accepted"
|
|
)
|
|
|
|
|
|
print()
|
|
print("PROXY_MONITOR_DELEGATION_TEST_PASS=1")
|