"""Updater transaction and service tests; all targets live in temporary roots."""

import importlib.util
from contextlib import ExitStack, nullcontext
import io
import json
import os
from pathlib import Path
import stat
import subprocess
import sys
import unittest
from unittest.mock import patch
import zipfile

from test_regressions import IsolatedTest, ROOT, bash_path, function


spec = importlib.util.spec_from_file_location('cups_project_updater', ROOT / 'update_cups.py')
updater = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = updater
spec.loader.exec_module(updater)


class FakeServices:
    def __init__(self, fail_restore=False, fail_pause=False):
        self.calls = []
        self.fail_restore = fail_restore
        self.fail_pause = fail_pause

    def snapshot(self, manager_changed):
        self.calls.append(('snapshot', manager_changed))

    def pause(self):
        self.calls.append('pause')
        if self.fail_pause:
            raise updater.UpdateError('simulated busy service')

    def restore(self):
        self.calls.append('restore')
        if self.fail_restore:
            self.fail_restore = False
            raise updater.UpdateError('simulated startup failure')

    def stop_for_rollback(self):
        self.calls.append('stop_for_rollback')


class UpdateTests(IsolatedTest):
    def setUp(self):
        super().setUp()
        self.source = self.directory / 'release'
        self.system = self.directory / 'system'
        self.source.mkdir()
        self.system.mkdir()
        self.write(self.source, 'setup_cups.sh', b'#!/bin/bash\necho installer\n')
        for source, target in updater.PROGRAMS.items():
            new = b'version = 2\n' if source.endswith('.py') else b'#!/bin/bash\necho new\n'
            old = b'version = 1\n' if source.endswith('.py') else b'#!/bin/bash\necho old\n'
            self.write(self.source, source, new)
            self.write(self.system, target, old)
        self.write(self.source, 'cups-templates-zh_CN/header.tmpl', b'<div class="header"><div class="body">new')
        self.write(self.source, 'cups-templates-zh_CN/trailer.tmpl', b'<div class="footer">new')
        self.write(self.source, 'cups-docroot-zh_CN/index.html', b'new homepage')
        for path in ('usr/share/cups/templates/header.tmpl', 'usr/share/cups/templates-zh_CN/header.tmpl',
                     'usr/share/cups/templates/trailer.tmpl', 'usr/share/cups/doc-root/index.html'):
            self.write(self.system, path, b'old template')
        self.write(self.system, 'usr/share/cups/doc-root/cups.css', b'.cups-header {}')
        self.protected = [
            'etc/cups/cupsd.conf', 'etc/cups/printers.conf', 'etc/network/interfaces',
            'etc/cups-watchdog/network-watchdog.conf', 'etc/cups-watchdog/print-watchdog.conf',
            'etc/systemd/system/cups-driver-manager.service',
            'etc/systemd/system/cups-driver-manager.service.d/auth.conf',
            'opt/cups-driver-manager/.username', 'opt/cups-driver-manager/.password',
        ]
        for relative in self.protected:
            self.write(self.system, relative, b'preserve this fixture exactly')

    @staticmethod
    def write(root, relative, content):
        path = root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(content)
        return path

    def plan(self):
        return updater.build_plan(self.source, self.system, bash=bash_path())

    def apply(self, changes, services):
        with patch('sys.stdout', new=io.StringIO()):
            return updater.apply_update(changes, self.system, services, {'source': 'test fixture'})

    def assert_originals(self, changes):
        for change in changes:
            self.assertEqual((self.system / change.relative).read_bytes(), change.previous)

    def test_success_preserves_config_credentials_and_makes_verified_backup(self):
        changes = self.plan()
        services = FakeServices()
        for kind in ('network', 'print'):
            (self.system / f'run/cups-watchdog/{kind}.lock').mkdir(parents=True)
        backup = self.apply(changes, services)
        for change in changes:
            target = self.system / change.relative
            self.assertEqual(target.read_bytes(), change.content)
            self.assertEqual(stat.S_IMODE(target.stat().st_mode), stat.S_IMODE(change.metadata.st_mode))
            self.assertEqual((backup / 'files' / change.relative).read_bytes(), change.previous)
        for relative in self.protected:
            self.assertEqual((self.system / relative).read_bytes(), b'preserve this fixture exactly')
            self.assertFalse((backup / 'files' / relative).exists())
        self.assertEqual(json.loads((backup / 'manifest.json').read_text())['status'], 'complete')
        self.assertEqual(services.calls, [('snapshot', True), 'pause', 'restore'])
        self.assertFalse((self.system / 'run/cups-watchdog/network.lock').exists())
        self.assertEqual(self.plan(), [])

    def test_new_files_and_optional_components_are_not_installed(self):
        (self.system / updater.PROGRAMS['cups-driver-manager/driver_manager.py']).unlink()
        self.write(self.source, 'cups-docroot-zh_CN/new.js', b'new feature')
        changes = self.plan()
        self.assertFalse(any(change.relative.startswith('opt/cups-driver-manager/') for change in changes))
        self.assertFalse(any(change.relative.endswith('new.js') for change in changes))

    def test_localized_template_matches_installed_css(self):
        header = next(change for change in self.plan() if change.relative == 'usr/share/cups/templates/header.tmpl')
        self.assertIn(b'class="cups-header"', header.content)
        self.assertIn(b'class="cups-body"', header.content)

    def test_start_failure_rolls_back_all_files_and_restores_services(self):
        changes = self.plan()
        services = FakeServices(fail_restore=True)
        with self.assertRaisesRegex(updater.UpdateError, 'Original files/service states restored'):
            self.apply(changes, services)
        self.assert_originals(changes)
        self.assertEqual(services.calls[-2:], ['stop_for_rollback', 'restore'])
        manifest = next((self.system / 'var/backups/cups-project').glob('*/manifest.json'))
        self.assertEqual(json.loads(manifest.read_text())['status'], 'rolled_back')

    def test_partial_write_failure_rolls_back_previous_replacements(self):
        changes = self.plan()
        real_write = updater.atomic_write
        calls = 0

        def fail_once(*args):
            nonlocal calls
            calls += 1
            if calls == 2:
                raise OSError('simulated disk write failure')
            real_write(*args)

        with patch.object(updater, 'atomic_write', side_effect=fail_once), self.assertRaises(updater.UpdateError):
            self.apply(changes, FakeServices())
        self.assert_originals(changes)

    def test_interrupt_during_write_rolls_back(self):
        changes = self.plan()
        real_write = updater.atomic_write
        calls = 0

        def interrupt_once(*args):
            nonlocal calls
            calls += 1
            real_write(*args)
            if calls == 2:
                raise KeyboardInterrupt()

        with patch.object(updater, 'atomic_write', side_effect=interrupt_once), self.assertRaises(updater.UpdateError):
            self.apply(changes, FakeServices())
        self.assert_originals(changes)

    def test_busy_service_aborts_before_replacing_files(self):
        changes = self.plan()
        services = FakeServices(fail_pause=True)
        with self.assertRaises(updater.UpdateError):
            self.apply(changes, services)
        self.assert_originals(changes)
        self.assertEqual(services.calls, [('snapshot', True), 'pause', 'restore'])

    def test_nonempty_legacy_lock_is_not_deleted(self):
        changes = self.plan()
        lock_file = self.write(self.system, 'run/cups-watchdog/network.lock/owner', b'unknown lock owner')
        with self.assertRaises(updater.UpdateError):
            self.apply(changes, FakeServices())
        self.assert_originals(changes)
        self.assertTrue(lock_file.is_file())

    def test_modified_file_aborts_before_service_changes(self):
        changes = self.plan()
        self.write(self.system, changes[0].relative, b'concurrent edit')
        services = FakeServices()
        with self.assertRaisesRegex(updater.UpdateError, 'changed during backup'):
            self.apply(changes, services)
        self.assertEqual(services.calls, [])
        self.assertEqual((self.system / changes[0].relative).read_bytes(), b'concurrent edit')

    def test_invalid_release_is_rejected_before_apply(self):
        self.write(self.source, 'watchdog/network-watchdog.sh', b'if then\n')
        with self.assertRaises(updater.UpdateError):
            self.plan()
        self.assertFalse((self.system / 'var/backups').exists())

    def test_incomplete_release_is_rejected(self):
        (self.source / 'cups-driver-manager/driver_manager.py').unlink()
        with self.assertRaisesRegex(updater.UpdateError, 'Incomplete release'):
            self.plan()

    def test_vanilla_cups_is_not_treated_as_a_project_install(self):
        blank = self.directory / 'vanilla'
        self.write(blank, 'usr/share/cups/doc-root/index.html', b'default CUPS')
        with self.assertRaisesRegex(updater.UpdateError, 'No existing installation'):
            updater.build_plan(self.source, blank, bash=bash_path())

    def test_rollback_failure_is_explicit_and_backup_is_retained(self):
        changes = self.plan()
        real_write = updater.atomic_write
        calls = 0

        def fail_rollback(*args):
            nonlocal calls
            calls += 1
            if calls > len(changes):
                raise OSError('simulated rollback failure')
            real_write(*args)

        with patch.object(updater, 'atomic_write', side_effect=fail_rollback), self.assertRaisesRegex(updater.UpdateError, 'rollback failure'):
            self.apply(changes, FakeServices(fail_restore=True))
        manifest = next((self.system / 'var/backups/cups-project').glob('*/manifest.json'))
        self.assertEqual(json.loads(manifest.read_text())['status'], 'rollback_failed')
        self.assertTrue((manifest.parent / 'files' / changes[0].relative).is_file())

    def archive(self, entries):
        archive = self.directory / 'release.zip'
        with zipfile.ZipFile(archive, 'w') as handle:
            for name, data in entries:
                handle.writestr(name, data)
        return archive

    def test_archive_traversal_is_rejected(self):
        archive = self.archive([('../escape', b'x')])
        with self.assertRaises(updater.UpdateError):
            updater.extract_archive(archive, self.directory / 'extracted')
        self.assertFalse((self.directory / 'escape').exists())

    def test_archive_symlink_is_rejected(self):
        entry = zipfile.ZipInfo('release/link')
        entry.create_system = 3
        entry.external_attr = (stat.S_IFLNK | 0o777) << 16
        archive = self.archive([(entry, b'/etc/cups')])
        with self.assertRaises(updater.UpdateError):
            updater.extract_archive(archive, self.directory / 'extracted')

    def test_archive_root_is_discovered(self):
        archive = self.archive([('project/setup_cups.sh', b'#!/bin/bash\n')])
        root = updater.extract_archive(archive, self.directory / 'extracted')
        self.assertEqual(root.name, 'project')

    def test_plain_http_download_is_rejected(self):
        with self.assertRaises(updater.UpdateError), patch.object(updater.urllib.request, 'urlopen') as request:
            updater.fetch_archive('http://example.invalid/release.zip', self.directory / 'download.zip')
        request.assert_not_called()


