"""Filesystem deployment tests with simulated systemd, Python setup and health. No SSH, package downloads, real services or Pi hardware are used. """ import os import json import sys from pathlib import Path import shutil import subprocess import tempfile import unittest PROJECT = Path(__file__).resolve().parents[1] class PiUpdateTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='halloween-update-') self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.root = self.base / 'pi' self.bin = self.base / 'bin' self.bin.mkdir() self.root.mkdir() (self.root / 'data').mkdir() (self.root / 'data/settings.json').write_text('original settings') (self.root / 'data/events.sqlite3').write_bytes(b'original database') self.new = self.root / 'releases/new' self.new.mkdir(parents=True) shutil.copytree(PROJECT / 'tools', self.new / 'tools') (self.new / 'halloween').mkdir() for file in ('run.py', 'requirements.txt'): shutil.copy(PROJECT / file, self.new / file) self.old = self.root / 'releases/old' self.old.mkdir() (self.old / 'run.py').write_text('# old version') (self.root / 'current').symlink_to(self.old) self.unit = self.base / 'unit' self.unit.write_text('# Managed by Halloween project: tools/install_pi_service.sh\nold unit\n') self.env = dict(os.environ, PATH=f'{self.bin}:{os.environ["PATH"]}', TEST_BASE=str(self.base), TEST_ROOT=str(self.root)) # Relocate the fixed unit path only in temporary test copies. No root # commands run and no host /etc files are ever accessed by these tests. for script in ('activate_pi_release.sh', 'install_pi_service.sh'): file = self.new / 'tools' / script contents = file.read_text() contents = contents.replace('unit=/etc/systemd/system/$service', f'unit={self.unit}') contents = contents.replace('unit_path="/etc/systemd/system/$service_name"', f'unit_path="{self.unit}"') file.write_text(contents) self.script('id', ''' if [[ $1 == -u ]]; then if [[ ${TEST_PRIVILEGED:-0} == 1 ]]; then echo 0; else echo 1000; fi else echo piuser fi ''') self.env['TEST_SUDO_UID'] = str(os.getuid()) self.env['TEST_SUDO_GID'] = str(os.getgid()) self.script('sudo', ''' echo called >> "$TEST_BASE/sudo.log" if [[ -f $TEST_BASE/deny-sudo ]]; then exit 1; fi # Fail a second sudo call, simulating an expired/zero-lifetime ticket. if [[ $(wc -l < "$TEST_BASE/sudo.log") != 1 ]]; then exit 1; fi export TEST_PRIVILEGED=1 SUDO_USER=piuser SUDO_UID="$TEST_SUDO_UID" SUDO_GID="$TEST_SUDO_GID" exec "$@" ''') self.script('runuser', ''' [[ $1 == -u && $2 == piuser && $3 == -- ]] || exit 2 shift 3 export TEST_PRIVILEGED=0 exec "$@" ''') self.script('systemctl', ''' echo "$*" >> "$TEST_BASE/systemctl.log" case $1 in is-enabled|is-active|daemon-reload|enable|disable|stop) exit 0 ;; restart) if [[ -f $TEST_BASE/fail-start && $(readlink "$TEST_ROOT/current") == */new ]]; then echo changed > "$TEST_ROOT/data/events.sqlite3" exit 1 fi ;; esac ''') self.script('python3', ''' if [[ $1 != -m || $2 != venv ]]; then exit 2; fi mkdir -p "$4/bin" cp "$TEST_BASE/bin/fake-python" "$4/bin/python" ''') self.script('fake-python', ''' if [[ $1 == -m && $2 == pip && -f $TEST_BASE/fail-deps ]]; then exit 1; fi if [[ $1 == - ]]; then code=$(cat) if [[ $code == *connect_ex* && -f $TEST_BASE/port-busy ]]; then exit 1; fi if [[ $code == *urlopen* && -f $TEST_BASE/fail-all-health ]]; then exit 1; fi if [[ $code == *urlopen* && -f $TEST_BASE/fail-health && $(readlink "$TEST_ROOT/current") == */new ]]; then echo migrated > "$TEST_ROOT/data/events.sqlite3" exit 1 fi fi exit 0 ''') def script(self, name, content): path = self.bin / name path.write_text('#!/usr/bin/env bash\nset -e\n' + content + '\n') path.chmod(0o755) def activate(self): return subprocess.run(['bash', str(self.new / 'tools/activate_pi_release.sh'), str(self.root), 'new', '0.0.0.0'], env=self.env, capture_output=True, text=True) def assert_single_sudo(self): self.assertEqual((self.base / 'sudo.log').read_text().splitlines(), ['called']) def assert_data(self): self.assertEqual((self.root / 'data/settings.json').read_text(), 'original settings') self.assertEqual((self.root / 'data/events.sqlite3').read_bytes(), b'original database') def test_success_preserves_data_and_enables_autostart(self): result = self.activate() self.assertEqual(result.returncode, 0, result.stderr) self.assert_single_sudo() self.assertEqual((self.root / 'current').resolve(), self.new) self.assert_data() self.assertIn(f'Environment=HALLOWEEN_STATE_DIR={self.root}/data', self.unit.read_text()) self.assertIn('--host 0.0.0.0', self.unit.read_text()) self.assertIn('enable halloween-grab.service', (self.base / 'systemctl.log').read_text()) def test_start_and_health_failure_restore_code_unit_and_data(self): for failure in ('fail-start', 'fail-health'): with self.subTest(failure=failure): (self.base / failure).touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertEqual((self.root / 'current').resolve(), self.old) self.assert_data() self.assertIn('old unit', self.unit.read_text()) self.assertIn('Rückfall abgeschlossen', result.stderr) self.assert_single_sudo() (self.base / failure).unlink() shutil.rmtree(self.root / 'backups/new') (self.base / 'sudo.log').unlink() def test_dependency_failure_leaves_service_running(self): (self.base / 'fail-deps').touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertEqual((self.root / 'current').resolve(), self.old) self.assert_data() self.assertFalse((self.base / 'systemctl.log').exists()) self.assertFalse((self.base / 'sudo.log').exists()) def test_first_install_failure_has_no_previous_version(self): self.unit.unlink() (self.root / 'current').unlink() (self.base / 'fail-health').touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertFalse(self.unit.exists()) self.assertFalse((self.root / 'current').is_symlink()) self.assert_data() def test_manual_server_blocks_first_install_without_stopping_it(self): self.unit.unlink() (self.root / 'current').unlink() (self.base / 'port-busy').touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertFalse((self.base / 'systemctl.log').exists()) self.assert_data() def test_flat_installation_is_available_for_rollback(self): self.unit.unlink() (self.root / 'current').unlink() (self.root / 'run.py').write_text('# legacy version') (self.root / '.venv/bin').mkdir(parents=True) shutil.copy(self.bin / 'fake-python', self.root / '.venv/bin/python') (self.base / 'fail-start').touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertEqual((self.root / 'current').resolve(), self.root) self.assertIn(str(self.root / 'current/.venv/bin/python'), self.unit.read_text()) self.assert_data() self.assertIn('Rückfall abgeschlossen', result.stderr) def test_first_install_success(self): self.unit.unlink() (self.root / 'current').unlink() result = self.activate() self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.root / 'current').resolve(), self.new) self.assert_data() def test_failed_rollback_health_is_reported(self): (self.base / 'fail-all-health').touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertIn('Rückfall unvollständig', result.stderr) self.assert_data() def test_upload_preview_never_connects(self): for name in ('ssh', 'rsync'): self.script(name, 'echo called > "$TEST_BASE/connected"; exit 1') # Use real Python for the local release ID; no Pi package setup in preview. (self.bin / 'python3').unlink() result = subprocess.run(['bash', str(PROJECT / 'tools/upload_to_pi.sh'), '--user', 'piuser', '--host', 'pi.local', '--remote-dir', '/home/piuser/halloween'], env=self.env, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.base / 'connected').exists()) self.assertIn('releases/', result.stdout) def test_denied_sudo_never_stops_service_or_changes_active_data(self): (self.base / 'deny-sudo').touch() result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertEqual((self.root / 'current').resolve(), self.old) self.assert_data() self.assertFalse((self.base / 'systemctl.log').exists()) self.assert_single_sudo() def test_privileged_entry_rejects_direct_unprivileged_call(self): result = subprocess.run(['bash', str(self.new / 'tools/activate_pi_release.sh'), '--privileged', str(self.root), 'new', '0.0.0.0'], env=self.env, capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) self.assertFalse((self.base / 'systemctl.log').exists()) self.assert_data() def test_foreign_unit_is_untouched(self): self.unit.write_text('foreign unit') result = self.activate() self.assertNotEqual(result.returncode, 0) self.assertEqual(self.unit.read_text(), 'foreign unit') self.assertEqual((self.root / 'current').resolve(), self.old) class UploadConnectionTest(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='halloween-ssh-test-') self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.bin = self.base / 'bin' self.bin.mkdir() self.env = dict(os.environ, PATH=f'{self.bin}:{os.environ["PATH"]}', TEST_SSH_BASE=str(self.base)) ssh = self.bin / 'ssh' ssh.write_text('#!' + sys.executable + '\n' + ''' import json, os, pathlib, sys base = pathlib.Path(os.environ['TEST_SSH_BASE']) args = sys.argv[1:] control = next((arg.split('=',1)[1] for arg in args if arg.startswith('ControlPath=')), None) if '-S' in args: control = args[args.index('-S')+1] with (base / 'calls.jsonl').open('a') as log: log.write(json.dumps({'tool':'ssh','args':args,'control':control})+'\\n') path = pathlib.Path(control) if '-M' in args: assert path.parent.stat().st_mode & 0o777 == 0o700 if (base / 'fail-master').exists(): sys.exit(7) path.touch() elif '-O' in args: assert args[args.index('-O')+1] == 'exit' path.unlink(missing_ok=True) else: assert 'BatchMode=yes' in args assert path.exists(), 'Shared connection must exist' if '-tt' in args and (base / 'fail-activation').exists(): sys.exit(7) ''') ssh.chmod(0o755) rsync = self.bin / 'rsync' rsync.write_text('#!' + sys.executable + '\n' + ''' import json, os, pathlib, shlex, sys base = pathlib.Path(os.environ['TEST_SSH_BASE']) args = sys.argv[1:] transport = shlex.split(args[args.index('-e')+1]) control = next(arg.split('=',1)[1] for arg in transport if arg.startswith('ControlPath=')) assert 'BatchMode=yes' in transport assert pathlib.Path(control).exists() with (base / 'calls.jsonl').open('a') as log: log.write(json.dumps({'tool':'rsync','args':args,'control':control})+'\\n') if (base / 'fail-transfer').exists(): sys.exit(7) ''') rsync.chmod(0o755) def upload(self, apply=True): command = ['bash', str(PROJECT / 'tools/upload_to_pi.sh'), '--user', 'piuser', '--host', 'pi.local', '--remote-dir', '/home/piuser/halloween'] if apply: command.append('--apply') return subprocess.run(command, env=self.env, capture_output=True, text=True) def calls(self): return [json.loads(line) for line in (self.base / 'calls.jsonl').read_text().splitlines()] def test_one_authenticated_connection_for_all_steps_and_cleanup(self): result = self.upload() self.assertEqual(result.returncode, 0, result.stderr) calls = self.calls() self.assertEqual(sum('-M' in call['args'] for call in calls), 1) self.assertEqual([call['tool'] for call in calls], ['ssh','ssh','rsync','ssh','ssh']) self.assertEqual(len({call['control'] for call in calls}), 1) self.assertIn('-tt', calls[3]['args']) self.assertIn('-O', calls[-1]['args']) self.assertFalse(Path(calls[0]['control']).parent.exists()) def test_failure_closes_connection_and_stops_following_steps(self): for failure in ('fail-master', 'fail-transfer', 'fail-activation'): with self.subTest(failure=failure): (self.base / failure).touch() result = self.upload() self.assertNotEqual(result.returncode, 0) calls = self.calls() self.assertEqual(sum('-M' in call['args'] for call in calls), 1) self.assertFalse(Path(calls[0]['control']).parent.exists()) if failure == 'fail-master': self.assertEqual(len(calls), 1) else: self.assertIn('-O', calls[-1]['args']) if failure == 'fail-transfer': self.assertFalse(any('-tt' in call['args'] for call in calls)) (self.base / failure).unlink() (self.base / 'calls.jsonl').unlink() def test_preview_does_not_connect_or_create_session(self): result = self.upload(apply=False) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse((self.base / 'calls.jsonl').exists()) self.assertIn('halloween-ssh-VORSCHAU', result.stdout) if __name__ == '__main__': unittest.main()