Lyncoln
ec2077f979
Extract PostgreSQL, Podman runtime and systemd lifecycle adapters while preserving existing lifecycle and scoring behavior. Update RPM layout and add guarded operator-run lab harnesses. Record successful manual stop/start, promotion, demotion and rewind-only rejoin with pinned installed hashes. The unfenced lab does not validate automatic failover, Pacemaker-managed transitions or divergent-data rewind. Validation: 103 offline tests, original-function differential checks, checksum manifest and whitespace checks passed.
56 lines
2.6 KiB
Python
56 lines
2.6 KiB
Python
"""Exercise the safety gates without running cluster or PostgreSQL commands."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('roles', Path(__file__).resolve().parents[1] / 'scripts/test-lab-roles.py')
|
|
roles = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(roles)
|
|
|
|
|
|
def db(recovery, source=None, **changes):
|
|
return dict(dict(system_id='same', recovery=recovery, paused=False,
|
|
receiver='streaming' if source else None,
|
|
sender=roles.NODES.get(source)), **changes)
|
|
|
|
|
|
class Gates(unittest.TestCase):
|
|
def test_demote_requires_streaming_successor(self):
|
|
roles.guard('demote', 'DAVID-DV6', db(False), db(True, 'DAVID-DV6'))
|
|
for peer in [db(False), db(True), db(True, 'DAVID-DV6', paused=True)]:
|
|
with self.assertRaises(RuntimeError):
|
|
roles.guard('demote', 'DAVID-DV6', db(False), peer)
|
|
|
|
def test_promote_refuses_live_peer_primary(self):
|
|
with self.assertRaises(RuntimeError):
|
|
roles.guard('promote', 'DAVID-ELITEDESK', db(True, 'DAVID-DV6'), db(False))
|
|
|
|
def test_promote_requires_demoted_peer_streaming_back(self):
|
|
roles.guard('promote', 'DAVID-ELITEDESK', db(True), db(True, 'DAVID-ELITEDESK'))
|
|
with self.assertRaises(RuntimeError):
|
|
roles.guard('promote', 'DAVID-ELITEDESK', db(True), db(True))
|
|
|
|
def test_identity_mismatch_refused(self):
|
|
with self.assertRaises(RuntimeError):
|
|
roles.guard('demote', 'DAVID-DV6', db(False), db(True, 'DAVID-DV6', system_id='other'))
|
|
|
|
def test_rejoin_only_healthy_elitedesk_standby(self):
|
|
roles.guard('rejoin', 'DAVID-ELITEDESK', db(True, 'DAVID-DV6'), db(False))
|
|
for node, local, peer in [('DAVID-DV6', db(True, 'DAVID-ELITEDESK'), db(False)),
|
|
('DAVID-ELITEDESK', db(False), db(True)),
|
|
('DAVID-ELITEDESK', db(True, 'DAVID-DV6'), db(True))]:
|
|
with self.assertRaises(RuntimeError):
|
|
roles.guard('rejoin', node, local, peer)
|
|
|
|
def test_pinned_versions_match_lab_snapshots(self):
|
|
root = Path(__file__).resolve().parents[1]
|
|
for node, name in [('DAVID-DV6', 'lab-installed-before-plugin.sha256'),
|
|
('DAVID-ELITEDESK', 'lab-installed-plugin.sha256')]:
|
|
expected = {p.removeprefix(str(roles.PROVIDER) + '/'): h for h, p in
|
|
(line.split(' ', 1) for line in (root / 'docs' / name).read_text().splitlines())}
|
|
self.assertEqual(roles.PINNED[node], expected)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|