#!/usr/bin/env python3
"""Real Pacemaker scheduler tests using synthetic CIB files only."""
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest
import xml.etree.ElementTree as ET


@unittest.skipUnless(shutil.which('crm_simulate') and shutil.which('cibadmin'), 'Pacemaker CLI not installed')
class Scheduler(unittest.TestCase):
    def simulate(self, scores, target=None, fail_primary=False):
        if fail_primary and target is None:
            target = 'Promoted'  # Required production policy.
        with tempfile.TemporaryDirectory(prefix='mc-scheduler-') as directory:
            root = ET.fromstring(subprocess.check_output(['cibadmin', '--empty'], text=True))
            root.set('have-quorum', '1'); root.set('dc-uuid', '1')
            config = root.find('configuration')
            props = ET.SubElement(config.find('crm_config'), 'cluster_property_set', id='test-props')
            # ONLY this synthetic file: no fencing device is required for an offline simulation.
            ET.SubElement(props, 'nvpair', id='test-stonith', name='stonith-enabled', value='true' if fail_primary else 'false')
            for i, score in enumerate(scores, 1):
                node = 'node-' + str(i)
                ET.SubElement(config.find('nodes'), 'node', id=str(i), uname=node)
                state = ET.SubElement(root.find('status'), 'node_state', id=str(i), uname=node,
                                      in_ccm='true', crmd='online', join='member', expected='member')
                transient = ET.SubElement(state, 'transient_attributes', id=str(i))
                attrs = ET.SubElement(transient, 'instance_attributes', id='attrs-' + str(i))
                # Names generated by crm_attribute --promotion=pg-ha, not old pgsql state.
                ET.SubElement(attrs, 'nvpair', id='score-' + str(i), name='master-pg-ha', value=score)
                ET.SubElement(attrs, 'nvpair', id='legacy-' + str(i), name='master-postgresql', value='-INFINITY')
            if fail_primary:
                ET.SubElement(config.find('resources'), 'primitive',
                              {'id': 'test-fence', 'class': 'stonith', 'type': 'fence_dummy'})
            clone = ET.SubElement(config.find('resources'), 'clone', id='pg-ha-clone')
            meta = ET.SubElement(clone, 'meta_attributes', id='clone-meta')
            values = {'promotable': 'true', 'clone-max': str(len(scores)), 'clone-node-max': '1',
                      'promoted-max': '1', 'promoted-node-max': '1', 'notify': 'true'}
            if target is not None:
                values['target-role'] = target
            for name, value in values.items():
                ET.SubElement(meta, 'nvpair', id='clone-' + name, name=name, value=value)
            ET.SubElement(clone, 'primitive', {'id': 'pg-ha', 'class': 'ocf', 'provider': 'loxthelion', 'type': 'managed-container'})
            cib = Path(directory) / 'cib.xml'; graph = Path(directory) / 'graph.xml'
            ET.ElementTree(root).write(cib)
            if fail_primary:
                # Establish running primary/standbys in a synthetic simulation;
                # then inject an unexpected failure. No agents/devices execute.
                subprocess.run(['crm_simulate', '--xml-file', str(cib), '--in-place'],
                               check=True, capture_output=True, text=True)
            subprocess.run(['crm_simulate', '--xml-file', str(cib), '--run', '--save-graph', str(graph)]
                           + (['--node-fail', 'node-1'] if fail_primary else []),
                           check=True, capture_output=True, text=True)
            parsed = ET.parse(graph)
            if fail_primary:
                # Verify fencing is a transitive prerequisite of every promote.
                actions = {op.get('id'): op for op in parsed.findall('./synapse/action_set/*')}
                dependencies = {}
                for synapse in parsed.findall('./synapse'):
                    for op in synapse.findall('./action_set/*'):
                        dependencies[op.get('id')] = {x.get('id') for x in synapse.findall('./inputs/trigger/*')}
                fences = {key for key, op in actions.items() if op.get('operation') == 'stonith'}
                self.assertTrue(fences, ET.tostring(parsed.getroot()).decode())
                def ancestors(key, seen=None):
                    seen = set() if seen is None else seen
                    for parent in dependencies.get(key, set()) - seen:
                        seen.add(parent); ancestors(parent, seen)
                    return seen
                for key, op in actions.items():
                    if op.get('operation') == 'promote':
                        self.assertTrue(ancestors(key) & fences, 'promotion must depend on fencing')
            return [op.get('on_node') for op in ET.parse(graph).findall('./synapse/action_set/rsc_op')
                    if op.get('operation') == 'promote']

    def test_default_started_allows_one_promotion(self):
        self.assertEqual(len(self.simulate(['100', '100'])), 1)

    def test_explicit_started_allows_one_promotion(self):
        self.assertEqual(len(self.simulate(['100', '100'], 'Started')), 1)

    def test_explicit_promoted_allows_one_promotion(self):
        self.assertEqual(len(self.simulate(['100', '100'], 'Promoted')), 1)

    def test_unpromoted_target_prevents_promotion(self):
        self.assertEqual(self.simulate(['100', '100'], 'Unpromoted'), [])

    def test_higher_score_wins_without_host_preference(self):
        self.assertEqual(self.simulate(['100', '1000']), ['node-2'])

    def test_ineligible_node_not_selected(self):
        self.assertEqual(self.simulate(['-INFINITY', '100']), ['node-2'])

    def test_all_ineligible_no_promotion(self):
        self.assertEqual(self.simulate(['-INFINITY', '-INFINITY']), [])

    def test_failed_primary_one_disconnected_candidate(self):
        self.assertEqual(self.simulate(['1000', '50'], fail_primary=True), ['node-2'])

    def test_failed_primary_two_disconnected_candidates(self):
        first = self.simulate(['1000', '50', '50'], fail_primary=True)
        self.assertEqual(len(first), 1)
        self.assertIn(first[0], ['node-2', 'node-3'])
        self.assertEqual(first, self.simulate(['1000', '50', '50'], fail_primary=True))

    def test_failed_primary_all_disconnected_stale(self):
        self.assertEqual(self.simulate(['1000', '-INFINITY', '-INFINITY'], fail_primary=True), [])

    def test_connected_evidence_preferred(self):
        self.assertEqual(self.simulate(['50', '100']), ['node-2'])


if __name__ == '__main__':
    unittest.main(verbosity=2)