class MainTests(IsolatedTest):
    def setUp(self):
        super().setUp()
        stack = ExitStack()
        self.addCleanup(stack.close)
        stack.enter_context(patch.object(updater, 'check_prerequisites'))
        stack.enter_context(patch.object(updater, 'update_lock', return_value=nullcontext()))
        stack.enter_context(patch.object(updater, 'safe_path', return_value=self.directory / 'lock'))
        stack.enter_context(patch.object(updater.tempfile, 'TemporaryDirectory', return_value=nullcontext(str(self.directory))))
        stack.enter_context(patch('sys.stdout', new=io.StringIO()))
        self.plan = stack.enter_context(patch.object(updater, 'build_plan', return_value=[
            updater.Change('opt/cups-watchdog/network-watchdog.sh', b'new', b'old', None),
        ]))
        self.apply = stack.enter_context(patch.object(updater, 'apply_update'))
        self.confirm = stack.enter_context(patch.object(updater, 'confirm', return_value=False))
        self.fetch = stack.enter_context(patch.object(updater, 'fetch_archive'))

    def run_main(self, *args):
        return updater.main(['--source-dir', str(self.directory), *args])

    def test_dry_run_never_changes_files_or_services(self):
        self.assertEqual(self.run_main('--dry-run', '--yes'), 0)
        self.apply.assert_not_called()
        self.confirm.assert_not_called()
        self.fetch.assert_not_called()

    def test_cancel_never_applies_the_plan(self):
        self.assertEqual(self.run_main(), 0)
        self.confirm.assert_called_once()
        self.apply.assert_not_called()

    def test_no_changes_never_prompts_or_applies(self):
        self.plan.return_value = []
        self.assertEqual(self.run_main(), 0)
        self.confirm.assert_not_called()
        self.apply.assert_not_called()

    def test_yes_applies_without_a_terminal(self):
        self.assertEqual(self.run_main('--yes'), 0)
        self.confirm.assert_not_called()
        self.apply.assert_called_once()


