addet test set

tests/test_pi_update.py

@@ -0,0 +1,239 @@
+"""Filesystem deployment tests with simulated systemd, Python setup and health.
+No SSH, package downloads, real services or Pi hardware are used.
+"""
+import os
+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)
+
+
+if __name__ == '__main__':
+    unittest.main()