only 2 PW

tests/instructions.md

@@ -59,3 +59,8 @@ Automatische Tests prüfen die aktuelle Software ohne Pi-Hardware.
   Rückfall unabhängig von Passwort-Tickets geprüft werden. Abgelehnte Anmeldung,
   unprivilegierten Aufruf des internen Rootteils und unveränderte Daten prüfen.
   Kein echter root-/PAM-/systemd- oder Pi-Nachweis aus diesen Tests.
+
+- UploadConnectionTest simuliert ssh/rsync und prüft genau einen Master, denselben
+  privaten Socket für alle Schritte, BatchMode auf Slaves, TTY für sudo sowie
+  Aufräumen bei Erfolg/Verbindungs-/Transfer-/Aktivierungsfehlern und pure Vorschau.
+  Dies bestätigt keine reale SSH-/Passwortanmeldung am Pi.

tests/test_pi_update.py

@@ -2,6 +2,8 @@
 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
@@ -235,5 +237,96 @@ exit 0
         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()