181 lines
7.0 KiB
Python
181 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Isolated tests of functions extracted from deactivate-srs.sh.
|
|
|
|
No mail services are started/stopped and no files outside TemporaryDirectory
|
|
are modified. Commands such as systemctl and postfix are shell mocks.
|
|
|
|
Run from the repository:
|
|
python3 tests/test_deactivation_rollback_mock.py
|
|
"""
|
|
|
|
import argparse
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1] / "deactivate-srs.sh"
|
|
FUNCTIONS = (
|
|
"rollback_deactivation",
|
|
"wait_for_srs_forward_drain",
|
|
)
|
|
|
|
|
|
def extract_functions(source):
|
|
result = {}
|
|
for name in FUNCTIONS:
|
|
match = re.search(rf"^{re.escape(name)}\(\) \{{\n.*?^\}}", source, re.M | re.S)
|
|
if not match:
|
|
raise ValueError(f"Function {name}() not found in supplied deactivate-srs.sh")
|
|
result[name] = match.group(0)
|
|
return result
|
|
|
|
|
|
class IsolatedShellTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.functions = extract_functions(SOURCE.read_text(encoding="utf-8"))
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="srs-rollback-mock-")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.base = Path(self.tmp.name)
|
|
self.backup = self.base / "backup"
|
|
self.backup.mkdir()
|
|
(self.backup / "main.cf").write_text("original main\n")
|
|
(self.backup / "master.cf").write_text("original master\n")
|
|
self.main = self.base / "main.cf"
|
|
self.master = self.base / "master.cf"
|
|
self.main.write_text("modified main\n")
|
|
self.master.write_text("modified master\n")
|
|
self.calls = self.base / "calls"
|
|
self.log = self.base / "log"
|
|
|
|
def run_bash(self, *, function, setup="", fail="", run_args=""):
|
|
self.assertIn(function, self.functions)
|
|
script = f'''#!/bin/bash
|
|
set -u
|
|
POSTFIX_MAIN_CF={str(self.main)!r}
|
|
POSTFIX_MASTER_CF={str(self.master)!r}
|
|
DEACTIVATION_BACKUP_DIR={str(self.backup)!r}
|
|
POSTFIX_SERVICE=postfix.service
|
|
log_file={str(self.log)!r}
|
|
CALLS={str(self.calls)!r}
|
|
DEACTIVATION_CHANGES_STARTED=true
|
|
blank_line() {{ :; }}
|
|
warn() {{ :; }}
|
|
error() {{ :; }}
|
|
info() {{ :; }}
|
|
echononl() {{ :; }}
|
|
echo_ok() {{ :; }}
|
|
echo_failed() {{ :; }}
|
|
cp() {{
|
|
printf 'cp %s\\n' "${{@: -1}}" >> "$CALLS"
|
|
if [[ "${{FAIL_CP_MAIN:-0}}" == 1 && "${{@: -1}}" == "$POSTFIX_MAIN_CF" ]]; then return 1; fi
|
|
command cp "$@"
|
|
}}
|
|
restore_systemd_dropins() {{
|
|
echo restore-dropins >> "$CALLS"
|
|
[[ "${{FAIL_DROPINS:-0}}" != 1 ]]
|
|
}}
|
|
restore_service_states() {{
|
|
echo restore-services >> "$CALLS"
|
|
[[ "${{FAIL_SERVICES:-0}}" != 1 ]]
|
|
}}
|
|
postfix() {{
|
|
printf 'postfix %s\\n' "$*" >> "$CALLS"
|
|
[[ "${{FAIL_POSTFIX_CHECK:-0}}" != 1 ]]
|
|
}}
|
|
systemctl() {{
|
|
printf 'systemctl %s\\n' "$*" >> "$CALLS"
|
|
case "$1" in
|
|
is-active) [[ "${{MAIN_INACTIVE:-0}}" != 1 ]] ;;
|
|
reload) [[ "${{FAIL_RELOAD:-0}}" != 1 ]] ;;
|
|
*) return 1 ;;
|
|
esac
|
|
}}
|
|
srs_old_queue_manager_is_running() {{ return "${{QMGR_RC:-1}}"; }}
|
|
srs_forward_worker_is_running() {{ return "${{WORKER_RC:-1}}"; }}
|
|
srs_delivery_queue_is_empty() {{ return "${{QUEUE_RC:-0}}"; }}
|
|
sleep() {{ SECONDS=$((SECONDS + 181)); }}
|
|
{self.functions[function]}
|
|
{setup}
|
|
{fail}
|
|
{function} {run_args}
|
|
'''
|
|
return subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
|
timeout=10)
|
|
|
|
def test_rollback_restores_config_and_reloads(self):
|
|
result = self.run_bash(function="rollback_deactivation")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(self.main.read_text(), "original main\n")
|
|
self.assertEqual(self.master.read_text(), "original master\n")
|
|
calls = self.calls.read_text()
|
|
for expected in ("restore-dropins", "restore-services", "postfix check",
|
|
"systemctl reload postfix.service"):
|
|
self.assertIn(expected, calls)
|
|
|
|
def test_rollback_cp_error_is_reported(self):
|
|
result = self.run_bash(function="rollback_deactivation", fail="FAIL_CP_MAIN=1")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.main.read_text(), "modified main\n")
|
|
self.assertEqual(self.master.read_text(), "original master\n")
|
|
|
|
def test_rollback_dropin_error_is_reported(self):
|
|
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
|
|
fail="FAIL_DROPINS=1").returncode, 0)
|
|
|
|
def test_rollback_service_restore_error_is_reported(self):
|
|
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
|
|
fail="FAIL_SERVICES=1").returncode, 0)
|
|
|
|
def test_rollback_postfix_check_error_is_reported(self):
|
|
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
|
|
fail="FAIL_POSTFIX_CHECK=1").returncode, 0)
|
|
|
|
def test_rollback_reload_error_is_reported(self):
|
|
self.assertNotEqual(self.run_bash(function="rollback_deactivation",
|
|
fail="FAIL_RELOAD=1").returncode, 0)
|
|
|
|
def test_rollback_inactive_main_postfix_is_not_silently_successful(self):
|
|
"""Safety regression: an inactive primary Postfix must not mean success."""
|
|
result = self.run_bash(function="rollback_deactivation", fail="MAIN_INACTIVE=1")
|
|
self.assertNotEqual(result.returncode, 0,
|
|
"BUG: inactive primary Postfix incorrectly reported rollback success")
|
|
self.assertNotIn("systemctl reload postfix.service", self.calls.read_text())
|
|
|
|
def test_drain_succeeds_when_qmgr_workers_and_queue_clear(self):
|
|
result = self.run_bash(function="wait_for_srs_forward_drain",
|
|
setup="sleep() { :; }")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_drain_times_out_on_running_old_qmgr(self):
|
|
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
|
|
fail="QMGR_RC=0").returncode, 1)
|
|
|
|
def test_drain_rejects_process_check_error(self):
|
|
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
|
|
fail="WORKER_RC=2").returncode, 2)
|
|
|
|
def test_drain_rejects_queue_read_error(self):
|
|
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
|
|
fail="QUEUE_RC=2").returncode, 2)
|
|
|
|
def test_drain_times_out_if_queue_remains_nonempty(self):
|
|
self.assertEqual(self.run_bash(function="wait_for_srs_forward_drain",
|
|
fail="QUEUE_RC=1").returncode, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source", type=Path, default=SOURCE,
|
|
help="Path to deactivate-srs.sh (defaults to sibling repo script)")
|
|
args, unittest_args = parser.parse_known_args()
|
|
SOURCE = args.source.resolve()
|
|
if not SOURCE.is_file():
|
|
parser.error(f"Source script not found: {SOURCE}")
|
|
unittest.main(argv=[sys.argv[0], *unittest_args], verbosity=2)
|