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.
79 lines
4.0 KiB
Python
79 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise publication and rollback using temp files and mocked lab boundaries."""
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location('installer', ROOT / 'scripts/test-elitedesk-plugin.py')
|
|
installer = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(installer)
|
|
|
|
|
|
class Installer(unittest.TestCase):
|
|
def scenario(self, fail_monitor=False):
|
|
with tempfile.TemporaryDirectory() as directory, contextlib.ExitStack() as stack:
|
|
directory = Path(directory)
|
|
provider = directory / 'provider'; provider.mkdir()
|
|
agent = provider / 'managed-container'; agent.write_text('previous agent')
|
|
modules = provider / 'managed-container.d'; modules.mkdir()
|
|
old_module = modules / 'old.sh'; old_module.write_text('previous module')
|
|
baseline = directory / 'baseline'
|
|
baseline.write_text(''.join(hashlib.sha256(p.read_bytes()).hexdigest() + ' ' + str(p.relative_to(provider)) + '\n'
|
|
for p in [agent, old_module]))
|
|
backup = directory / 'backups'; backup.mkdir()
|
|
for name, value in dict(PROVIDER=provider, AGENT=agent, MODULES=modules,
|
|
BASELINE=baseline, BACKUP_ROOT=backup).items():
|
|
stack.enter_context(patch.object(installer, name, value))
|
|
stack.enter_context(patch.object(installer.os, 'geteuid', return_value=0))
|
|
stack.enter_context(patch.object(installer.socket, 'gethostname', return_value='DAVID-ELITEDESK'))
|
|
stack.enter_context(patch.object(installer.signal, 'signal'))
|
|
stack.enter_context(patch.object(installer, 'cluster', return_value={}))
|
|
stack.enter_context(patch.object(installer, 'observe', return_value={'unchanged': True}))
|
|
stack.enter_context(patch.dict(os.environ))
|
|
calls = []
|
|
def fake_run(args, **kwargs):
|
|
calls.append(args)
|
|
if args[0] == 'restorecon': return ''
|
|
if args[0] == 'crm_attribute': return '100'
|
|
require_action = args[1]
|
|
self.assertIn(require_action, ['meta-data', 'validate-all', 'monitor', 'status', 'start'])
|
|
if require_action == 'meta-data':
|
|
return '<resource-agent><parameters>' + '<parameter/>' * 16 + '</parameters></resource-agent>'
|
|
if fail_monitor and Path(args[0]) == agent and require_action == 'monitor':
|
|
raise RuntimeError('injected installed monitor failure')
|
|
return ''
|
|
stack.enter_context(patch.object(installer, 'run', side_effect=fake_run))
|
|
old_mask = os.umask(0o022)
|
|
try:
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
if fail_monitor:
|
|
with self.assertRaisesRegex(RuntimeError, 'injected'):
|
|
installer.main()
|
|
self.assertEqual(agent.read_text(), 'previous agent')
|
|
self.assertEqual(old_module.read_text(), 'previous module')
|
|
else:
|
|
installer.main()
|
|
self.assertEqual(agent.read_bytes(), (ROOT / 'refactored/managed-container').read_bytes())
|
|
self.assertTrue((modules / 'applications/postgresql/plugin.sh').is_file())
|
|
self.assertFalse(old_module.exists())
|
|
finally:
|
|
os.umask(old_mask)
|
|
self.assertEqual(len(list(backup.glob('*/results.json'))), 1)
|
|
|
|
def test_success_only_runs_basic_actions(self):
|
|
self.scenario()
|
|
|
|
def test_failed_installed_monitor_restores_agent_and_modules(self):
|
|
self.scenario(fail_monitor=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|