276 lines
10 KiB
Python
276 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Isolated status, preflight, snapshot and rollback tests for activate-srs.sh.
|
|
|
|
All systemctl and ss operations are mocked. No real services or Postfix
|
|
configuration are modified. Run from the repository root:
|
|
python3 tests/test_activation_status_mock.py
|
|
"""
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1] / 'activate-srs.sh'
|
|
TEXT = SOURCE.read_text(encoding='utf-8')
|
|
|
|
|
|
def function(name):
|
|
match = re.search(rf'^{re.escape(name)}\(\) \{{\n.*?^\}}', TEXT, re.M | re.S)
|
|
if not match:
|
|
raise ValueError(f'Missing function {name}')
|
|
return match.group(0)
|
|
|
|
|
|
FUNCTIONS = '\n'.join(function(n) for n in (
|
|
'master_srs_forward_present', 'port_is_listening', 'service_exists',
|
|
'read_srs_service_state', 'service_active', 'service_enabled_state',
|
|
'snapshot_service_states', 'restore_service_states', 'rollback_activation',
|
|
))
|
|
|
|
|
|
def section(start, end):
|
|
return TEXT[TEXT.index(start):TEXT.index(end, TEXT.index(start))]
|
|
|
|
|
|
PREFLIGHT = section('services_exist=true\n', '# ----------\n# Determine main.cf state')
|
|
CLASSIFICATION = section('srs_integration_enabled=false\n', '# ----------\n# Summary')
|
|
|
|
|
|
class ActivationMockTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory(prefix='srs-activation-status-')
|
|
self.addCleanup(tmp.cleanup)
|
|
self.root = Path(tmp.name)
|
|
self.master = self.root / 'master.cf'
|
|
self.master.write_text('smtp inet n - y - - smtpd\n')
|
|
self.calls = self.root / 'calls'
|
|
self.log = self.root / 'log'
|
|
self.drop1 = self.root / 'postfix.srs.conf'
|
|
self.drop2 = self.root / 'delivery.srs.conf'
|
|
|
|
def run_shell(self, commands, *, extra='', include_funcs=True):
|
|
sh = f'''#!/bin/bash
|
|
set -u
|
|
POSTFIX_MASTER_CF={str(self.master)!r}
|
|
POSTFIX_SRS_DROPIN_FILE={str(self.drop1)!r}
|
|
SRS_DELIVERY_DROPIN_FILE={str(self.drop2)!r}
|
|
POSTSRSD_SERVICE=postsrsd-srs.service
|
|
SRS_REVERSE_MAP_SERVICE=srs-reverse-map.service
|
|
SRS_DELIVERY_SERVICE=postfix@srs-delivery.service
|
|
POSTFIX_SERVICE=postfix.service
|
|
POSTSRSD_SOCKET_HOST=127.0.0.1
|
|
POSTSRSD_SOCKET_PORT=10003
|
|
SRS_REVERSE_SOCKET_HOST=127.0.0.1
|
|
SRS_REVERSE_SOCKET_PORT=10004
|
|
SRS_REINJECT_HOST=127.0.0.1
|
|
SRS_REINJECT_PORT=10028
|
|
SRS_REVERSE_HOST=127.0.0.1
|
|
SRS_REVERSE_PORT=10029
|
|
CALLS={str(self.calls)!r}
|
|
log_file={str(self.log)!r}
|
|
ACTIVATION_CHANGES_STARTED=true
|
|
ACTIVATION_BACKUP_DIR={str(self.root)!r}
|
|
declare -A SERVICE_WAS_ACTIVE=() SERVICE_WAS_ENABLED=()
|
|
blank_line() {{ :; }}
|
|
warn() {{ :; }}
|
|
error() {{ :; }}
|
|
info() {{ :; }}
|
|
echononl() {{ :; }}
|
|
echo_ok() {{ :; }}
|
|
echo_failed() {{ :; }}
|
|
postfix() {{ return 0; }}
|
|
restore_systemd_dropins() {{ return 0; }}
|
|
systemctl() {{
|
|
printf 'systemctl %s\\n' "$*" >> "$CALLS"
|
|
case "$1" in
|
|
cat) return 0 ;;
|
|
show)
|
|
[[ "${{SHOW_FAIL_UNIT:-}}" == "$2" ]] && return 4
|
|
if [[ "${{BAD_STATE_UNIT:-}}" == "$2" ]]; then
|
|
printf 'LoadState=loaded\\nActiveState=failed\\nUnitFileState=enabled\\n'
|
|
elif [[ "${{ACTIVE_UNIT:-}}" == "$2" ]]; then
|
|
printf 'LoadState=loaded\\nActiveState=active\\nUnitFileState=enabled\\n'
|
|
elif [[ "${{ENABLED_UNIT:-}}" == "$2" ]]; then
|
|
printf 'LoadState=loaded\\nActiveState=inactive\\nUnitFileState=enabled\\n'
|
|
else
|
|
printf '%s\\n' "${{SHOW_DATA:-$'LoadState=loaded\\nActiveState=inactive\\nUnitFileState=disabled'}}"
|
|
fi ;;
|
|
is-active) [[ "${{MAIN_INACTIVE:-0}}" != 1 ]] ;;
|
|
enable|disable|stop|start|reload)
|
|
[[ "${{FAIL_ACTION:-}}" != "$1" ]] ;;
|
|
*) return 88 ;;
|
|
esac
|
|
}}
|
|
ss() {{
|
|
[[ "${{SS_FAIL:-0}}" == 1 ]] && return 7
|
|
printf '%s\\n' "${{SS_DATA:-}}"
|
|
}}
|
|
{FUNCTIONS if include_funcs else ''}
|
|
{extra}
|
|
{commands}
|
|
'''
|
|
return subprocess.run(['bash','-c',sh],capture_output=True,text=True,timeout=10)
|
|
|
|
def check(self, commands, extra=''):
|
|
result = self.run_shell(commands, extra=extra)
|
|
self.assertEqual(result.returncode, 0, f'{result.stdout}\n{result.stderr}')
|
|
return result
|
|
|
|
def test_valid_disabled_preflight(self):
|
|
self.check(PREFLIGHT + '''
|
|
[[ "$services_disabled" == true && "$listeners_absent" == true && "$dropins_absent" == true && "$services_status_known" == true && "$primary_postfix_active" == true ]]
|
|
''')
|
|
|
|
def test_active_srs_service_is_not_disabled(self):
|
|
self.check(PREFLIGHT + '[[ "$services_disabled" == false ]]',
|
|
extra='ACTIVE_UNIT=postsrsd-srs.service')
|
|
|
|
def test_enabled_but_inactive_srs_service_is_not_disabled(self):
|
|
self.check(PREFLIGHT + '[[ "$services_disabled" == false ]]',
|
|
extra='ENABLED_UNIT=postsrsd-srs.service')
|
|
|
|
def test_systemctl_query_error_marks_unknown(self):
|
|
self.check(PREFLIGHT + '[[ "$services_status_known" == false && "$services_disabled" == false ]]',
|
|
extra='SHOW_FAIL_UNIT=srs-reverse-map.service')
|
|
|
|
def test_transient_systemctl_state_marks_unknown(self):
|
|
self.check(PREFLIGHT + '[[ "$services_status_known" == false ]]',
|
|
extra='BAD_STATE_UNIT=postsrsd-srs.service')
|
|
|
|
def test_active_port_not_disabled(self):
|
|
self.check(PREFLIGHT + '[[ "$listeners_absent" == false ]]',
|
|
extra="SS_DATA='LISTEN 0 5 127.0.0.1:10003 0.0.0.0:*'")
|
|
|
|
def test_ss_failure_marks_unknown(self):
|
|
self.check(PREFLIGHT + '[[ "$listeners_status_known" == false && "$listeners_absent" == false ]]',
|
|
extra='SS_FAIL=1')
|
|
|
|
def test_main_postfix_inactive_is_not_ready(self):
|
|
self.check(PREFLIGHT + '[[ "$primary_postfix_active" == false ]]',
|
|
extra='MAIN_INACTIVE=1')
|
|
|
|
def test_stray_dropin_not_disabled(self):
|
|
self.drop1.write_text('[Unit]\n')
|
|
self.check(PREFLIGHT + '[[ "$dropins_absent" == false ]]')
|
|
|
|
def test_broken_symlink_dropin_not_disabled(self):
|
|
self.drop1.symlink_to(self.root/'not-existing')
|
|
self.check(PREFLIGHT + '[[ "$dropins_absent" == false ]]')
|
|
|
|
def test_master_cf_unreadable_is_query_error(self):
|
|
self.master.unlink()
|
|
self.check('master_srs_forward_present; [[ $? -eq 2 ]]')
|
|
|
|
def test_master_cf_absent_and_present(self):
|
|
self.check('master_srs_forward_present; [[ $? -eq 1 ]]')
|
|
self.master.write_text('srs-forward unix - n n - - pipe\n argv=/usr/local/sbin/srs-forward.py ${sender}\n')
|
|
self.check('master_srs_forward_present; [[ $? -eq 0 ]]')
|
|
|
|
def test_atomic_snapshot_disabled(self):
|
|
self.check('''snapshot_service_states || exit 1
|
|
for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
|
|
[[ "${SERVICE_WAS_ACTIVE[$unit]}" == false ]] || exit 2
|
|
[[ "${SERVICE_WAS_ENABLED[$unit]}" == disabled ]] || exit 3
|
|
done''')
|
|
self.check('''SHOW_FAIL_UNIT=srs-reverse-map.service
|
|
snapshot_service_states; [[ $? -eq 1 ]] || exit 1
|
|
[[ "${#SERVICE_WAS_ACTIVE[@]}" -eq 0 ]] || exit 2''')
|
|
|
|
def test_snapshot_rejects_changed_state(self):
|
|
self.check('''ACTIVE_UNIT=postsrsd-srs.service
|
|
snapshot_service_states; [[ $? -eq 1 ]] || exit 1
|
|
[[ "${#SERVICE_WAS_ENABLED[@]}" -eq 0 ]] || exit 2''')
|
|
|
|
def test_restore_rejects_invalid_snapshot_without_actions(self):
|
|
self.check('''for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
|
|
SERVICE_WAS_ACTIVE[$unit]=false
|
|
SERVICE_WAS_ENABLED[$unit]=disabled
|
|
done
|
|
SERVICE_WAS_ENABLED[$SRS_REVERSE_MAP_SERVICE]=masked
|
|
restore_service_states; [[ $? -eq 1 ]]''')
|
|
self.assertFalse(self.calls.exists() and any(
|
|
part in self.calls.read_text() for part in ('systemctl start ', 'systemctl stop ', 'systemctl enable ', 'systemctl disable ')
|
|
))
|
|
|
|
def test_restore_respects_enabled_runtime(self):
|
|
self.check('''for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
|
|
SERVICE_WAS_ACTIVE[$unit]=false
|
|
SERVICE_WAS_ENABLED[$unit]=disabled
|
|
done
|
|
SERVICE_WAS_ACTIVE[$POSTSRSD_SERVICE]=true
|
|
SERVICE_WAS_ENABLED[$POSTSRSD_SERVICE]=enabled-runtime
|
|
restore_service_states''')
|
|
self.assertIn('systemctl enable --runtime postsrsd-srs.service',self.calls.read_text())
|
|
|
|
def test_rollback_inactive_primary_postfix_is_failure(self):
|
|
(self.root/'main.cf').write_text('before\n')
|
|
(self.root/'master.cf').write_text('before\n')
|
|
self.check('''POSTFIX_MAIN_CF=${ACTIVATION_BACKUP_DIR}/working-main.cf
|
|
POSTFIX_MASTER_CF=${ACTIVATION_BACKUP_DIR}/working-master.cf
|
|
for unit in "$POSTSRSD_SERVICE" "$SRS_REVERSE_MAP_SERVICE" "$SRS_DELIVERY_SERVICE"; do
|
|
SERVICE_WAS_ACTIVE[$unit]=false
|
|
SERVICE_WAS_ENABLED[$unit]=disabled
|
|
done
|
|
MAIN_INACTIVE=1
|
|
rollback_activation; [[ $? -eq 1 ]]''')
|
|
self.assertNotIn('systemctl reload postfix.service',self.calls.read_text())
|
|
|
|
def test_enabled_classification_still_works(self):
|
|
baseline = '''main_block_status=present
|
|
master_block_status=present
|
|
master_srs_forward_known=true
|
|
main_srs_values_ok=true
|
|
master_srs_forward_ok=true
|
|
smtp_override_ok=true
|
|
amavis_override_ok=true
|
|
pre_srs_main_ok=false
|
|
pre_srs_master_ok=false
|
|
installation_ok=true
|
|
services_exist=true
|
|
services_status_known=true
|
|
services_disabled=false
|
|
services_enabled=true
|
|
services_active=true
|
|
primary_postfix_active=true
|
|
listeners_status_known=true
|
|
listeners_absent=false
|
|
listeners_ok=true
|
|
dropins_absent=false
|
|
'''
|
|
self.check(CLASSIFICATION + '[[ "$srs_state" == enabled ]]', extra=baseline)
|
|
|
|
def test_disabled_classification_requires_every_prerequisite(self):
|
|
baseline = '''main_block_status=absent
|
|
master_block_status=absent
|
|
master_srs_forward_known=true
|
|
main_srs_values_ok=false
|
|
master_srs_forward_ok=false
|
|
smtp_override_ok=false
|
|
amavis_override_ok=false
|
|
pre_srs_main_ok=true
|
|
pre_srs_master_ok=true
|
|
installation_ok=true
|
|
services_exist=true
|
|
services_status_known=true
|
|
services_disabled=true
|
|
services_enabled=false
|
|
services_active=false
|
|
primary_postfix_active=true
|
|
listeners_status_known=true
|
|
listeners_absent=true
|
|
listeners_ok=false
|
|
dropins_absent=true
|
|
'''
|
|
self.check(CLASSIFICATION+'[[ "$srs_state" == disabled ]]',extra=baseline)
|
|
for name in ('services_status_known','services_disabled','primary_postfix_active',
|
|
'listeners_status_known','listeners_absent','dropins_absent',
|
|
'master_srs_forward_known'):
|
|
with self.subTest(name=name):
|
|
self.check(CLASSIFICATION+'[[ "$srs_state" == inconsistent ]]',
|
|
extra=baseline+f'{name}=false\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|