class ServiceTests(unittest.TestCase):
    def setUp(self):
        self.calls = []
        self.units = {
            updater.TIMERS[0]: ['loaded', 'active', 'enabled'],
            updater.TIMERS[1]: ['loaded', 'inactive', 'disabled'],
            updater.WATCHDOGS[0]: ['loaded', 'inactive', 'static'],
            updater.WATCHDOGS[1]: ['loaded', 'inactive', 'static'],
            updater.MANAGER: ['loaded', 'active', 'enabled'],
        }
        self.services = updater.Services(runner=self.run_command, sleeper=lambda _: None)
        self.busy = patch.object(self.services, 'busy_processes', return_value=([], set()))
        self.busy.start()
        self.addCleanup(self.busy.stop)

    def run_command(self, cmd, **kwargs):
        self.calls.append(cmd)
        operation, unit = cmd[1:3]
        state = self.units.get(unit, ['not-found', 'inactive', ''])
        if operation == 'show':
            output = f'LoadState={state[0]}\nActiveState={state[1]}\nUnitFileState={state[2]}\nMainPID=123\n'
            return subprocess.CompletedProcess(cmd, 1 if state[0] == 'not-found' else 0, output, '')
        state[1] = 'inactive' if operation == 'stop' else 'active'
        return subprocess.CompletedProcess(cmd, 0, '', '')

    def test_only_previously_active_units_are_restored(self):
        self.services.snapshot(True)
        self.services.pause()
        self.services.restore()
        actions = [(cmd[1], cmd[2]) for cmd in self.calls if cmd[1] != 'show']
        self.assertEqual(actions, [('stop', updater.TIMERS[0]), ('stop', updater.MANAGER),
                                   ('start', updater.MANAGER), ('start', updater.TIMERS[0])])
        self.assertNotIn('cups.service', str(self.calls))
        self.assertNotIn('enable', str(actions))

    def test_missing_optional_units_are_supported(self):
        self.units.clear()
        self.services.snapshot(True)
        self.services.pause()
        self.services.restore()
        self.assertFalse(any(cmd[1] != 'show' for cmd in self.calls))

    def test_active_watchdog_is_not_killed(self):
        self.units[updater.WATCHDOGS[0]][1] = 'activating'
        self.services.snapshot(True)
        with self.assertRaisesRegex(updater.UpdateError, 'still running'):
            self.services.pause()
        self.services.restore()
        self.assertNotIn(['systemctl', 'stop', updater.WATCHDOGS[0]], self.calls)
        self.assertEqual(self.units[updater.TIMERS[0]][1], 'active')

    def test_driver_installation_blocks_manager_stop(self):
        self.services.snapshot(True)
        with patch.object(self.services, 'busy_processes', return_value=([], {124})), self.assertRaisesRegex(updater.UpdateError, 'installation'):
            self.services.pause()
        self.assertNotIn(['systemctl', 'stop', updater.MANAGER], self.calls)


class LauncherTests(unittest.TestCase):
    def test_update_dispatch_does_not_call_the_installer(self):
        source = (ROOT / 'setup_cups.sh').read_text(encoding='utf-8')
        dispatch = source.rsplit('case "${1:-}" in', 1)[1]
        script = 'set -- --update --dry-run\nupdate_cups() { echo "UPDATE:$*"; }; main() { echo UNEXPECTED_INSTALL; };\n'
        result = subprocess.run([bash_path(), '-s'], input=script + 'case "${1:-}" in' + dispatch,
                                capture_output=True, text=True, encoding='utf-8', timeout=10)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout.strip(), 'UPDATE:--dry-run')

    def test_update_launcher_has_no_install_configuration_calls(self):
        source = (ROOT / 'setup_cups.sh').read_text(encoding='utf-8')
        body = function(source, 'update_cups')
        for command in ('configure_cups', 'configure_static_ip', 'install_driver_manager', 'apt install'):
            self.assertNotIn(command, body)


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