#!/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)