import contextlib
import fcntl
import io
import json
from pathlib import Path
import runpy
import shlex
import subprocess
import tempfile
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[2]
helper = runpy.run_path(str(ROOT / 'scripts/deploy-update-remote.py'))
controls = runpy.run_path(str(ROOT / 'scripts/service-control-remote.py'), init_globals=helper)
control = controls['control']
wrapper = runpy.run_path(str(ROOT / 'scripts/service-control.py'))


class Controlled:
    service = 'werkjournal-backend.service'

    def __init__(self, home):
        self.app = home / 'opt/werkjournal'
        self.app.mkdir(parents=True)
        self.pending = self.app / 'pending-deployment.json'
        self.flag = home / 'maintenance.flag'
        self.calls = []
        self.expected = {'commit': 'verified'}
        (self.app / 'deployed.json').write_text(json.dumps(self.expected))
        self.fail_ready = False

    def preflight(self, check_data=True):
        self.calls.append(('preflight', check_data))

    def stop(self):
        self.calls.append(('stop',))

    def start(self):
        self.calls.append(('start',))

    def ready(self, expected):
        self.calls.append(('ready', expected))
        if self.fail_ready:
            raise RuntimeError('not ready')

    def command(self, *args):
        self.calls.append(args)


class ServiceControlTest(unittest.TestCase):
    def setUp(self):
        directory = tempfile.TemporaryDirectory()
        self.addCleanup(directory.cleanup)
        self.deployment = Controlled(Path(directory.name))
        patcher = patch.dict(control.__globals__, manifest=lambda path: self.deployment.expected)
        patcher.start()
        self.addCleanup(patcher.stop)
        self.output = contextlib.redirect_stdout(io.StringIO())
        self.output.__enter__()
        self.addCleanup(self.output.__exit__, None, None, None)

    def test_restart_only_verified_release_and_preserves_maintenance(self):
        self.deployment.flag.write_text('Operator maintenance')
        self.assertEqual(0, control(self.deployment, 'restart'))
        self.assertEqual([('preflight', True), ('stop',), ('start',), ('ready', {'commit': 'verified'})], self.deployment.calls)
        self.assertEqual('Operator maintenance', self.deployment.flag.read_text())
        self.assertEqual({'commit': 'verified'}, json.loads((self.deployment.app / 'deployed.json').read_text()))

    def test_unverified_release_and_pending_deployment_cannot_be_started(self):
        (self.deployment.app / 'deployed.json').write_text('{"commit":"other"}')
        with self.assertRaisesRegex(RuntimeError, 'differs'):
            control(self.deployment, 'start')
        self.assertEqual([('preflight', True)], self.deployment.calls)
        self.deployment.calls.clear()
        self.deployment.pending.write_text('{}')
        with self.assertRaisesRegex(RuntimeError, 'pending'):
            control(self.deployment, 'stop')
        self.assertEqual([], self.deployment.calls)

    def test_concurrent_deployment_lock_blocks_service_mutations(self):
        with (self.deployment.app / '.deploy.lock').open('a') as lock:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            for action in controls['WRITE_ACTIONS']:
                with self.assertRaises(BlockingIOError):
                    control(self.deployment, action)
        self.assertEqual([], self.deployment.calls)

    def test_enable_and_disable_do_not_start_or_stop_the_service(self):
        for action in ('enable', 'disable'):
            self.deployment.calls.clear()
            control(self.deployment, action)
            self.assertEqual([('preflight', False), ('systemctl', '--user', action, self.deployment.service)], self.deployment.calls)

    def test_failed_readiness_is_not_success_and_preserves_maintenance(self):
        self.deployment.fail_ready = True
        self.deployment.flag.write_text('Keep closed')
        with self.assertRaisesRegex(RuntimeError, 'not ready'):
            control(self.deployment, 'start')
        self.assertEqual('Keep closed', self.deployment.flag.read_text())

    def test_read_only_actions_work_during_deployment_and_propagate_exit_code(self):
        self.deployment.pending.write_text('{}')
        with patch.object(subprocess, 'call', return_value=3) as call:
            self.assertEqual(3, control(self.deployment, 'status'))
            call.assert_called_once_with(['systemctl', '--user', 'status', self.deployment.service, '--no-pager'])
        args = ['--since', '1 hour ago', '--grep', '$(touch /tmp/should-not-exist);`id`']
        with patch.object(subprocess, 'call', return_value=0) as call:
            control(self.deployment, 'log', args)
            call.assert_called_once_with(['journalctl', '--user', '-u', self.deployment.service, '-n', '100', *args, '--no-pager'])
        self.assertEqual([], self.deployment.calls)

    def test_ssh_wrapper_round_trips_arguments_without_shell_expansion(self):
        args = ['--since', '1 hour ago', '--grep', "$(touch /tmp/no); 'quoted' `id`"]
        command = wrapper['command']('user@host', 'log', args)
        self.assertEqual(['ssh', '-o', 'BatchMode=yes', 'user@host'], command[:4])
        words = shlex.split(command[4])
        self.assertEqual(['log', *args], words[3:])
        self.assertEqual(['python3', '-c'], words[:2])
        with self.assertRaises(ValueError):
            wrapper['command']('user@host', 'restart', ['unwanted'])
        with self.assertRaises(ValueError):
            control(self.deployment, 'unknown')