#!/usr/bin/env python3
"""Plugin contract tests: no real runtime, systemd, or cluster operations."""
import json
import os
from pathlib import Path
import shlex
import shutil
import subprocess
import tempfile
import unittest
import xml.etree.ElementTree as ET

from plugin_test_support import LOAD, ROOT

AGENT = ROOT / 'refactored/managed-container'


class Engine(unittest.TestCase):
    def run_shell(self, code, rc=0, env=None):
        result = subprocess.run(['bash', '-c', LOAD + '\n' + code], text=True,
                                capture_output=True, timeout=10, env=env)
        self.assertEqual(result.returncode, rc, result.stderr)
        return result

    def trace(self, command, arguments, expected, rc=0):
        args = ' '.join(shlex.quote(x) for x in arguments)
        fixture = '''
as_service_user() {
    python3 -c 'import json,sys; print(json.dumps(sys.argv[1:]))' "$@"
    return ''' + str(rc) + '''
}
'''
        result = self.run_shell(fixture + command + ' ' + args, rc)
        self.assertEqual(json.loads(result.stdout), expected)

    def test_exec_preserves_argument_boundaries(self):
        args = ['--user', 'postgres', '--env', 'VALUE=a b\n$(touch /UNEXPECTED)',
                'container', 'sh', '-c', 'printf "%s" "$1"', 'sh', "quotes'\" $HOME ;"]
        self.trace('mc_runtime_exec', args, ['podman', 'exec', *args])

    def test_exec_error_is_not_systemd_success_exit_status(self):
        self.trace('mc_runtime_exec', ['container', 'false'], ['podman', 'exec', 'container', 'false'], 143)

    def test_bounded_exec_preserves_timeout_location_and_status(self):
        self.trace('mc_runtime_exec_bounded', ['3s', '--user', 'postgres', 'container', 'psql'],
                   ['timeout', '--kill-after=1s', '3s', 'podman', 'exec', '--user', 'postgres', 'container', 'psql'], 124)

    def test_inspect_and_info(self):
        self.trace('mc_runtime_inspect', ['--format', '{{.State.Running}}', 'container'],
                   ['podman', 'inspect', '--format', '{{.State.Running}}', 'container'])
        self.trace('mc_runtime_info', [], ['podman', 'info'], 125)

    def test_lifecycle_uses_systemd_only(self):
        for action, args in [('start', ['start']), ('stop', ['stop']),
                             ('running', ['is-active', '--quiet']), ('cat', ['cat'])]:
            self.trace('mc_lifecycle_' + action, ['postgres.service'],
                       ['systemctl', '--user', *args, 'postgres.service'])
        self.trace('mc_lifecycle_stop', ['postgres.service'],
                   ['systemctl', '--user', 'stop', 'postgres.service'], 1)

    def test_application_dispatch_and_return_codes(self):
        for action in ['start', 'stop', 'promote', 'demote', 'rejoin']:
            result = self.run_shell('managed_' + action + '() { echo ' + action + '; return 6; }; '
                                    'mc_application_action ' + action, 6)
            self.assertEqual(result.stdout.strip(), action)
        self.run_shell('mc_application_action invalid', 3)
        self.run_shell('managed_monitor() { return 8; }; mc_application_monitor', 8)
        self.run_shell('managed_notify() { return 1; }; mc_application_notify', 1)

    def test_unsupported_or_path_like_selection_rejected_before_commands(self):
        for key, value in [('application', 'mysql'), ('application', '../postgresql'),
                           ('container_runtime', 'docker'), ('lifecycle_manager', 'runtime'),
                           ('application', '$(touch /UNEXPECTED)')]:
            env = dict(os.environ, **{'OCF_RESKEY_' + key: value})
            result = subprocess.run([str(AGENT), 'monitor'], env=env, text=True, capture_output=True)
            self.assertEqual(result.returncode, 6)
            self.assertIn('Supported plugin combination', result.stderr)

    def test_missing_and_incompatible_plugins(self):
        for kind in ['missing', 'version', 'hook']:
            with tempfile.TemporaryDirectory(prefix='mc-plugin-loader-') as directory:
                target = Path(directory) / 'refactored'
                shutil.copytree(AGENT.parent, target, ignore=shutil.ignore_patterns('__pycache__'))
                plugin = target / 'managed-container.d/runtimes/podman.sh'
                if kind == 'missing':
                    plugin.unlink()
                elif kind == 'version':
                    plugin.write_text(plugin.read_text().replace('MC_RUNTIME_API=1', 'MC_RUNTIME_API=2'))
                else:
                    plugin.write_text(plugin.read_text().replace('mc_runtime_exec()', 'missing_hook()'))
                result = subprocess.run([str(target / 'managed-container'), 'meta-data'],
                                        text=True, capture_output=True, cwd='/tmp')
                self.assertEqual(result.returncode, 6, result.stderr)
                self.assertEqual(result.stdout, '')

    def test_metadata_without_instance_and_with_explicit_defaults(self):
        env = {k: v for k, v in os.environ.items() if not k.startswith('OCF_')}
        baseline = subprocess.check_output([str(AGENT), 'meta-data'], env=env, cwd='/tmp', text=True)
        tree = ET.fromstring(baseline)
        params = {p.get('name'): p.find('content').get('default') for p in tree.findall('./parameters/parameter')}
        self.assertEqual(len(params), 16)
        for key, value in [('application', 'postgresql'), ('container_runtime', 'podman'), ('lifecycle_manager', 'systemd')]:
            self.assertEqual(params[key], value)
            env['OCF_RESKEY_' + key] = value
        self.assertEqual(subprocess.check_output([str(AGENT), 'meta-data'], env=env, text=True), baseline)

    def test_metadata_failure_is_not_hidden(self):
        self.run_shell('mc_application_metadata() { return 1; }; mc_engine_metadata', 1)

    def test_application_cannot_call_runtime_or_systemd_directly(self):
        # All execution crosses an explicit adapter; command dependency names
        # remain part of PostgreSQL's compatibility validation in this revision.
        for name in ['postgres', 'auth', 'replication', 'notify', 'promote', 'demote', 'rejoin']:
            text = (AGENT.parent / 'managed-container.d/applications/postgresql' / (name + '.sh')).read_text()
            self.assertNotIn('as_service_user', text, name)
            self.assertNotIn('systemctl --user', text, name)
            self.assertNotIn('podman exec', text, name)


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