[ADD] script todo: lancer l'émulateur Android et son tunnel adb
Le tunnel n'était que du texte à recopier, et le démarrage restait manuel :
un second émulateur sur le même AVD tue le premier (« Running multiple
emulators »), rencontré deux fois. Le menu le détecte donc avant de lancer.
La fenêtre décide du reste. Sans elle, todo.py démarre à distance, détaché
par setsid dans le groupe kvm, et enchaîne sur scrcpy ; avec elle, la
commande doit partir du poste qui possède l'affichage, pas d'ici.
Mesuré : l'émulateur n'écoute que sur le 127.0.0.1 de la VM — l'hyperviseur
est refusé sur IP:5555 — d'où le -J qui met la VM en dernier saut. Vérifié
sur erplibre-mobile-proof : poignée de main adb CNXN rendant
sdk_gphone64_x86_64, enveloppe détachée survivant au ssh ; 30 tests.
--- EN ---
The tunnel was only text to copy, and starting the emulator stayed manual:
a second emulator on the same AVD kills the first ("Running multiple
emulators"), hit twice. The menu now detects it before starting.
The window decides the rest. Without one, todo.py starts it remotely,
detached by setsid in the kvm group, and chains into scrcpy; with one, the
command must run from the workstation that owns the display, not from here.
Measured: the emulator only listens on the VM's 127.0.0.1 — the hypervisor
is refused on IP:5555 — hence the -J putting the VM last. Verified on
erplibre-mobile-proof: an adb CNXN handshake returning sdk_gphone64_x86_64,
a detached wrapper outliving the ssh; 30 tests.
Assisted-by: Claude Opus 5
This commit is contained in:
parent
7a65568098
commit
13e8b1bd2d
4 changed files with 664 additions and 57 deletions
|
|
@ -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é.
|
||||
|
|
|
|||
|
|
@ -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)",
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
354
test/test_qemu_emulator_menu.py
Normal file
354
test/test_qemu_emulator_menu.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue