Samuel OnojaandSamuel Onoja
6b0d903ad0
* Add Live Activities as configurable bar items * Fix activity owner monitoring and Morph list interactions * Keep activity rows stable during updates and tighten their layout * Use subtle theme tint for activity cards * Tighten activity title and progress spacing * Use theme accent for active progress rings * Fit compact activities to measured titles and badge * Expose Live Activities display controls in Settings * Open activity terminals and streamline the Morph header * Give Morph continuous icons and prioritize workspace navigation * Show activity progress inside its ring * Inset activity progress rings and strengthen their stroke * Fix activity lifecycle, command output handling and list navigation * Reduce activity update work and harden publisher lifetimes * Set up Live Activities for the detected user shell during installation * Document Activities setup, daily use and troubleshooting * Preserve Morph idle width across empty activity updates * Fix dimming flicker, hide inactive music tab and consolidate Activities docs * Fix panel clearance across styles and match previews to theme settings * Prevent panel blur from repainting neighboring islands --------- Co-authored-by: Samuel Onoja <[email protected]>
597 lines
30 KiB
Python
597 lines
30 KiB
Python
"""Run complete installer commands against disposable destination filesystems."""
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
HELPER = REPO / 'scripts/installer/install.py'
|
|
spec = importlib.util.spec_from_file_location('ferese_install', HELPER)
|
|
installer = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(installer)
|
|
PREFIX = installer.PREFIX.lstrip('/')
|
|
PORTAL = 'usr/share/xdg-desktop-portal/ferese-portals.conf'
|
|
UNIT = 'usr/local/lib/systemd/user/xdg-desktop-portal-ferese.service'
|
|
DESKTOP = 'usr/share/wayland-sessions/ferese.desktop'
|
|
|
|
|
|
class InstallerTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.shared = tempfile.TemporaryDirectory()
|
|
cls.source = Path(cls.shared.name) / 'source'
|
|
for rule in installer.INVENTORY['files']:
|
|
if rule['source'].startswith('target/release/'):
|
|
paths = [REPO / rule['source']]
|
|
else:
|
|
paths = sorted(REPO.glob(rule['source']))
|
|
|
|
for path in paths:
|
|
target = cls.source / path.relative_to(REPO)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(b'#!/bin/sh\nexit 0\n' if rule['source'].startswith('target/release/') else path.read_bytes())
|
|
target.chmod(0o755 if rule['source'].startswith('target/release/') else path.stat().st_mode & 0o777)
|
|
|
|
cls.bundles = []
|
|
for name in ('first', 'second'):
|
|
for relative in ('portal/ferese-portals.conf', 'systemd/xdg-desktop-portal-ferese.service', 'ferese.desktop'):
|
|
with (cls.source / 'packaging' / relative).open('a') as stream:
|
|
stream.write(f'\n# {name}\n')
|
|
|
|
output = Path(cls.shared.name) / name
|
|
subprocess.run([sys.executable, str(HELPER), 'bundle', '--source-root', str(cls.source),
|
|
'--output', str(output), '--release-id', name], check=True, capture_output=True)
|
|
cls.bundles.append(output)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.shared.cleanup()
|
|
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name) / 'root'
|
|
pam = self.root / 'etc/pam.d/login'
|
|
pam.parent.mkdir(parents=True)
|
|
pam.write_text('test PAM stack\n')
|
|
self.base = self.root / PREFIX
|
|
|
|
def run_command(self, *args, success=True, fault=None, fault_after=4):
|
|
command = [sys.executable, str(HELPER)]
|
|
if fault:
|
|
# Inject failure at the filesystem boundary, without shipping production test hooks.
|
|
code = f'''import importlib.util, os, sys
|
|
spec = importlib.util.spec_from_file_location('installer', {str(HELPER)!r})
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
original = m.Installation.write_state
|
|
count = 0
|
|
def write(self, *args, **kwargs):
|
|
global count
|
|
original(self, *args, **kwargs)
|
|
count += 1
|
|
if count == {fault_after}:
|
|
{fault}
|
|
m.Installation.write_state = write
|
|
m.main()
|
|
'''
|
|
command = [sys.executable, '-c', code]
|
|
|
|
result = subprocess.run([*command, *map(str, args), '--root', str(self.root)], capture_output=True, text=True)
|
|
if success:
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
else:
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
|
|
return result
|
|
|
|
def install(self, index=0, **kwargs):
|
|
return self.run_command('install', self.bundles[index], **kwargs)
|
|
|
|
def snapshot(self):
|
|
result = {}
|
|
for path in self.root.rglob('*'):
|
|
if path.is_relative_to(self.base / 'transactions') or path.is_relative_to(self.base / 'releases'):
|
|
continue
|
|
if path.is_symlink():
|
|
result[str(path.relative_to(self.root))] = os.readlink(path)
|
|
elif path.is_file() and path.name != '.install.lock':
|
|
result[str(path.relative_to(self.root))] = path.read_bytes()
|
|
|
|
return result
|
|
|
|
def test_launcher_permissions_are_assigned_when_bundling(self):
|
|
self.assertEqual((self.source / 'packaging/ferese-session').stat().st_mode & 0o111, 0)
|
|
for name in ('ferese-session', 'ferese-session-shell', 'ferese-screenshot'):
|
|
self.assertEqual((self.bundles[0] / name).stat().st_mode & 0o777, 0o755)
|
|
|
|
def without_appearance_wallpapers(self, bundle):
|
|
self.without_files(bundle, {'wallpapers/ferese-wallpaper-dark.png', 'wallpapers/ferese-wallpaper-light.png'})
|
|
|
|
def without_files(self, bundle, removed):
|
|
manifest_path = bundle / 'manifest.json'
|
|
manifest = json.loads(manifest_path.read_text())
|
|
for target in removed:
|
|
(bundle / target).unlink()
|
|
|
|
manifest['files'] = [item for item in manifest['files'] if item['target'] not in removed]
|
|
manifest_path.write_text(json.dumps(manifest))
|
|
|
|
def test_license_is_bundled_and_required_for_new_releases(self):
|
|
self.assertEqual((self.bundles[0] / 'licenses/Ferese-LICENSE.txt').read_bytes(), (REPO / 'LICENSE').read_bytes())
|
|
incomplete = Path(self.temporary.name) / 'incomplete'
|
|
shutil.copytree(self.bundles[0], incomplete)
|
|
self.without_files(incomplete, {'licenses/Ferese-LICENSE.txt'})
|
|
result = self.run_command('install', incomplete, success=False)
|
|
self.assertIn('Bundle is incomplete', result.stderr)
|
|
self.assertFalse((self.base / 'current').exists())
|
|
|
|
def test_upgrade_and_rollback_accept_installed_release_before_license(self):
|
|
self.install()
|
|
self.without_files(self.base / 'releases/first', {'licenses/Ferese-LICENSE.txt'})
|
|
self.install(1)
|
|
self.run_command('rollback')
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/first')
|
|
self.assertEqual(installer.Installation(self.root).manifest('first')['release'], 'first')
|
|
|
|
def test_upgrade_and_rollback_accept_installed_release_before_appearance_wallpapers(self):
|
|
self.install()
|
|
legacy = self.base / 'releases/first'
|
|
self.without_appearance_wallpapers(legacy)
|
|
self.install(1)
|
|
self.run_command('rollback')
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/first')
|
|
self.assertEqual(installer.Installation(self.root).manifest('first')['release'], 'first')
|
|
(legacy / 'ferese').write_bytes(b'corrupt')
|
|
with self.assertRaisesRegex(ValueError, 'Checksum mismatch'):
|
|
installer.Installation(self.root).manifest('first')
|
|
|
|
def replace_dark_wallpaper_with_jpeg(self, bundle):
|
|
source = bundle / 'wallpapers/ferese-wallpaper-dark.png'
|
|
source.rename(source.with_suffix('.jpg'))
|
|
path = bundle / 'manifest.json'
|
|
manifest = json.loads(path.read_text())
|
|
for item in manifest['files']:
|
|
if item['target'] == 'wallpapers/ferese-wallpaper-dark.png':
|
|
item['target'] = 'wallpapers/ferese-wallpaper-dark.jpg'
|
|
path.write_text(json.dumps(manifest))
|
|
|
|
def test_upgrade_and_rollback_verify_installed_jpeg_wallpaper(self):
|
|
self.install()
|
|
installed = self.base / 'releases/first'
|
|
self.replace_dark_wallpaper_with_jpeg(installed)
|
|
self.assertEqual(installer.Installation(self.root).manifest('first')['release'], 'first')
|
|
self.install(1)
|
|
self.assertTrue((self.base / 'current/wallpapers/ferese-wallpaper-dark.png').is_file())
|
|
self.assertFalse((self.base / 'current/wallpapers/ferese-wallpaper-dark.jpg').exists())
|
|
self.run_command('rollback')
|
|
self.assertTrue((self.base / 'current/wallpapers/ferese-wallpaper-dark.jpg').is_file())
|
|
(installed / 'wallpapers/ferese-wallpaper-dark.jpg').write_bytes(b'corrupt asset')
|
|
with self.assertRaisesRegex(ValueError, 'Checksum mismatch: wallpapers/ferese-wallpaper-dark.jpg'):
|
|
installer.Installation(self.root).manifest('first')
|
|
|
|
def test_new_bundle_rejects_jpeg_in_place_of_required_png(self):
|
|
changed = Path(self.temporary.name) / 'jpeg-bundle'
|
|
shutil.copytree(self.bundles[0], changed)
|
|
self.replace_dark_wallpaper_with_jpeg(changed)
|
|
with self.assertRaisesRegex(ValueError, 'Invalid inventory entry: wallpapers/ferese-wallpaper-dark.jpg'):
|
|
installer.validate_bundle(changed)
|
|
|
|
def test_new_bundle_requires_appearance_wallpapers(self):
|
|
incomplete = Path(self.temporary.name) / 'incomplete'
|
|
shutil.copytree(self.bundles[0], incomplete)
|
|
self.without_appearance_wallpapers(incomplete)
|
|
result = self.run_command('install', incomplete, success=False)
|
|
self.assertIn('Bundle is incomplete', result.stderr)
|
|
self.assertFalse((self.base / 'current').exists())
|
|
|
|
def test_fresh_upgrade_and_rollback_restore_all_integration(self):
|
|
self.install()
|
|
first = [(self.root / path).read_bytes() for path in (PORTAL, UNIT, DESKTOP)]
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/first')
|
|
self.assertEqual(os.readlink(self.root / 'usr/local/bin/ferese'), '/usr/local/lib/ferese/current/ferese')
|
|
self.install(1)
|
|
self.assertEqual(os.readlink(self.base / 'previous'), 'releases/first')
|
|
for path, old in zip((PORTAL, UNIT, DESKTOP), first):
|
|
self.assertNotEqual((self.root / path).read_bytes(), old)
|
|
|
|
self.run_command('rollback')
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/first')
|
|
self.assertEqual(os.readlink(self.base / 'previous'), 'releases/second')
|
|
self.assertEqual([(self.root / path).read_bytes() for path in (PORTAL, UNIT, DESKTOP)], first)
|
|
|
|
def test_upgrade_and_rollback_accept_retired_wallpaper_svg_with_verified_checksum(self):
|
|
legacy = Path(self.temporary.name) / 'legacy'
|
|
shutil.copytree(self.bundles[0], legacy)
|
|
svg = legacy / 'wallpapers/ferese.svg'
|
|
svg.write_text('<svg xmlns="http://www.w3.org/2000/svg"/>\n')
|
|
svg.chmod(0o644)
|
|
manifest_path = legacy / 'manifest.json'
|
|
manifest = json.loads(manifest_path.read_text())
|
|
manifest['files'].append({'target': 'wallpapers/ferese.svg', 'mode': '0644',
|
|
'sha256': installer.digest(svg)})
|
|
manifest_path.write_text(json.dumps(manifest))
|
|
|
|
self.run_command('install', legacy)
|
|
self.assertTrue((self.base / 'current/wallpapers/ferese.svg').is_file())
|
|
self.install(1)
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/second')
|
|
self.assertFalse((self.base / 'current/wallpapers/ferese.svg').exists())
|
|
self.run_command('rollback')
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/first')
|
|
self.assertEqual((self.base / 'current/wallpapers/ferese.svg').read_bytes(), svg.read_bytes())
|
|
|
|
svg.write_text('corrupt legacy wallpaper')
|
|
with self.assertRaisesRegex(ValueError, 'Checksum mismatch: wallpapers/ferese.svg'):
|
|
installer.validate_bundle(legacy)
|
|
|
|
def test_upgrade_uses_previous_installer_ownership_receipts(self):
|
|
self.install()
|
|
release = self.base / 'releases/first'
|
|
(release / 'manifest.json').unlink()
|
|
for rule in installer.INVENTORY['files']:
|
|
if 'destination' in rule and ('portal/' in rule['source'] or 'systemd/' in rule['source']):
|
|
receipt = release / 'installer-files' / rule['source'].removeprefix('packaging/')
|
|
receipt.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(self.root / rule['destination'].lstrip('/'), receipt)
|
|
|
|
# The old installer saved no desktop receipt: only identical incoming files are safe.
|
|
(self.root / DESKTOP).write_bytes((self.bundles[1] / ('integration/' + DESKTOP)).read_bytes())
|
|
self.install(1)
|
|
result = self.run_command('rollback', success=False)
|
|
self.assertIn('predates verified bundles', result.stderr)
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/second')
|
|
|
|
def test_reinstall_is_idempotent_and_keeps_previous(self):
|
|
self.install()
|
|
self.install(1)
|
|
before = self.snapshot()
|
|
self.install(1)
|
|
self.assertEqual(before, self.snapshot())
|
|
self.assertEqual(len(list((self.base / 'transactions').iterdir())), 2)
|
|
|
|
def test_modified_configuration_is_preserved_on_upgrade_and_rollback(self):
|
|
self.install()
|
|
self.install(1)
|
|
(self.root / UNIT).write_text('administrator override\n')
|
|
before = self.snapshot()
|
|
self.run_command('rollback', success=False)
|
|
self.assertEqual(before, self.snapshot())
|
|
self.install(success=False)
|
|
self.assertEqual(before, self.snapshot())
|
|
|
|
def test_explicit_portal_replacement_keeps_backup(self):
|
|
self.install()
|
|
(self.root / PORTAL).write_text('administrator preferences\n')
|
|
self.install(1, success=False)
|
|
self.run_command('install', self.bundles[1], '--replace-portal-config')
|
|
backups = list((self.base / 'transactions').glob('*/backup/*'))
|
|
self.assertTrue(any(path.read_text() == 'administrator preferences\n' for path in backups))
|
|
|
|
def test_normal_failure_restores_previous_files_and_can_retry(self):
|
|
self.install()
|
|
before = self.snapshot()
|
|
self.install(1, success=False, fault="raise OSError('injected write failure')")
|
|
self.assertEqual(before, self.snapshot())
|
|
self.assertFalse((self.base / '.transaction').exists())
|
|
self.install(1)
|
|
|
|
def test_killed_upgrade_recovers_before_next_install(self):
|
|
self.install()
|
|
self.install(1, success=False, fault='os._exit(91)')
|
|
self.assertTrue((self.base / '.transaction/journal.json').is_file())
|
|
result = self.install(1)
|
|
self.assertIn('Recovered interrupted', result.stdout)
|
|
self.assertFalse((self.base / '.transaction').exists())
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/second')
|
|
|
|
def test_failure_after_switching_current_restores_previous_release(self):
|
|
self.install()
|
|
before = self.snapshot()
|
|
self.install(1, success=False, fault="raise OSError('failure after current switch')", fault_after=5)
|
|
self.assertEqual(before, self.snapshot())
|
|
|
|
def test_killed_install_after_current_switch_is_recovered(self):
|
|
self.install()
|
|
before = self.snapshot()
|
|
self.install(1, success=False, fault='os._exit(91)', fault_after=5)
|
|
self.assertEqual(os.readlink(self.base / 'current'), 'releases/second')
|
|
self.run_command('recover')
|
|
self.assertEqual(before, self.snapshot())
|
|
|
|
def test_killed_fresh_install_recovers_absent_files(self):
|
|
before = self.snapshot()
|
|
self.install(success=False, fault='os._exit(91)')
|
|
self.run_command('recover')
|
|
self.assertEqual(before, self.snapshot())
|
|
|
|
def test_recovery_preserves_edits_made_after_interruption(self):
|
|
self.install()
|
|
self.install(1, success=False, fault='os._exit(91)')
|
|
(self.root / UNIT).write_text('edited after crash\n')
|
|
result = self.run_command('recover', success=False)
|
|
self.assertIn('Recovery conflict', result.stderr)
|
|
self.assertEqual((self.root / UNIT).read_text(), 'edited after crash\n')
|
|
self.assertTrue((self.base / '.transaction').exists())
|
|
|
|
def test_concurrent_install_is_rejected(self):
|
|
self.base.mkdir(parents=True)
|
|
with (self.base / '.install.lock').open('w') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
result = self.install(success=False)
|
|
|
|
self.assertIn('in progress', result.stderr)
|
|
self.assertFalse((self.base / 'current').exists())
|
|
|
|
def test_restrictive_umask_does_not_make_installed_release_private(self):
|
|
result = subprocess.run([sys.executable, '-c',
|
|
f"import os, runpy; os.umask(0o077); runpy.run_path({str(HELPER)!r}, run_name='__main__')",
|
|
'install', str(self.bundles[0]), '--root', str(self.root)],
|
|
capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
for relative in ('usr', 'usr/local', PREFIX, PREFIX + '/releases', PREFIX + '/releases/first',
|
|
PREFIX + '/releases/first/wallpapers', 'usr/share/wayland-sessions'):
|
|
self.assertEqual((self.root / relative).stat().st_mode & 0o777, 0o755, relative)
|
|
|
|
def test_dry_run_does_not_write(self):
|
|
before = self.snapshot()
|
|
self.run_command('install', self.bundles[0], '--dry-run')
|
|
self.assertEqual(before, self.snapshot())
|
|
self.assertFalse(self.base.exists())
|
|
|
|
def test_killed_directory_creation_can_retry_with_public_permissions(self):
|
|
# Cover persistent parents, including creation before the lock/journal and
|
|
# during publication. Staging trees are unpublished and rebuilt on retry.
|
|
directories = {'usr', 'usr/local', 'usr/local/lib', PREFIX, PREFIX + '/releases',
|
|
PREFIX + '/transactions', 'usr/local/bin'}
|
|
for rule in installer.INVENTORY['files']:
|
|
if 'destination' in rule:
|
|
parent = Path(rule['destination'].lstrip('/')).parent
|
|
directories.update(str(path) for path in (parent, *parent.parents) if str(path) != '.')
|
|
for relative in sorted(directories):
|
|
with self.subTest(directory=relative), tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary) / 'root'
|
|
pam = root / 'etc/pam.d/login'
|
|
pam.parent.mkdir(parents=True)
|
|
pam.write_text('test PAM stack\n')
|
|
target = root / relative
|
|
code = f'''import importlib.util, os
|
|
from pathlib import Path
|
|
spec = importlib.util.spec_from_file_location('installer', {str(HELPER)!r})
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
original = Path.mkdir
|
|
def mkdir(self, *args, **kwargs):
|
|
original(self, *args, **kwargs)
|
|
if self == Path({str(target)!r}):
|
|
os._exit(91)
|
|
Path.mkdir = mkdir
|
|
os.umask(0o077)
|
|
m.main()
|
|
'''
|
|
args = ['install', str(self.bundles[0]), '--root', str(root)]
|
|
result = subprocess.run([sys.executable, '-c', code, *args], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 91, result.stderr)
|
|
mode = 0o700 if relative.endswith('/transactions') else 0o755
|
|
created_mode = target.stat().st_mode & 0o777
|
|
retry = f"import os, runpy; os.umask(0o077); runpy.run_path({str(HELPER)!r}, run_name='__main__')"
|
|
result = subprocess.run([sys.executable, '-c', retry, *args], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(created_mode, mode)
|
|
self.assertEqual(target.stat().st_mode & 0o777, mode)
|
|
self.assertEqual(os.readlink(root / PREFIX / 'current'), 'releases/first')
|
|
self.assertFalse((root / PREFIX / '.transaction').exists())
|
|
|
|
def test_existing_directory_permissions_are_preserved(self):
|
|
for relative in ('usr', 'usr/local', 'usr/local/lib', PREFIX,
|
|
PREFIX + '/releases', 'usr/share/wayland-sessions'):
|
|
path = self.root / relative
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
path.chmod(0o750)
|
|
self.install()
|
|
for relative in ('usr', 'usr/local', 'usr/local/lib', PREFIX,
|
|
PREFIX + '/releases', 'usr/share/wayland-sessions'):
|
|
self.assertEqual((self.root / relative).stat().st_mode & 0o777, 0o750)
|
|
|
|
def test_directory_creation_restores_umask_on_success_and_failure(self):
|
|
previous = os.umask(0o077)
|
|
try:
|
|
installer.ensure_directory(self.root / 'new')
|
|
self.assertEqual(os.umask(0o077), 0o077)
|
|
with mock.patch.object(Path, 'mkdir', side_effect=OSError('mkdir failed')):
|
|
with self.assertRaises(OSError):
|
|
installer.ensure_directory(self.root / 'failed')
|
|
self.assertEqual(os.umask(0o077), 0o077)
|
|
finally:
|
|
os.umask(previous)
|
|
|
|
def test_custom_pam_is_preserved(self):
|
|
pam = self.root / 'etc/pam.d/ferese-lock'
|
|
pam.write_text('administrator authentication\n')
|
|
self.install()
|
|
self.install(1)
|
|
self.run_command('rollback')
|
|
self.assertEqual(pam.read_text(), 'administrator authentication\n')
|
|
|
|
def test_missing_pam_stack_fails_before_publication(self):
|
|
(self.root / 'etc/pam.d/login').unlink()
|
|
self.install(success=False)
|
|
self.assertFalse((self.base / 'current').exists())
|
|
self.assertFalse((self.root / UNIT).exists())
|
|
|
|
def test_symlinked_destination_parent_is_rejected(self):
|
|
outside = Path(self.temporary.name) / 'outside'
|
|
outside.mkdir()
|
|
(self.root / 'usr').symlink_to(outside)
|
|
self.install(success=False)
|
|
self.assertEqual(list(outside.iterdir()), [])
|
|
|
|
def test_uninstall_preserves_releases_pam_and_unrelated_files(self):
|
|
self.install()
|
|
unrelated = self.root / 'usr/local/bin/unrelated'
|
|
unrelated.write_text('keep\n')
|
|
self.run_command('uninstall')
|
|
self.assertFalse(os.path.lexists(self.base / 'current'))
|
|
self.assertFalse(os.path.lexists(self.root / 'usr/local/bin/ferese'))
|
|
self.assertFalse((self.root / PORTAL).exists())
|
|
self.assertTrue((self.base / 'releases/first/ferese').is_file())
|
|
self.assertTrue((self.root / 'etc/pam.d/ferese-lock').is_file())
|
|
self.assertEqual(unrelated.read_text(), 'keep\n')
|
|
|
|
def test_corrupt_incomplete_unlisted_or_symlink_bundle_is_rejected(self):
|
|
for mutation in ('checksum', 'missing', 'extra', 'symlink', 'mode', 'traversal'):
|
|
with self.subTest(mutation=mutation):
|
|
copy = Path(self.temporary.name) / mutation
|
|
shutil.copytree(self.bundles[0], copy)
|
|
binary = copy / 'ferese'
|
|
if mutation == 'checksum':
|
|
binary.write_text('corrupt')
|
|
elif mutation == 'missing':
|
|
binary.unlink()
|
|
elif mutation == 'extra':
|
|
(copy / 'unexpected').write_text('extra')
|
|
elif mutation == 'symlink':
|
|
binary.unlink()
|
|
binary.symlink_to('/etc/passwd')
|
|
elif mutation == 'mode':
|
|
binary.chmod(0o777)
|
|
else:
|
|
manifest = json.loads((copy / 'manifest.json').read_text())
|
|
manifest['files'][0]['target'] = '../../etc/passwd'
|
|
(copy / 'manifest.json').write_text(json.dumps(manifest))
|
|
|
|
self.run_command('install', copy, success=False)
|
|
self.assertFalse((self.base / 'current').exists())
|
|
|
|
def test_release_id_collision_does_not_change_current(self):
|
|
self.install()
|
|
changed = Path(self.temporary.name) / 'changed'
|
|
shutil.copytree(self.bundles[0], changed)
|
|
manifest = json.loads((changed / 'manifest.json').read_text())
|
|
manifest['features'] = 'different-build'
|
|
(changed / 'manifest.json').write_text(json.dumps(manifest))
|
|
before = self.snapshot()
|
|
self.run_command('install', changed, success=False)
|
|
self.assertEqual(before, self.snapshot())
|
|
|
|
|
|
class FrontendTest(unittest.TestCase):
|
|
def test_elevation_preserves_arguments_for_sudo_and_pkexec(self):
|
|
import shlex
|
|
for elevation in ('sudo', 'pkexec'):
|
|
with self.subTest(elevation=elevation), tempfile.TemporaryDirectory() as directory:
|
|
bin_dir = Path(directory)
|
|
for tool in ('python3', 'dirname', 'uname', 'date'):
|
|
(bin_dir / tool).symlink_to(shutil.which(tool))
|
|
elevate = bin_dir / elevation
|
|
elevate.write_text('#!/bin/sh\nexit 99\n')
|
|
elevate.chmod(0o755)
|
|
result = subprocess.run(['/bin/bash', str(REPO / 'scripts/install.sh'), '--bundle',
|
|
'/tmp/bundle with spaces', '--replace-portal-config', '--dry-run'],
|
|
env=dict(os.environ, PATH=str(bin_dir)), capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
command = shlex.split(result.stdout.removeprefix('Install: '))
|
|
# Building is prohibited as root, but installing an existing bundle is allowed.
|
|
expected = ([] if os.geteuid() == 0 else [elevation]) + [
|
|
'/usr/bin/bash', str(REPO / 'scripts/install-session.sh'), 'install',
|
|
'/tmp/bundle with spaces', '--replace-portal-config']
|
|
self.assertEqual(command, expected)
|
|
|
|
def test_incompatible_or_missing_options_are_rejected(self):
|
|
for options in (['--bundle'], ['--release-id'], ['--bundle-only'],
|
|
['--bundle', '/tmp/bundle', '--offline'],
|
|
['--bundle', '/tmp/bundle', '--resize-metrics'],
|
|
['--bundle', '/tmp/bundle', '--release-id', 'another'],
|
|
['--skip-build']):
|
|
with self.subTest(options=options):
|
|
result = subprocess.run(['/bin/bash', str(REPO / 'scripts/install.sh'), *options],
|
|
capture_output=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
|
|
|
|
class UserSetupTest(unittest.TestCase):
|
|
def test_activity_setup_detects_shell_and_preserves_later_config_edits(self):
|
|
for shell in ('fish', 'bash', 'zsh'):
|
|
with self.subTest(shell=shell), tempfile.TemporaryDirectory(prefix="ferese user's ") as directory:
|
|
home = Path(directory)
|
|
config = home / 'custom config'
|
|
startup = (config / 'fish/conf.d/ferese-activities.fish' if shell == 'fish' else
|
|
home / ('.bashrc' if shell == 'bash' else 'zdir/.zshrc'))
|
|
startup.parent.mkdir(parents=True, exist_ok=True)
|
|
if shell != 'fish':
|
|
original = '# My existing shell configuration\nexport MY_VALUE=kept\n'
|
|
real = home / 'dotfile'
|
|
real.write_text(original)
|
|
startup.symlink_to(real)
|
|
env = dict(HOME=str(home), SHELL='/usr/bin/' + shell, ZDOTDIR=str(home / 'zdir'))
|
|
with mock.patch.dict(os.environ, env), mock.patch.object(installer.subprocess, 'run') as run:
|
|
installer.setup_activities(config, ctl=Path('/test/feresectl'), assets=REPO / 'contrib/activities')
|
|
first = startup.read_text()
|
|
installer.setup_activities(config, ctl=Path('/test/feresectl'), assets=REPO / 'contrib/activities')
|
|
self.assertEqual(run.call_count, 1, 'later installs must not reset edited config')
|
|
self.assertEqual(startup.read_text(), first)
|
|
self.assertTrue((config / f'ferese/activities/ferese.{shell}').exists())
|
|
if shell != 'fish':
|
|
self.assertTrue(first.startswith(original))
|
|
self.assertTrue(startup.is_symlink())
|
|
self.assertEqual(len(list(home.glob('dotfile.before-activities-*'))), 1)
|
|
if shutil.which(shell):
|
|
check = subprocess.run([shell, '-n', str(startup)], capture_output=True)
|
|
self.assertEqual(check.returncode, 0, check.stderr)
|
|
if shell == 'bash':
|
|
check = subprocess.run(['bash', '--rcfile', str(startup), '-i', '-c',
|
|
'[[ ${preexec_functions[*]} == *_ferese_preexec* ]]'],
|
|
env=dict(os.environ, **env), capture_output=True)
|
|
self.assertEqual(check.returncode, 0, check.stderr)
|
|
|
|
def test_failed_activity_configuration_can_be_retried(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = Path(directory)
|
|
with mock.patch.dict(os.environ, SHELL='/usr/bin/fish'), mock.patch.object(installer.subprocess, 'run') as run:
|
|
run.side_effect = subprocess.CalledProcessError(1, 'feresectl')
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
installer.setup_activities(config, assets=REPO / 'contrib/activities')
|
|
self.assertFalse((config / 'ferese/activities/installed.json').exists())
|
|
run.side_effect = None
|
|
installer.setup_activities(config, assets=REPO / 'contrib/activities')
|
|
self.assertTrue((config / 'ferese/activities/installed.json').exists())
|
|
|
|
def test_repeatable_migration_and_reload_after_failure(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
home = Path(directory)
|
|
override = home / '.config/systemd/user/xdg-desktop-portal-ferese.service.d/override.conf'
|
|
override.parent.mkdir(parents=True)
|
|
override.write_text(f'[Service]\nExecStart=\nExecStart={home}/.local/libexec/ferese/xdg-desktop-portal-ferese\n')
|
|
bin_dir = home / 'bin'
|
|
bin_dir.mkdir()
|
|
systemctl = bin_dir / 'systemctl'
|
|
systemctl.write_text('#!/bin/sh\nexit 1\n')
|
|
systemctl.chmod(0o755)
|
|
env = dict(os.environ, HOME=str(home), XDG_CONFIG_HOME=str(home / '.config'), PATH=str(bin_dir))
|
|
# User setup intentionally refuses root; emulate the user identity in root-run CI.
|
|
command = [sys.executable, '-c', f"import runpy, os; os.geteuid = lambda: 1000; runpy.run_path({str(HELPER)!r}, run_name='__main__')", 'user-setup', '--no-activities']
|
|
result = subprocess.run(command, env=env, capture_output=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(override.exists())
|
|
self.assertEqual(len(list(override.parent.glob('*.before-*'))), 1)
|
|
systemctl.write_text('#!/bin/sh\nexit 0\n')
|
|
for _ in range(2):
|
|
result = subprocess.run(command, env=env, capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(len(list(override.parent.glob('*.before-*'))), 1)
|
|
override.write_text('custom configuration\n')
|
|
result = subprocess.run(command, env=env, capture_output=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(override.read_text(), 'custom configuration\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|