#!/usr/bin/env python3 """Offline scoring/action tests. All cluster and PostgreSQL boundaries are fake.""" import json import os from pathlib import Path import shlex import shutil import xml.etree.ElementTree as ET import subprocess import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] AGENT = ROOT / 'refactored/managed-container' MODULES = ('common', 'container', 'postgres', 'auth', 'replication', 'notify', 'promote', 'demote', 'rejoin') LOAD = 'MC_PROMOTION_RESOURCE_INSTANCE="${OCF_RESOURCE_INSTANCE:-}"\n: "${OCF_ROOT:=/usr/lib/ocf}"\n. "$OCF_ROOT/lib/heartbeat/ocf-shellfuncs"\n: "${OCF_RUNNING_PROMOTED:=${OCF_RUNNING_MASTER:-8}}"\n' LOAD += '\n'.join('. ' + shlex.quote(str(AGENT.parent / 'managed-container.d' / (m + '.sh'))) for m in MODULES) CRM = '''#!/usr/bin/env python3 import json, os, sys from pathlib import Path args=sys.argv[1:] p=Path(os.environ['SCORE_DB']) d=json.loads(p.read_text()) assert '--lifetime=reboot' in args, args resource=next(x.split('=',1)[1] for x in args if x.startswith('--promotion=')) assert not any(x.startswith('--name') for x in args), args node=os.environ.get('OCF_RESKEY_CRM_meta_on_node','node-a') key=resource+'|'+node d['calls'].append(args) if '--query' in args: p.write_text(json.dumps(d)) if os.environ.get('QUERY_FAIL')=='1': sys.exit(1) if key not in d['scores']: sys.exit(6) print(d['scores'][key]); sys.exit(0) value=next(x.split('=',1)[1] for x in args if x.startswith('--update=')) if os.environ.get('UPDATE_FAIL')=='1': p.write_text(json.dumps(d)); sys.exit(1) d['scores'][key]=value d['writes'].append([key,value]) p.write_text(json.dumps(d)) ''' # Original action implementations execute against these fake external boundaries. FIXTURE = r''' ROLE=t RUNNING=1 SERVICE=1 CONTAINER=1 ELIGIBLE=1 SIGNAL=1 ocf_log() { :; } ocf_exit_reason() { printf '%s\n' "$*" >&2; } service_running() { [ "$SERVICE" = 1 ]; } container_running() { [ "$CONTAINER" = 1 ]; } pg_running() { [ "$RUNNING" = 1 ]; } pg_ready() { return 0; } pg_role() { echo "$ROLE"; } pg_wal_receiver_status() { echo streaming; } mc_pg_promotion_eligible() { [ "$ELIGIBLE" = 1 ]; } mc_pg_standby_score() { mc_pg_promotion_eligible && echo 100; } validate_all() { return 0; } ensure_local_ha_slots() { return 0; } ensure_rewind_privileges() { return 0; } demote_target() { echo node-b; } node_to_addr() { echo 192.0.2.2; } candidate_is_standby() { return 0; } candidate_caught_up() { return 0; } ensure_remote_slot() { return 0; } pgdata_path() { echo /test/pgdata; } pg_remote_query() { echo f; } pg_replication_exec() { return 0; } pg_command() { case "$*" in *'test -f'*) [ "$SIGNAL" = 1 ] ;; *'promote -w'*) ROLE=f ;; *'stop -m fast -w'*) RUNNING=0 ;; *'start -w'*) RUNNING=1 ;; *) echo "Unexpected pg_command: $*" >&2; return 99 ;; esac } as_service_user() { case "$*" in 'systemctl --user stop '*) SERVICE=0; CONTAINER=0 ;; 'systemctl --user start '*) SERVICE=1; CONTAINER=1 ;; *RA_PRIMARY_HOST=*) ROLE=t ;; *) echo "Unexpected external access: $*" >&2; return 99 ;; esac } sleep() { :; } ''' class Scores(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix='mc-score-test-') self.addCleanup(self.tmp.cleanup) self.path = Path(self.tmp.name) self.db = self.path / 'scores.json' self.db.write_text(json.dumps({'scores': {}, 'calls': [], 'writes': []})) (self.path / 'crm_attribute').write_text(CRM) (self.path / 'crm_attribute').chmod(0o755) self.env = dict(os.environ, PATH=str(self.path) + ':' + os.environ['PATH'], SCORE_DB=str(self.db), OCF_RESOURCE_INSTANCE='pg-ha:0', OCF_RESKEY_CRM_meta_on_node='node-a') def run_shell(self, code, rc=0, fixture=FIXTURE, env=None): result = subprocess.run(['bash', '-c', LOAD + fixture + '\n' + code], env=env or self.env, text=True, capture_output=True, timeout=15) self.assertEqual(result.returncode, rc, result.stderr) return result def state(self): return json.loads(self.db.read_text()) def score(self, value, node='node-a'): self.assertEqual(self.state()['scores']['pg-ha:0|' + node], value) def test_primary_and_standby(self): self.run_shell('ROLE=f; mc_promotion_monitor', 8) self.score('1000') self.env['OCF_RESKEY_CRM_meta_on_node'] = 'node-b' self.run_shell('mc_promotion_monitor') self.score('100', 'node-b') def test_two_standbys_equal(self): for node in ('node-a', 'node-b'): self.env['OCF_RESKEY_CRM_meta_on_node'] = node self.run_shell('mc_promotion_monitor') self.score('100', node) def test_ineligible_replaces_stale_positive(self): cases = [('RUNNING=0', 7), ('CONTAINER=0', 7), ('SERVICE=0', 7), ('ELIGIBLE=0', 0), ('pg_ready() { return 1; }', 1), ('pg_role() { return 1; }', 1), ('ROLE=invalid', 1)] for setup, rc in cases: with self.subTest(setup=setup): self.run_shell('mc_promotion_score_set 1000') self.run_shell(setup + '; mc_promotion_monitor', rc) self.score('-INFINITY') def test_repeat_monitors_write_once(self): for _ in range(3): self.run_shell('mc_promotion_monitor') self.assertEqual(len(self.state()['writes']), 1) self.assertEqual(len(self.state()['calls']), 4) # Three queries, one update. def test_restart_reprobe_recreates_attribute(self): self.run_shell('mc_promotion_monitor') data = self.state(); data['scores'] = {}; self.db.write_text(json.dumps(data)) self.run_shell('mc_promotion_monitor') self.score('100') self.assertEqual(len(self.state()['writes']), 2) def test_stop_start_original_actions(self): self.run_shell('mc_promotion_monitor; mc_promotion_action stop; ' '[ "$RUNNING$SERVICE$CONTAINER" = 000 ]; ' 'mc_promotion_action start') self.score('100') self.assertEqual([x[1] for x in self.state()['writes']], ['100', '-INFINITY', '100']) def test_primary_start_refusal(self): self.run_shell('ROLE=f; mc_promotion_action start', 1) self.score('-INFINITY') def test_missing_standby_signal(self): self.run_shell('RUNNING=0; SIGNAL=0; mc_promotion_action start', 1) self.score('-INFINITY') def test_promote_original_action(self): self.run_shell('mc_promotion_monitor; mc_promotion_action promote') self.score('1000') self.assertEqual([x[1] for x in self.state()['writes']], ['100', '-INFINITY', '1000']) def test_demote_original_action(self): self.run_shell('ROLE=f; mc_promotion_monitor; mc_promotion_action demote; ' '[ "$ROLE$SERVICE$CONTAINER" = t11 ]') self.score('100') self.assertEqual([x[1] for x in self.state()['writes']], ['1000', '-INFINITY', '100']) def test_failed_demote_stays_negative_until_next_observation(self): self.run_shell('ROLE=f; demote_target() { return 1; }; mc_promotion_action demote', 1) self.score('-INFINITY') self.run_shell('ROLE=f; mc_promotion_monitor', 8) self.score('1000') def test_failed_promote_invariants_withdraw_score(self): self.run_shell('ensure_local_ha_slots() { return 1; }; mc_promotion_action promote', 1) self.score('-INFINITY') def test_validation_failure_withdraws_score(self): self.run_shell('mc_promotion_monitor; validate_all() { return 6; }; mc_promotion_action start', 6) self.score('-INFINITY') def test_stop_does_not_require_validation(self): self.run_shell('validate_all() { return 6; }; mc_promotion_action stop') self.score('-INFINITY') def test_rejoin_original_action(self): self.run_shell('RUNNING=0; REJOIN_SOURCE_HOST=192.0.2.2; REPLICATION_SLOT=ha_node; ' 'mc_promotion_action rejoin') self.score('100') def test_update_failure_preserves_primary_role_rc(self): self.env['UPDATE_FAIL'] = '1' result = self.run_shell('ocf_log() { echo "$*" >&2; }; ROLE=f; mc_promotion_monitor', 8) self.assertIn('prior score may remain', result.stderr) self.assertFalse(self.state()['scores']) def test_update_failure_does_not_block_stop(self): self.env['UPDATE_FAIL'] = '1' self.run_shell('mc_promotion_action stop; [ "$RUNNING" = 0 ]') def test_query_failure_attempts_replace(self): self.env['QUERY_FAIL'] = '1' self.run_shell('ELIGIBLE=0; mc_promotion_monitor') self.score('-INFINITY') def test_scoring_dependency_validation(self): self.run_shell('command() { return 1; }; mc_promotion_validate', 5) def test_missing_instance_never_uses_default_attribute(self): env = dict(self.env) env.pop('OCF_RESOURCE_INSTANCE') self.run_shell('mc_promotion_monitor', env=env) self.assertFalse(self.state()['calls']) def test_target_role_does_not_affect_scoring(self): for value in ('', 'Started', 'Promoted'): self.env['OCF_RESKEY_CRM_meta_target_role'] = value self.run_shell('mc_promotion_monitor') self.score('100') def test_unrelated_legacy_attributes_untouched(self): d = self.state() d['scores']['master-postgresql'] = '-INFINITY' d['scores']['postgresql-data-status'] = 'DISCONNECT' d['scores']['postgresql-status'] = 'STOP' self.db.write_text(json.dumps(d)) self.run_shell('mc_promotion_monitor') for key, value in d['scores'].items(): self.assertEqual(self.state()['scores'][key], value) @unittest.skipUnless(shutil.which('crm_attribute') and shutil.which('cibadmin'), 'Pacemaker CLI not installed') def test_real_crm_attribute_against_offline_cib(self): # CIB_file forces Pacemaker's file backend: no live daemon is contacted. cib = self.path / 'offline.xml' tree = ET.fromstring(subprocess.check_output(['cibadmin', '--empty'], text=True)) ET.SubElement(tree.find('./configuration/nodes'), 'node', id='1', uname='node-a') ET.SubElement(tree.find('./status'), 'node_state', id='1', uname='node-a') cib.write_text(ET.tostring(tree, encoding='unicode')) env = dict(self.env, PATH=os.environ['PATH'], CIB_file=str(cib)) self.run_shell('mc_promotion_score_set 100', env=env) before = cib.read_bytes() self.run_shell('mc_promotion_score_set 100', env=env) self.assertEqual(cib.read_bytes(), before, 'Identical observations must not rewrite CIB') self.run_shell('mc_promotion_score_set -INFINITY', env=env) pairs = ET.parse(cib).findall('./status/node_state/transient_attributes/instance_attributes/nvpair') self.assertEqual([(p.get('name'), p.get('value')) for p in pairs], [('master-pg-ha:0', '-INFINITY')]) # Reprobe after loss of transient state must recreate the correct score. tree = ET.parse(cib) state = tree.find('./status/node_state') state.remove(state.find('./transient_attributes')) tree.write(cib) self.run_shell('mc_promotion_score_set 1000', env=env) self.assertEqual(ET.parse(cib).find('.//transient_attributes//nvpair').get('value'), '1000') def test_actual_eligibility_helper_rejects_non_true_or_failed_sql(self): # Exercise the production shell helper's output and error handling. for answer, sql_rc, expected in [('t', 0, 0), ('f', 0, 1), ('', 0, 1), ('t\nf', 0, 1), ('t', 1, 1)]: fixture = '\nas_service_user() { printf "%s" ' + shlex.quote(answer) + '; return ' + str(sql_rc) + '; }\n' self.run_shell('mc_pg_promotion_eligible', expected, fixture=fixture) def test_dispatch_routes_only_operational_actions_through_scoring(self): # Use the real dispatcher with fake handlers; no cluster calls occur. dispatch = AGENT.read_text().split('case "$1" in', 1)[1] for action in ('start', 'stop', 'promote', 'demote', 'rejoin'): code = 'mc_promotion_action() { echo "scored:$1"; return 0; }; set -- ' + action + '\ncase "$1" in' + dispatch result = self.run_shell(code) self.assertEqual(result.stdout.strip(), 'scored:' + action) for action in ('monitor', 'status'): code = 'mc_promotion_monitor() { echo scored-monitor; return 8; }; set -- ' + action + '\ncase "$1" in' + dispatch self.assertEqual(self.run_shell(code, 8).stdout.strip(), 'scored-monitor') for action, handler in [('meta-data', 'mc_promotion_metadata'), ('validate-all', 'mc_promotion_validate'), ('notify', 'managed_notify')]: code = handler + '() { echo plain; return 0; }; set -- ' + action + '\ncase "$1" in' + dispatch self.assertEqual(self.run_shell(code).stdout.strip(), 'plain') self.assertFalse(self.state()['calls']) if __name__ == '__main__': unittest.main(verbosity=2)