diff --git a/script/todo/todo.py b/script/todo/todo.py index b0a2f01..7fabba3 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -1315,6 +1315,11 @@ class TODO: "Remote desktop tunnel (VNC/RDP through SSH)" ) }, + { + "prompt_description": t( + "Android emulator (start, tunnel, scrcpy)" + ) + }, {"section": t("Catalog")}, {"prompt_description": t("List available images and specs")}, ] @@ -1357,6 +1362,8 @@ class TODO: elif status == "14": self._qemu_tunnel_menu() elif status == "15": + self._qemu_emulator_menu() + elif status == "16": self._qemu_list_images() else: cmd_no_found = True @@ -1607,7 +1614,159 @@ class TODO: ) print(f" {t('The tunnel stays open as long as that ssh runs.')}") - def _qemu_scrcpy_tunnel(self, name, src): + @staticmethod + def _qemu_ssh_opts(src): + """Options ssh selon la provenance de la cible. + + Une VM libvirt locale est jointe par son IP, et son IP est recyclée d'un + déploiement à l'autre : sa clé d'hôte change sous le même adresse, et + ssh refuse alors de se connecter — « Host key verification failed », + vécu. C'est la raison pour laquelle le suivi d'installation et l'attente + de sshd emploient déjà ces deux options. + + Un hôte de ~/.ssh/config, lui, est une machine que l'utilisateur a + configurée : on ne touche PAS à sa politique de clés. Sa clé est un + garde-fou qui lui appartient.""" + if src == "ssh_config": + return ["-o", "BatchMode=yes", "-o", "ConnectTimeout=8"] + return [ + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=8", + "-o", + "StrictHostKeyChecking=no", + "-o", + "UserKnownHostsFile=/dev/null", + ] + + def _qemu_ssh_target(self, name, src): + """Destination ssh d'une cible du menu, selon sa provenance. + + Un hôte de ~/.ssh/config se nomme tel quel — c'est lui qui porte le + ProxyJump, et le réécrire à la main reviendrait à le deviner. Un domaine + libvirt local, lui, n'a qu'une IP, et l'utilisateur des VM ERPLibre est + « erplibre ». Renvoie une chaîne vide quand l'IP manque.""" + if src == "ssh_config": + return name + ip = self._qemu_resolve_ips([name]).get(name) + return f"erplibre@{ip}" if ip else "" + + # Commande de l'émulateur dans la VM. Le chemin est ABSOLU : un + # « ssh hôte 'commande' » ne lit ni ~/.profile ni ~/.bashrc. + _QEMU_EMULATOR_BIN = "$HOME/android/emulator/emulator" + _QEMU_AVD_NAME = "erplibre" + + def _qemu_emulator_running(self, target, src="virsh"): + """Nombre d'émulateurs en cours dans la VM. + + Deux sur le même AVD, et le second s'arrête sur « Running multiple + emulators with the same AVD is an experimental feature ». Le savoir + AVANT de lancer évite de lire cette phrase sans la comprendre — vécu, + deux fois.""" + try: + res = subprocess.run( + ["ssh"] + + self._qemu_ssh_opts(src) + + [target, "pgrep -c qemu-system 2>/dev/null || echo 0"], + capture_output=True, + text=True, + timeout=25, + ) + return int((res.stdout or "0").strip().splitlines()[-1]) + except (OSError, subprocess.SubprocessError, ValueError, IndexError): + return -1 + + def _qemu_emulator_menu(self): + """Démarre l'émulateur Android d'une VM, et donne la suite qui va avec. + + La question qui décide de tout est celle de la FENÊTRE : + - avec fenêtre, l'écran voyage en pixels bruts par X11, et la commande + doit partir du poste qui possède l'affichage — donc pas d'ici ; + - sans fenêtre, on peut la lancer d'ici, détachée, et l'image arrive + ensuite par scrcpy en H.264. C'est la voie fluide. + """ + print(f"\n📱 {t('Android emulator')}") + targets = [(h, "ssh_config") for h in self._ssh_config_hosts()] + if not targets: + targets = [(n, "virsh") for n in self._qemu_list_domains()] + if not targets: + print(f" {t('No host in ~/.ssh/config and no local VM.')}") + return + for i, (nm, sr) in enumerate(targets, 1): + mark = "" if sr == "ssh_config" else f" ({t('local VM')})" + print(f" [{i}] {nm}{mark}") + answer = input(f"{t('Which VM?')} [1]: ").strip() or "1" + if not answer.isdigit() or not (1 <= int(answer) <= len(targets)): + print(t("Cancelled.")) + return + name, src = targets[int(answer) - 1] + target = self._qemu_ssh_target(name, src) + if not target: + print(f" {t('No IP for this VM; is it running?')}") + return + + running = self._qemu_emulator_running(target, src) + if running > 0: + print(f"\n ⚠ {t('An emulator is already running on this VM.')}") + print(f" {t('Only one per AVD; close it first:')}") + print(f"\n ssh {target} 'pkill -f \"[q]emu-system-x86_64\"'\n") + if not self._is_yes(input(t("Close it now? (y/N): "))): + return + subprocess.run( + ["ssh"] + + self._qemu_ssh_opts(src) + + [target, 'pkill -f "[q]emu-system-x86_64"'], + capture_output=True, + timeout=30, + ) + print(f" {t('Closed.')}") + + print(f"\n {t('Show a window?')}") + print(f" [1] {t('No window - stream with scrcpy (smoother)')} *") + print(f" [2] {t('Window over ssh -X (raw pixels, slower)')}") + kind = input(f"{t('Choice')} [1]: ").strip() or "1" + emu = self._QEMU_EMULATOR_BIN + avd = self._QEMU_AVD_NAME + + if kind == "2": + # L'affichage appartient au POSTE : cette commande ne peut pas + # partir d'ici, où il n'y a pas d'écran à lui donner. + print(f"\n {t('Run this on YOUR workstation:')}") + print( + f"\n ssh -XC {target} '{emu} -avd {avd}" + " -no-audio -no-boot-anim'\n" + ) + print( + f" {t('X11 compression is on (-XC); the screen is 540x1140.')}" + ) + return + + print(f"\n {t('Starting the emulator without a window...')}") + # « sg kvm » : l'appartenance au groupe est posée à l'installation, mais + # une VM créée avant ce correctif ne l'a pas dans sa session — sans KVM + # l'émulateur refuse de démarrer. setsid le détache, pour qu'il survive + # à la fermeture de ce ssh. + start = ( + f'setsid -f sg kvm -c "{emu} -avd {avd} -no-window -no-audio' + ' -no-boot-anim > /tmp/erplibre-emulator.log 2>&1"' + ) + res = subprocess.run( + ["ssh"] + self._qemu_ssh_opts(src) + [target, start], + capture_output=True, + text=True, + timeout=60, + ) + if res.returncode: + print(f" ⚠ {t('Could not start it:')} {res.stderr.strip()[:200]}") + return + print( + f" {t('Started. Boot takes about a minute; log in the VM:')}" + " /tmp/erplibre-emulator.log" + ) + self._qemu_scrcpy_tunnel(name, src, started=True) + + def _qemu_scrcpy_tunnel(self, name, src, started=False): """Tunnel adb vers l'émulateur Android d'une VM, pour scrcpy. Pourquoi cette voie plutôt que « ssh -X » : par X11, chaque image de @@ -1627,21 +1786,21 @@ class TODO: l'émulateur lui-même — c'est exactement ce que fait « adb connect ». """ port = 5555 + target = self._qemu_ssh_target(name, src) + if not target: + print(f" {t('No IP for this VM; is it running?')}") + return print(f"\n 📱 {t('Android emulator over adb + scrcpy')}") - print(f"\n {t('1. In the VM, start the emulator WITHOUT a window:')}") - emu = "$HOME/android/emulator/emulator" - if src == "ssh_config": - print( - f"\n ssh {name} '{emu} -avd erplibre" - " -no-window -no-audio -no-boot-anim'\n" - ) + if started: + # Inutile de redire comment le démarrer : on vient de le faire. + print(f"\n {t('1. Emulator started, without a window.')}") else: - ip = self._qemu_resolve_ips([name]).get(name) - if not ip: - print(f" {t('No IP for this VM; is it running?')}") - return print( - f"\n ssh erplibre@{ip} '{emu} -avd erplibre" + f"\n {t('1. In the VM, start the emulator WITHOUT a window:')}" + ) + print( + f"\n ssh {target} '{self._QEMU_EMULATOR_BIN} " + f"-avd {self._QEMU_AVD_NAME}" " -no-window -no-audio -no-boot-anim'\n" ) print(f" {t('2. Open the tunnel from YOUR workstation:')}") @@ -1653,18 +1812,69 @@ class TODO: else: host, from_ssh = self._qemu_self_address() user = os.environ.get("USER", "user") + vm_ip = target.split("@")[-1] if not from_ssh: print( f" ⚠ {t('Not in an SSH session: check the host address.')}" ) - print(f"\n ssh -N -L {port}:{ip}:{port} {user}@{host}\n") + # DEUX sauts, et non un seul vers l'hyperviseur : l'émulateur + # n'écoute que sur le 127.0.0.1 de la VM — « ss -ltn » le montre, et + # l'hyperviseur reçoit un refus sur IP_VM:5555. Or « localhost » se + # résout sur le DERNIER hôte de la chaîne : la VM doit donc être ce + # dernier saut, l'hyperviseur n'étant que le relais (-J). + print( + f"\n ssh -N -L {port}:localhost:{port}" + f" -J {user}@{host} erplibre@{vm_ip}\n" + ) + print( + f" {t('(the hypervisor only relays; -J puts the VM last)')}" + ) print(f" {t('3. Then, still on your workstation:')}") print(f"\n adb connect localhost:{port}") print(f" scrcpy -s localhost:{port}\n") print(f" {t('The tunnel stays open as long as that ssh runs.')}") - print( - f" {t('scrcpy on Debian/Ubuntu:')} sudo apt install scrcpy" " adb" + print(f" {t('scrcpy on Debian/Ubuntu:')} sudo apt install scrcpy adb") + + # Ouvrir le tunnel D'ICI n'a de sens que si scrcpy tournera ici : le + # port ressort sur CETTE machine. On le propose donc en le disant, + # plutôt que de le faire d'office depuis un hyperviseur sans écran. + print(f"\n {t('If scrcpy will run on THIS machine, I can open it.')}") + if not self._is_yes(input(t("Open the tunnel now? (y/N): "))): + return + if self._port_in_use(port): + print(f" ⚠ {t('Port already in use here:')} {port}") + print( + f" {t('Close the other tunnel first:')}" + f' pkill -f "{port}:localhost:{port}"' + ) + return + # « ExitOnForwardFailure » : sans lui, un ssh détaché rend 0 alors que + # la redirection a échoué — un succès annoncé pour un tunnel absent. + cmd = ( + ["ssh", "-f", "-N", "-o", "ExitOnForwardFailure=yes"] + + self._qemu_ssh_opts(src) + + ["-L", f"{port}:localhost:{port}", target] ) + res = subprocess.run(cmd, capture_output=True, text=True, timeout=40) + if res.returncode: + print(f" ⚠ {t('Tunnel failed:')} {res.stderr.strip()[:200]}") + return + print(f" ✅ {t('Tunnel open on localhost:')}{port}") + print( + f" {t('Then:')} adb connect localhost:{port}" + f" && scrcpy -s localhost:{port}" + ) + print(f' {t("To close it:")} pkill -f "{port}:localhost:{port}"') + + @staticmethod + def _port_in_use(port): + """Le port est-il déjà pris sur CETTE machine ? + + Un second tunnel sur le même port échouerait, et le message d'ssh + (« bind: Address already in use ») se perd en mode détaché.""" + with socket.socket() as sock: + sock.settimeout(1) + return sock.connect_ex(("127.0.0.1", port)) == 0 def _qemu_console_tunnel(self): """Tunnel vers l'ÉCRAN QEMU d'une VM, pas vers un serveur de l'invité. diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index c20924a..dd91e31 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -2248,6 +2248,10 @@ TRANSLATIONS = { "fr": "(par le ProxyJump déjà dans ~/.ssh/config)", "en": "(through the ProxyJump already in ~/.ssh/config)", }, + "(the hypervisor only relays; -J puts the VM last)": { + "fr": "(l'hyperviseur ne fait que relayer ; -J met la VM en dernier)", + "en": "(the hypervisor only relays; -J puts the VM last)", + }, "No ~/.ssh/config entry; see SSH configuration.": { "fr": "Aucune entrée ~/.ssh/config ; voir Configuration SSH.", "en": "No ~/.ssh/config entry; see SSH configuration.", @@ -7822,6 +7826,79 @@ TRANSLATIONS = { "fr": "plus fluide, sans X11 :", "en": "smoother, without X11:", }, + "Android emulator": {"fr": "Émulateur Android", "en": "Android emulator"}, + "Android emulator (start, tunnel, scrcpy)": { + "fr": "Émulateur Android (démarrer, tunnel, scrcpy)", + "en": "Android emulator (start, tunnel, scrcpy)", + }, + "An emulator is already running on this VM.": { + "fr": "Un émulateur tourne déjà sur cette VM.", + "en": "An emulator is already running on this VM.", + }, + "Only one per AVD; close it first:": { + "fr": "Un seul par AVD ; le fermer d'abord :", + "en": "Only one per AVD; close it first:", + }, + "Close it now? (y/N): ": { + "fr": "Le fermer maintenant ? (o/N) : ", + "en": "Close it now? (y/N): ", + }, + "Closed.": {"fr": "Fermé.", "en": "Closed."}, + "Show a window?": { + "fr": "Afficher une fenêtre ?", + "en": "Show a window?", + }, + "No window - stream with scrcpy (smoother)": { + "fr": "Sans fenêtre — flux par scrcpy (plus fluide)", + "en": "No window - stream with scrcpy (smoother)", + }, + "Window over ssh -X (raw pixels, slower)": { + "fr": "Fenêtre par ssh -X (pixels bruts, plus lent)", + "en": "Window over ssh -X (raw pixels, slower)", + }, + "X11 compression is on (-XC); the screen is 540x1140.": { + "fr": "La compression X11 est active (-XC) ; l'écran fait 540x1140.", + "en": "X11 compression is on (-XC); the screen is 540x1140.", + }, + "Starting the emulator without a window...": { + "fr": "Démarrage de l'émulateur sans fenêtre…", + "en": "Starting the emulator without a window...", + }, + "Could not start it:": { + "fr": "Démarrage impossible :", + "en": "Could not start it:", + }, + "Started. Boot takes about a minute; log in the VM:": { + "fr": "Démarré. L'amorçage prend environ une minute ; journal dans la VM :", + "en": "Started. Boot takes about a minute; log in the VM:", + }, + "1. Emulator started, without a window.": { + "fr": "1. Émulateur démarré, sans fenêtre.", + "en": "1. Emulator started, without a window.", + }, + "If scrcpy will run on THIS machine, I can open it.": { + "fr": "Si scrcpy tournera sur CETTE machine, je peux l'ouvrir.", + "en": "If scrcpy will run on THIS machine, I can open it.", + }, + "Open the tunnel now? (y/N): ": { + "fr": "Ouvrir le tunnel maintenant ? (o/N) : ", + "en": "Open the tunnel now? (y/N): ", + }, + "Port already in use here:": { + "fr": "Port déjà occupé ici :", + "en": "Port already in use here:", + }, + "Close the other tunnel first:": { + "fr": "Fermer l'autre tunnel d'abord :", + "en": "Close the other tunnel first:", + }, + "Tunnel failed:": {"fr": "Tunnel en échec :", "en": "Tunnel failed:"}, + "Tunnel open on localhost:": { + "fr": "Tunnel ouvert sur localhost:", + "en": "Tunnel open on localhost:", + }, + "Then:": {"fr": "Puis :", "en": "Then:"}, + "To close it:": {"fr": "Pour le fermer :", "en": "To close it:"}, "Android emulator (adb 5555, then scrcpy)": { "fr": "Émulateur Android (adb 5555, puis scrcpy)", "en": "Android emulator (adb 5555, then scrcpy)", diff --git a/test/test_qemu_desktop_tools.py b/test/test_qemu_desktop_tools.py index d4df110..25b59c1 100644 --- a/test/test_qemu_desktop_tools.py +++ b/test/test_qemu_desktop_tools.py @@ -582,51 +582,17 @@ class TestPycharmFirstOpen(unittest.TestCase): self.assertEqual(0, res.returncode, res.stderr) -class TestScrcpyTunnel(unittest.TestCase): - """Tunnel adb pour scrcpy : la voie sans X11.""" +class TestAvdStep(unittest.TestCase): + """L'étape AVD, et ce qu'elle laisse comme piste dans le journal. + + Le tunnel adb lui-même est vérifié dans test_qemu_emulator_menu.py, qui + couvre aussi le démarrage sans fenêtre et la question de la fenêtre. + """ def setUp(self): self.todo = TODO.__new__(TODO) - def _printed(self, name, src): - import io - import contextlib - - buf = io.StringIO() - with contextlib.redirect_stdout(buf): - self.todo._qemu_scrcpy_tunnel(name, src) - return buf.getvalue() - - def test_it_tunnels_the_emulator_port_not_the_adb_server(self): - """5555 est l'adb de l'émulateur ; 5037 est le serveur adb, et le - tunneliser obligerait à tuer celui du poste, qui occupe le même port. - Vérifié dans la VM par « ss -ltn » : 5554, 5555 et 5037 écoutent tous - sur 127.0.0.1.""" - out = self._printed("saut+vm", "ssh_config") - self.assertIn("-L 5555:localhost:5555", out) - self.assertNotIn("5037", out) - - def test_it_starts_the_emulator_without_a_window(self): - """Tout l'intérêt : plus de X11 nulle part, le flux est du H.264.""" - out = self._printed("saut+vm", "ssh_config") - self.assertIn("-no-window", out) - self.assertNotIn("ssh -X", out) - - def test_it_gives_the_three_workstation_commands(self): - out = self._printed("saut+vm", "ssh_config") - self.assertIn("adb connect localhost:5555", out) - self.assertIn("scrcpy -s localhost:5555", out) - self.assertIn("apt install scrcpy", out) - - def test_it_uses_the_ssh_config_target_as_is(self): - """« localhost » est résolu par le DERNIER saut : le ProxyJump déjà - écrit dans ~/.ssh/config traverse les niveaux, on ne le réinvente pas. - """ - out = self._printed("test-vm_02+erplibre-mobile-proof", "ssh_config") - self.assertIn("test-vm_02+erplibre-mobile-proof", out) - self.assertIn("ProxyJump", out) - - def test_the_avd_step_points_at_it(self): + def test_the_avd_step_points_at_the_tunnel(self): """La question vient juste après « c'est trop lent » : la réponse doit être à portée de journal.""" self.assertIn("tunnel > 4", self.todo._qemu_avd_remote_cmd()) diff --git a/test/test_qemu_emulator_menu.py b/test/test_qemu_emulator_menu.py new file mode 100644 index 0000000..4e871cd --- /dev/null +++ b/test/test_qemu_emulator_menu.py @@ -0,0 +1,354 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Menu de l'émulateur Android : démarrage, fenêtre ou non, tunnel adb. + +Ce qui se vérifie ici sans VM tient à ce qui a réellement cassé sur la VM de +preuve : une commande privée de son chemin absolu (« emulator: command not +found », un ssh non interactif ne lisant aucun rc), un second émulateur sur le +même AVD (« Running multiple emulators »), une clé d'hôte refusée sur une IP +recyclée, et un tunnel visant une adresse que l'émulateur n'écoute pas. + +Le tunnel est le point délicat : l'émulateur n'écoute que sur le 127.0.0.1 de +la VM (« ss -ltn » dans l'invité ; l'hyperviseur reçoit un refus sur +IP_VM:5555). Une redirection vers l'IP de la VM ne peut donc PAS aboutir, et +seul un dernier saut dans la VM place « localhost » au bon endroit. +""" + +import socket +import subprocess +import sys +import unittest +from unittest import mock + +sys.argv = ["todo.py"] +from script.todo.todo import TODO # noqa: E402 + + +def _run_ok(stdout="", returncode=0, stderr=""): + return subprocess.CompletedProcess( + args=[], returncode=returncode, stdout=stdout, stderr=stderr + ) + + +class TestSshOptions(unittest.TestCase): + """La politique de clés d'hôte n'est PAS la même selon la provenance.""" + + def test_local_vm_tolerates_a_recycled_host_key(self): + """Une IP libvirt est réattribuée d'un déploiement au suivant : la clé + change sous la même adresse et ssh refuse — « Host key verification + failed », vécu dans ce menu même.""" + opts = TODO._qemu_ssh_opts("virsh") + self.assertIn("StrictHostKeyChecking=no", opts) + self.assertIn("UserKnownHostsFile=/dev/null", opts) + + def test_a_configured_host_keeps_its_own_key_policy(self): + """Un hôte de ~/.ssh/config appartient à l'utilisateur : sa clé est un + garde-fou, et le désarmer en son nom serait une décision volée.""" + opts = TODO._qemu_ssh_opts("ssh_config") + self.assertNotIn("StrictHostKeyChecking=no", opts) + self.assertNotIn("UserKnownHostsFile=/dev/null", opts) + self.assertIn("BatchMode=yes", opts) + + def test_both_refuse_to_hang_on_a_password_prompt(self): + for src in ("virsh", "ssh_config"): + self.assertIn("BatchMode=yes", TODO._qemu_ssh_opts(src), src) + + +class TestSshTarget(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + + def test_a_configured_host_is_named_as_is(self): + """C'est l'entrée ssh_config qui porte le ProxyJump : la réécrire à la + main reviendrait à deviner la chaîne de sauts.""" + self.todo._qemu_resolve_ips = lambda *a, **k: {} + self.assertEqual( + self.todo._qemu_ssh_target("test-vm_02+proof", "ssh_config"), + "test-vm_02+proof", + ) + + def test_a_local_vm_is_reached_by_ip_as_erplibre(self): + self.todo._qemu_resolve_ips = lambda names, labels=None: { + "vm-a": "192.168.123.81" + } + self.assertEqual( + self.todo._qemu_ssh_target("vm-a", "virsh"), + "erplibre@192.168.123.81", + ) + + def test_no_ip_yields_no_target_rather_than_a_broken_one(self): + """Une VM éteinte n'a pas d'IP. Rendre « erplibre@None » enverrait ssh + résoudre un nom absurde au lieu de le dire.""" + self.todo._qemu_resolve_ips = lambda names, labels=None: {} + self.assertEqual(self.todo._qemu_ssh_target("vm-a", "virsh"), "") + + +class TestEmulatorRunning(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + + def test_counts_what_pgrep_reports(self): + with mock.patch("subprocess.run", return_value=_run_ok("1\n")): + self.assertEqual(self.todo._qemu_emulator_running("h"), 1) + + def test_reads_the_last_line_so_a_warning_does_not_fool_it(self): + """ssh écrit ses avertissements sur stdout dans certains cas ; le compte + est la DERNIÈRE ligne.""" + with mock.patch( + "subprocess.run", return_value=_run_ok("Warning: added key\n0\n") + ): + self.assertEqual(self.todo._qemu_emulator_running("h"), 0) + + def test_an_unreachable_vm_is_unknown_not_zero(self): + """Zéro voudrait dire « libre » et autoriserait un second émulateur sur + le même AVD. L'inconnu se distingue donc du vide.""" + with mock.patch("subprocess.run", side_effect=OSError): + self.assertEqual(self.todo._qemu_emulator_running("h"), -1) + with mock.patch("subprocess.run", return_value=_run_ok("bavardage\n")): + self.assertEqual(self.todo._qemu_emulator_running("h"), -1) + + +class _MenuCase(unittest.TestCase): + """Socle commun : une VM locale, des réponses scriptées, aucun vrai ssh.""" + + def setUp(self): + self.todo = TODO.__new__(TODO) + self.todo._ssh_config_hosts = lambda: [] + self.todo._qemu_list_domains = lambda: ["vm-a"] + self.todo._qemu_resolve_ips = lambda names, labels=None: { + "vm-a": "192.168.123.81" + } + self.todo._qemu_self_address = staticmethod(lambda: ("10.0.0.2", True)) + self.calls = [] + + def _play(self, answers, running=0, start_rc=0, port_taken=False): + """Joue le menu avec des réponses données ; rend (sortie, commandes).""" + + def fake_run(cmd, *a, **k): + self.calls.append(cmd) + joined = " ".join(cmd) + if "pgrep -c qemu-system" in joined: + return _run_ok(f"{running}\n") + if "setsid" in joined: + return _run_ok(returncode=start_rc, stderr="boum") + return _run_ok() + + it = iter(answers) + with mock.patch("subprocess.run", side_effect=fake_run), mock.patch( + "builtins.input", lambda *a: next(it) + ), mock.patch.object( + TODO, "_port_in_use", staticmethod(lambda p: port_taken) + ), mock.patch( + "sys.stdout", new_callable=__import__("io").StringIO + ) as out: + self.todo._qemu_emulator_menu() + return out.getvalue(), self.calls + + @staticmethod + def _started(calls): + return [c for c in calls if any("setsid" in x for x in c)] + + @staticmethod + def _tunnels(calls): + return [c for c in calls if "-f" in c and "-N" in c] + + +class TestEmulatorMenu(_MenuCase): + def test_no_target_at_all_says_so_without_touching_ssh(self): + self.todo._qemu_list_domains = lambda: [] + out, calls = self._play([]) + self.assertIn("~/.ssh/config", out) + self.assertEqual(calls, []) + + def test_an_off_vm_is_reported_before_any_start(self): + self.todo._qemu_resolve_ips = lambda names, labels=None: {} + out, calls = self._play(["1"]) + self.assertIn("IP", out.upper()) + self.assertEqual(self._started(calls), []) + + def test_a_running_emulator_is_seen_and_refusing_stops_there(self): + """Deux émulateurs sur un même AVD, et le second meurt sur « Running + multiple emulators with the same AVD » — vécu deux fois. On le dit + AVANT, et un refus ne démarre rien.""" + out, calls = self._play(["1", "n"], running=1) + self.assertIn("pkill -f", out) + # « [q]emu » : la classe empêche le pkill de se trouver lui-même. + self.assertIn("[q]emu-system", out) + self.assertEqual(self._started(calls), []) + + def test_accepting_closes_the_other_one_then_starts(self): + out, calls = self._play(["1", "o", "1", "n"], running=1) + killed = [c for c in calls if any("pkill" in x for x in c)] + self.assertTrue(killed, "aucun pkill envoyé") + self.assertTrue(self._started(calls), "rien démarré après fermeture") + + def test_the_window_choice_is_delegated_to_the_workstation(self): + """L'écran appartient au poste : cette commande ne peut pas partir de + l'hyperviseur, qui n'a aucun affichage à lui prêter.""" + out, calls = self._play(["1", "2"]) + self.assertIn("ssh -XC", out) + self.assertEqual(self._started(calls), []) + self.assertEqual(self._tunnels(calls), []) + + def test_the_headless_start_carries_what_the_vm_needs(self): + out, calls = self._play(["1", "1", "n"]) + started = self._started(calls) + self.assertEqual(len(started), 1) + cmd = started[0][-1] + # Chemin absolu : « ssh hôte 'commande' » ne lit ni ~/.profile ni + # ~/.bashrc, et « emulator » seul rend « command not found » — vécu. + self.assertIn("$HOME/android/emulator/emulator", cmd) + self.assertIn("-no-window", cmd) + # sg kvm : sans le groupe, l'émulateur n'a pas /dev/kvm et renonce. + self.assertIn("sg kvm", cmd) + # setsid : il doit survivre à la fermeture de ce ssh. + self.assertIn("setsid -f", cmd) + self.assertIn("/tmp/erplibre-emulator.log", cmd) + + def test_the_start_command_is_valid_shell(self): + """Une apostrophe ou un guillemet de trop, et la VM répond par une + erreur de syntaxe — déjà rencontré dans ce même fichier.""" + _, calls = self._play(["1", "1", "n"]) + cmd = self._started(calls)[0][-1] + res = subprocess.run( + ["bash", "-n"], input=cmd, capture_output=True, text=True + ) + self.assertEqual(res.returncode, 0, res.stderr) + + def test_a_failed_start_does_not_offer_a_tunnel_to_nothing(self): + out, calls = self._play(["1", "1"], start_rc=1) + self.assertIn("boum", out) + self.assertEqual(self._tunnels(calls), []) + + def test_a_successful_start_chains_into_the_tunnel_help(self): + out, _ = self._play(["1", "1", "n"]) + self.assertIn("scrcpy", out) + self.assertIn("adb connect localhost:5555", out) + + +class TestScrcpyTunnel(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + self.todo._qemu_resolve_ips = lambda names, labels=None: { + "vm-a": "192.168.123.81" + } + self.todo._qemu_self_address = staticmethod(lambda: ("10.0.0.2", True)) + self.calls = [] + + def _play( + self, + src="virsh", + answers=("n",), + rc=0, + port_taken=False, + started=False, + name="vm-a", + ): + def fake_run(cmd, *a, **k): + self.calls.append(cmd) + return _run_ok(returncode=rc, stderr="refus") + + it = iter(answers) + with mock.patch("subprocess.run", side_effect=fake_run), mock.patch( + "builtins.input", lambda *a: next(it) + ), mock.patch.object( + TODO, "_port_in_use", staticmethod(lambda p: port_taken) + ), mock.patch.dict( + "os.environ", {"USER": "poste"} + ), mock.patch( + "sys.stdout", new_callable=__import__("io").StringIO + ) as out: + self.todo._qemu_scrcpy_tunnel(name, src, started=started) + return out.getvalue(), self.calls + + def test_a_local_vm_needs_the_vm_as_the_LAST_hop(self): + """L'émulateur n'écoute que sur le 127.0.0.1 de la VM. Une redirection + vers IP_VM:5555 depuis l'hyperviseur est refusée (mesuré) : « localhost » + ne vaut que sur le dernier saut, d'où -J.""" + out, _ = self._play(src="virsh") + self.assertIn("-L 5555:localhost:5555", out) + self.assertIn("-J poste@10.0.0.2", out) + self.assertIn("erplibre@192.168.123.81", out) + self.assertNotIn("-L 5555:192.168.123.81:5555", out) + + def test_a_configured_host_rides_its_own_proxyjump(self): + out, _ = self._play(src="ssh_config", name="test-vm_02+proof") + self.assertIn("-L 5555:localhost:5555 test-vm_02+proof", out) + self.assertNotIn("-J", out) + self.assertIn("ProxyJump", out) + + def test_it_gives_the_three_workstation_commands(self): + """Le tunnel ne sert à rien seul : la connexion adb et scrcpy le + suivent, et scrcpy n'est pas installé par défaut.""" + out, _ = self._play() + self.assertIn("adb connect localhost:5555", out) + self.assertIn("scrcpy -s localhost:5555", out) + self.assertIn("apt install scrcpy", out) + + def test_this_path_never_falls_back_to_x11(self): + """Tout l'intérêt : plus de X11 nulle part, le flux est du H.264.""" + out, _ = self._play() + self.assertNotIn("ssh -X", out) + + def test_the_tunneled_port_is_the_device_not_the_adb_server(self): + """5037 est le serveur adb du poste : le tunneler obligerait à tuer + celui de l'utilisateur, qui occupe le même port. 5555 est l'appareil. + """ + out, _ = self._play() + self.assertNotIn("5037", out) + self.assertIn("5555", out) + + def test_it_does_not_repeat_the_start_command_after_starting(self): + out, _ = self._play(started=True) + self.assertNotIn("-no-window", out) + out, _ = self._play(started=False) + self.assertIn("-no-window", out) + self.assertIn("$HOME/android/emulator/emulator", out) + + def test_declining_opens_nothing(self): + _, calls = self._play(answers=("n",)) + self.assertEqual(calls, []) + + def test_accepting_opens_a_detached_tunnel_that_fails_loudly(self): + """« -f » sans « ExitOnForwardFailure » rend 0 alors que la redirection + a échoué : un succès annoncé pour un tunnel absent.""" + out, calls = self._play(answers=("o",)) + self.assertEqual(len(calls), 1) + cmd = calls[0] + self.assertIn("-f", cmd) + self.assertIn("-N", cmd) + self.assertIn("ExitOnForwardFailure=yes", cmd) + self.assertIn("5555:localhost:5555", cmd) + self.assertIn("erplibre@192.168.123.81", cmd) + self.assertIn("✅", out) + + def test_an_occupied_port_is_named_instead_of_a_silent_bind_error(self): + """Le « bind: Address already in use » d'ssh se perd en mode détaché.""" + out, calls = self._play(answers=("o",), port_taken=True) + self.assertEqual(calls, []) + self.assertIn("5555", out) + self.assertIn("pkill", out) + + def test_a_refused_tunnel_is_reported(self): + out, _ = self._play(answers=("o",), rc=255) + self.assertIn("refus", out) + self.assertNotIn("✅", out) + + +class TestPortInUse(unittest.TestCase): + def test_a_listening_socket_is_seen(self): + with socket.socket() as srv: + srv.bind(("127.0.0.1", 0)) + srv.listen(1) + self.assertTrue(TODO._port_in_use(srv.getsockname()[1])) + + def test_a_closed_port_is_free(self): + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + self.assertFalse(TODO._port_in_use(port)) + + +if __name__ == "__main__": + unittest.main()