feat: add configuration-preserving CUPS project updater
This commit is contained in:
Codex
@@ -0,0 +1,381 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user