diff --git a/script/todo/todo.py b/script/todo/todo.py index ef32dc5..aac50f4 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -9442,6 +9442,214 @@ class TODO: pass print(f"\n {t('Tunnel closed.')}") + # sshfs lit « a+b » comme un CHAÎNAGE d'hôtes — « ssh a, puis ssh b depuis + # a » — et ne consulte donc PAS ~/.ssh/config pour l'alias entier. Or c'est + # todo.py qui nomme les VM découvertes « jump+domaine » (voir la marche + # SSH) : ce sont les alias les plus utiles, et les seuls que sshfs échoue à + # monter tel quel. Vécu : « read: Connection reset by peer », parce que la + # seconde moitié du nom est un domaine libvirt, pas un alias SSH du rebond. + SSHFS_CHAIN_SEP = "+" + + # Options à rendre à sshfs quand on contourne l'alias : exactement celles + # que todo.py écrit dans l'entrée qu'il génère. Sans elles, une VM dont la + # clé d'hôte a changé — IP DHCP réutilisée — ferait échouer le montage. + SSH_FORWARD_OPTS = ( + ("port", "Port"), + ("proxyjump", "ProxyJump"), + ("identityfile", "IdentityFile"), + ("identitiesonly", "IdentitiesOnly"), + ("stricthostkeychecking", "StrictHostKeyChecking"), + ("userknownhostsfile", "UserKnownHostsFile"), + ) + + # Ce que dit stderr, et ce qu'il faut aller corriger. L'ordre compte : le + # premier motif trouvé gagne. + SSH_FAILURE_HINTS = ( + ("could not resolve hostname", "unknown host name: check HostName"), + ("name or service not known", "unknown host name: check HostName"), + ("connection timed out", "no answer: is the server up and reachable?"), + ("operation timed out", "no answer: is the server up and reachable?"), + ("no route to host", "no route: check the network or the ProxyJump"), + ("connection refused", "nothing listening on the SSH port"), + ("permission denied", "authentication refused: check User and key"), + ("host key verification failed", "host key changed for this address"), + ) + + @staticmethod + def _ssh_config_entries(path): + """[(alias, {hostname, user})] de ~/.ssh/config, dans l'ordre du fichier. + + Pendant de `_ssh_config_hosts`, qui ne rend que les NOMS : ici le menu + de montage a besoin d'afficher aussi l'adresse et l'utilisateur. + + « Host a b » déclare DEUX alias pour la même machine — c'est ce que + todo.py écrit lui-même quand une VM porte plusieurs noms. Les prendre + pour un seul nom donnait un alias « a b », que sshfs ne peut pas + monter. Les motifs génériques (« * », « web-? ») sont écartés : ils ne + désignent aucune machine. + """ + hosts = [] + noms = [] + info = {} + + def clore(): + for nom in noms: + hosts.append((nom, dict(info))) + + try: + with open(path, encoding="utf-8", errors="replace") as fh: + lignes = fh.readlines() + except OSError: + return [] + for ligne in lignes: + ligne = ligne.strip() + if ligne.lower().startswith("host "): + clore() + noms = [ + m + for m in ligne.split()[1:] + if "*" not in m and "?" not in m and not m.startswith("!") + ] + info = {} + elif noms: + paire = ligne.split(None, 1) + if len(paire) == 2 and paire[0].lower() in ( + "hostname", + "user", + ): + info[paire[0].lower()] = paire[1].strip() + clore() + return hosts + + @staticmethod + def _ssh_resolve(alias): + """Configuration RÉSOLUE de l'alias, telle que ssh la voit (ssh -G). + + On délègue à ssh au lieu de relire le fichier : lui seul connaît les + Include, les Match, l'ordre des motifs et ses propres défauts. + """ + try: + res = subprocess.run( + ["ssh", "-G", alias], + capture_output=True, + text=True, + timeout=15, + env=TODO._qemu_c_env(), + ) + except (OSError, subprocess.SubprocessError): + return {} + if res.returncode != 0: + return {} + out = {} + for ligne in res.stdout.splitlines(): + cle, _, val = ligne.strip().partition(" ") + # ssh -G répète « identityfile » : la PREMIÈRE est celle qui compte. + if cle and val and cle.lower() not in out: + out[cle.lower()] = val + return out + + def _sshfs_command(self, alias, mount_point, resolved=None): + """(commande sshfs, alias contourné ?) pour monter cet alias. + + Sans « + » dans le nom, on laisse sshfs faire : c'est ssh qui lit la + config, et rien ne vaut mieux. Avec un « + », on résout l'alias + soi-même et on rend à sshfs une cible qu'il ne peut plus mal lire. + """ + base = "sshfs -o follow_symlinks" + if self.SSHFS_CHAIN_SEP not in alias: + return f"{base} {alias}:/ {mount_point}", False + cfg = resolved if resolved is not None else self._ssh_resolve(alias) + host = cfg.get("hostname") + # Un hostname qui contient encore un « + » ne réglerait rien, et un + # alias non résolu vaut mieux qu'une cible inventée. + if not host or self.SSHFS_CHAIN_SEP in host: + return f"{base} {alias}:/ {mount_point}", False + opts = [] + for cle, nom in self.SSH_FORWARD_OPTS: + val = cfg.get(cle) + if val and val.lower() != "none": + opts.append(f"-o {nom}={val}") + user = cfg.get("user") + cible = f"{user}@{host}" if user else host + pieces = [base] + opts + [f"{cible}:/", mount_point] + return " ".join(pieces), True + + @classmethod + def _ssh_failure_hint(cls, stderr): + """Première ligne utile de stderr, et ce qu'elle désigne.""" + texte = (stderr or "").lower() + for motif, indice in cls.SSH_FAILURE_HINTS: + if motif in texte: + return indice + return "" + + @staticmethod + def _ssh_probe(alias, timeout=8): + """(code, stderr) d'un « ssh true » sans invite de mot de passe. + + BatchMode : une invite bloquerait le menu. Un refus d'authentification + se distingue donc d'un hôte injoignable, et le diagnostic le dit. + """ + try: + res = subprocess.run( + [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + f"ConnectTimeout={timeout}", + alias, + "true", + ], + capture_output=True, + text=True, + timeout=timeout + 12, + env=TODO._qemu_c_env(), + ) + except subprocess.TimeoutExpired: + return 255, "Connection timed out" + except (OSError, subprocess.SubprocessError) as exc: + return 255, str(exc) + return res.returncode, res.stderr.strip() + + def _sshfs_diagnose(self, alias, mount_point, bypassed): + """Dit POURQUOI le montage a échoué, et où aller corriger. + + Le message est ciblé, pas une liste de causes possibles : on interroge + ssh, et selon qu'il passe ou non, le fautif n'est pas le même. + """ + print(f"\n ⚠ {t('sshfs mount failed.')}") + if not alias: + print(f" → {t('Check the SSH host and that the server is up.')}") + return + print(f" {t('Checking SSH access…')} ({alias})") + code, err = self._ssh_probe(alias) + if code == 0: + print(f" ✓ {t('SSH reaches this host: ~/.ssh/config is fine.')}") + if not bypassed and self.SSHFS_CHAIN_SEP in alias: + print(f" → {t('sshfs reads the « + » as host chaining.')}") + cmd, ok = self._sshfs_command(alias, mount_point) + if ok: + print(f" → {t('Run this instead:')}") + print(f" {cmd}") + else: + # Annoncer une commande puis n'en donner aucune serait + # pire que se taire : on dit ce qui manque. + print( + f" → {t('ssh -G resolved nothing: check ~/.ssh/config.')}" + ) + else: + print(f" → {t('Is sshfs (and fuse) installed here?')}") + return + indice = self._ssh_failure_hint(err) + if indice: + print(f" ✗ {t('SSH fails too:')} {t(indice)}") + else: + print( + f" ✗ {t('SSH fails too:')} {err.splitlines()[0] if err else code}" + ) + print(f" → {t('Update ~/.ssh/config, or check the server is up.')}") + def _configure_sshfs(self): import getpass import re @@ -9458,31 +9666,7 @@ class TODO: if choice == "2": ssh_config_path = os.path.expanduser("~/.ssh/config") - hosts = [] - if os.path.exists(ssh_config_path): - current_host = None - current_info = {} - with open(ssh_config_path) as f: - for line in f: - line = line.strip() - if line.lower().startswith("host "): - host_val = line.split(None, 1)[1].strip() - if host_val != "*": - if current_host: - hosts.append((current_host, current_info)) - current_host = host_val - current_info = {} - elif current_host: - key = line.split(None, 1) - if len(key) == 2: - k = key[0].lower() - v = key[1].strip() - if k == "hostname": - current_info["hostname"] = v - elif k == "user": - current_info["user"] = v - if current_host: - hosts.append((current_host, current_info)) + hosts = self._ssh_config_entries(ssh_config_path) if not hosts: print(t("No SSH hosts found in ~/.ssh/config")) @@ -9539,17 +9723,41 @@ class TODO: # une chaîne de symlinks relatifs profonds (.repo/projects -> project- # objects) que git ne peut pas traverser sur un montage sshfs par # défaut (« erreur à la lecture de .git » -> git status/commit KO). - cmd = f"sshfs -o follow_symlinks {target} {mount_point}" + # L'alias vient de ~/.ssh/config : c'est lui qui peut porter un « + », + # et lui qu'on peut interroger en cas d'échec. Une saisie manuelle est + # rendue telle quelle — si elle contient un « + », c'est un chaînage + # demandé exprès. + alias = ssh_name if choice == "2" else "" + if alias: + cmd, bypassed = self._sshfs_command(alias, mount_point) + else: + cmd, bypassed = ( + f"sshfs -o follow_symlinks {target} {mount_point}", + False, + ) print(f"{t('Mounting sshfs on: ')}{mount_point}") print(f"{t('Will execute:')} {cmd}") try: - self.execute.exec_command_live(cmd, source_erplibre=False) - print(f"{t('Mounted on: ')}{mount_point}") - print(f"mount | grep sshfs") - print(f"{t('To unmount: ')}" f"fusermount -u {mount_point}") - print(f"nautilus {mount_point}/home/{user}") + status = self.execute.exec_command_live(cmd, source_erplibre=False) except Exception as e: print(f"{t('Error mounting sshfs: ')}{e}") + status = 1 + # Le reste ne s'affiche QUE si le montage a réussi : « Monté sur … » + # après un code 1 envoyait chercher des fichiers dans un répertoire + # vide, et faisait passer l'échec pour un détail. + if status: + self._sshfs_diagnose(alias, mount_point, bypassed) + # Le point de montage n'a jamais servi : le laisser accumulerait + # un répertoire vide dans /tmp à chaque tentative. + try: + os.rmdir(mount_point) + except OSError: + pass + return + print(f"{t('Mounted on: ')}{mount_point}") + print("mount | grep sshfs") + print(f"{t('To unmount: ')}" f"fusermount -u {mount_point}") + print(f"nautilus {mount_point}/home/{user}") def _get_ssh_params(self): """Prompt for SSH connection parameters. Returns dict or None on cancel.""" diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 014a81d..fa00db3 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -3103,6 +3103,70 @@ TRANSLATIONS = { "fr": "Nœud de rendu", "en": "Render node", }, + "sshfs mount failed.": { + "fr": "Le montage sshfs a échoué.", + "en": "sshfs mount failed.", + }, + "Checking SSH access…": { + "fr": "Vérification de l'accès SSH…", + "en": "Checking SSH access…", + }, + "SSH reaches this host: ~/.ssh/config is fine.": { + "fr": "SSH joint cet hôte : ~/.ssh/config est bon.", + "en": "SSH reaches this host: ~/.ssh/config is fine.", + }, + "sshfs reads the « + » as host chaining.": { + "fr": "sshfs lit le « + » comme un chaînage d'hôtes.", + "en": "sshfs reads the « + » as host chaining.", + }, + "Run this instead:": { + "fr": "À lancer plutôt :", + "en": "Run this instead:", + }, + "ssh -G resolved nothing: check ~/.ssh/config.": { + "fr": "« ssh -G » n'a rien résolu : vérifier ~/.ssh/config.", + "en": "ssh -G resolved nothing: check ~/.ssh/config.", + }, + "Is sshfs (and fuse) installed here?": { + "fr": "sshfs (et fuse) sont-ils installés ici ?", + "en": "Is sshfs (and fuse) installed here?", + }, + "SSH fails too:": { + "fr": "SSH échoue aussi :", + "en": "SSH fails too:", + }, + "Update ~/.ssh/config, or check the server is up.": { + "fr": "Mettre à jour ~/.ssh/config, ou vérifier que le serveur répond.", + "en": "Update ~/.ssh/config, or check the server is up.", + }, + "Check the SSH host and that the server is up.": { + "fr": "Vérifier l'hôte SSH saisi et que le serveur répond.", + "en": "Check the SSH host and that the server is up.", + }, + "unknown host name: check HostName": { + "fr": "nom d'hôte inconnu : vérifier HostName", + "en": "unknown host name: check HostName", + }, + "no answer: is the server up and reachable?": { + "fr": "aucune réponse : le serveur est-il allumé et joignable ?", + "en": "no answer: is the server up and reachable?", + }, + "no route: check the network or the ProxyJump": { + "fr": "pas de route : vérifier le réseau ou le ProxyJump", + "en": "no route: check the network or the ProxyJump", + }, + "nothing listening on the SSH port": { + "fr": "rien n'écoute sur le port SSH", + "en": "nothing listening on the SSH port", + }, + "authentication refused: check User and key": { + "fr": "authentification refusée : vérifier User et la clé", + "en": "authentication refused: check User and key", + }, + "host key changed for this address": { + "fr": "la clé d'hôte a changé pour cette adresse", + "en": "host key changed for this address", + }, "no statistics yet": { "fr": "pas encore de statistiques", "en": "no statistics yet", diff --git a/test/test_todo_sshfs.py b/test/test_todo_sshfs.py new file mode 100644 index 0000000..934e109 --- /dev/null +++ b/test/test_todo_sshfs.py @@ -0,0 +1,389 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Montage sshfs : ne rien annoncer qui n'ait eu lieu, et dire pourquoi. + +Le symptôme rapporté : « read: Connection reset by peer », « code : 1 », et +juste après « Monté sur /tmp/sshfs_… », la commande pour démonter et celle +pour ouvrir le répertoire dans un explorateur. Le montage n'avait pas eu lieu. + +La cause, elle, est plus profonde : sshfs lit « a+b » comme un CHAÎNAGE +d'hôtes — « ssh a, puis ssh b depuis a » — et ne consulte donc jamais +~/.ssh/config pour l'alias entier. Or c'est todo.py qui nomme les VM +découvertes « rebond+domaine », et la seconde moitié de ce nom est un domaine +libvirt, pas un alias SSH du rebond. Ces alias-là, les plus utiles, étaient +les seuls que sshfs ne pouvait pas monter. + +Ce que ces tests gardent : + +- Aucune ligne de succès après un code non nul. +- Le point de montage inutilisé est retiré, pas laissé dans /tmp. +- Un alias à « + » est résolu par ssh lui-même (« ssh -G ») et rendu à sshfs + sous une forme qu'il ne peut plus mal lire, options comprises — sans + StrictHostKeyChecking, une VM à l'IP recyclée échouerait sur sa clé d'hôte. +- « Host a b » déclare DEUX alias : c'est ce que le générateur du dépôt écrit. +""" + +import contextlib +import io +import os +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +sys.argv = ["todo.py"] +from script.todo.todo import TODO # noqa: E402 +from script.todo.todo_i18n import t # noqa: E402 + +# Entrée réelle écrite par todo.py pour une VM derrière un rebond. +CONFIG = """Host * + ServerAliveInterval 60 + +Host novipro_private + HostName 192.168.100.110 + User mathben + +Host novipro_private+ERPLibre01 + HostName 192.168.122.50 + User mathben + StrictHostKeyChecking no + UserKnownHostsFile /dev/null + IdentityFile /home/erplibre/.ssh/id_ed25519 + IdentitiesOnly yes + ProxyJump novipro_private + +Host erplibre-ubuntu-2604 erplibre-2604-bis + HostName 192.168.123.165 + User erplibre + +Host web-? + User www +""" + +# Sortie de « ssh -G » pour l'alias à « + », réduite à ce qui compte. +SSH_G = """host novipro_private+erplibre01 +hostname 192.168.122.50 +user mathben +port 22 +proxyjump novipro_private +identityfile /home/erplibre/.ssh/id_ed25519 +identityfile ~/.ssh/id_rsa +identitiesonly yes +stricthostkeychecking false +userknownhostsfile /dev/null +""" + + +def _config(texte=CONFIG): + tmp = tempfile.NamedTemporaryFile( + "w", suffix=".config", delete=False, encoding="utf-8" + ) + tmp.write(texte) + tmp.close() + return tmp.name + + +class TestLectureConfig(unittest.TestCase): + def setUp(self): + self.chemin = _config() + self.addCleanup(os.unlink, self.chemin) + + def test_it_reads_hosts_in_file_order(self): + hosts = TODO._ssh_config_entries(self.chemin) + noms = [n for n, _i in hosts] + self.assertEqual("novipro_private", noms[0]) + self.assertIn("novipro_private+ERPLibre01", noms) + + def test_a_host_line_with_two_patterns_gives_two_aliases(self): + """C'est ce que le générateur du dépôt écrit (« Host {' '.join(names)} »). + Les prendre pour un seul nom donnait l'alias « a b », que sshfs ne peut + pas monter — et qui n'existe pour personne.""" + hosts = dict(TODO._ssh_config_entries(self.chemin)) + self.assertIn("erplibre-ubuntu-2604", hosts) + self.assertIn("erplibre-2604-bis", hosts) + self.assertEqual( + "192.168.123.165", hosts["erplibre-ubuntu-2604"]["hostname"] + ) + self.assertEqual( + hosts["erplibre-ubuntu-2604"], hosts["erplibre-2604-bis"] + ) + + def test_wildcard_patterns_are_left_out(self): + """« Host * » et « Host web-? » ne désignent aucune machine : les + proposer dans un menu de montage n'a pas de sens.""" + noms = [n for n, _i in TODO._ssh_config_entries(self.chemin)] + self.assertNotIn("*", noms) + self.assertNotIn("web-?", noms) + + def test_the_plus_alias_stays_one_name(self): + noms = [n for n, _i in TODO._ssh_config_entries(self.chemin)] + self.assertNotIn("novipro_private", noms[1:2] and []) + self.assertIn("novipro_private+ERPLibre01", noms) + + def test_hostname_and_user_are_kept(self): + hosts = dict(TODO._ssh_config_entries(self.chemin)) + info = hosts["novipro_private+ERPLibre01"] + self.assertEqual("192.168.122.50", info["hostname"]) + self.assertEqual("mathben", info["user"]) + + def test_a_missing_file_is_not_a_crash(self): + self.assertEqual([], TODO._ssh_config_entries("/nexistepas/config")) + + +class TestResolution(unittest.TestCase): + def test_it_reads_ssh_dash_g(self): + with mock.patch( + "subprocess.run", + return_value=subprocess.CompletedProcess([], 0, SSH_G, ""), + ): + cfg = TODO._ssh_resolve("novipro_private+ERPLibre01") + self.assertEqual("192.168.122.50", cfg["hostname"]) + self.assertEqual("novipro_private", cfg["proxyjump"]) + + def test_the_first_identityfile_wins(self): + """ssh -G les répète toutes ; la première est celle qu'il essaiera.""" + with mock.patch( + "subprocess.run", + return_value=subprocess.CompletedProcess([], 0, SSH_G, ""), + ): + cfg = TODO._ssh_resolve("x") + self.assertEqual("/home/erplibre/.ssh/id_ed25519", cfg["identityfile"]) + + def test_a_failing_ssh_resolves_to_nothing(self): + with mock.patch( + "subprocess.run", + return_value=subprocess.CompletedProcess([], 255, "", "bad"), + ): + self.assertEqual({}, TODO._ssh_resolve("x")) + with mock.patch("subprocess.run", side_effect=OSError): + self.assertEqual({}, TODO._ssh_resolve("x")) + + +class TestCommandeSshfs(unittest.TestCase): + def setUp(self): + # Méthode d'instance : elle interroge self._ssh_resolve quand aucune + # résolution ne lui est fournie. + self.todo = TODO.__new__(TODO) + self.todo._ssh_resolve = lambda alias: {} + + def test_an_alias_without_plus_is_handed_over_untouched(self): + """Sans « + », c'est ssh qui lit la config, et rien ne vaut mieux.""" + cmd, contourne = self.todo._sshfs_command("vm-a", "/tmp/mnt") + self.assertEqual("sshfs -o follow_symlinks vm-a:/ /tmp/mnt", cmd) + self.assertFalse(contourne) + + def _resolue(self): + return { + "hostname": "192.168.122.50", + "user": "mathben", + "port": "22", + "proxyjump": "novipro_private", + "identityfile": "/home/erplibre/.ssh/id_ed25519", + "identitiesonly": "yes", + "stricthostkeychecking": "false", + "userknownhostsfile": "/dev/null", + } + + def test_a_plus_alias_becomes_a_resolved_target(self): + cmd, contourne = self.todo._sshfs_command( + "novipro_private+ERPLibre01", "/tmp/mnt", self._resolue() + ) + self.assertTrue(contourne) + self.assertIn("mathben@192.168.122.50:/", cmd) + self.assertNotIn("+", cmd) + + def test_the_options_that_matter_travel_with_it(self): + """Sans ProxyJump, la VM est injoignable ; sans StrictHostKeyChecking, + une IP DHCP recyclée fait échouer le montage sur sa clé d'hôte.""" + cmd, _ = self.todo._sshfs_command("a+b", "/tmp/mnt", self._resolue()) + for attendu in ( + "-o ProxyJump=novipro_private", + "-o Port=22", + "-o IdentityFile=/home/erplibre/.ssh/id_ed25519", + "-o IdentitiesOnly=yes", + "-o StrictHostKeyChecking=false", + "-o UserKnownHostsFile=/dev/null", + ): + self.assertIn(attendu, cmd) + + def test_options_ssh_reports_as_none_are_not_forwarded(self): + """« ssh -G » écrit « proxyjump none » quand il n'y en a pas ; le + transmettre ferait échouer ssh sur une valeur qu'il vient d'inventer. + """ + cfg = dict(self._resolue(), proxyjump="none") + cmd, _ = self.todo._sshfs_command("a+b", "/tmp/mnt", cfg) + self.assertNotIn("ProxyJump", cmd) + + def test_an_unresolvable_alias_is_left_alone(self): + """Mieux vaut la commande d'origine, qui échouera en le disant, qu'une + cible inventée qui monterait la mauvaise machine.""" + cmd, contourne = self.todo._sshfs_command("a+b", "/tmp/mnt", {}) + self.assertEqual("sshfs -o follow_symlinks a+b:/ /tmp/mnt", cmd) + self.assertFalse(contourne) + + def test_the_mount_point_stays_last(self): + """La syntaxe de sshfs : cible puis point de montage. Une option glissée + après monterait ailleurs.""" + cmd, _ = self.todo._sshfs_command("a+b", "/tmp/mnt", self._resolue()) + self.assertTrue(cmd.endswith(" /tmp/mnt"), cmd) + + +class TestDiagnostic(unittest.TestCase): + def test_each_kind_of_failure_names_its_culprit(self): + cas = ( + ( + "ssh: Could not resolve hostname zz: Name or service not known", + "unknown host name: check HostName", + ), + ( + "ssh: connect to host x port 22: Connection timed out", + "no answer: is the server up and reachable?", + ), + ( + "ssh: connect to host x port 22: No route to host", + "no route: check the network or the ProxyJump", + ), + ( + "ssh: connect to host x port 22: Connection refused", + "nothing listening on the SSH port", + ), + ( + "mathben@x: Permission denied (publickey).", + "authentication refused: check User and key", + ), + ( + "Host key verification failed.", + "host key changed for this address", + ), + ) + for stderr, attendu in cas: + self.assertEqual(attendu, TODO._ssh_failure_hint(stderr), stderr) + + def test_an_unknown_error_is_not_guessed(self): + """Rien à dire plutôt qu'un diagnostic inventé : la ligne brute de ssh + sera affichée telle quelle.""" + self.assertEqual("", TODO._ssh_failure_hint("sshfs: fuse: bidule")) + self.assertEqual("", TODO._ssh_failure_hint("")) + + +class TestFlux(unittest.TestCase): + """Le parcours complet, exécuteur bouchonné.""" + + def _todo(self, code, probe=(0, "")): + todo = TODO.__new__(TODO) + todo.lances = [] + exe = mock.Mock() + + def lance(cmd, **kw): + todo.lances.append(cmd) + return code + + exe.exec_command_live = lance + todo.execute = exe + todo._ssh_probe = lambda alias, timeout=8: probe + todo._ssh_resolve = lambda alias: { + "hostname": "192.168.122.50", + "user": "mathben", + "proxyjump": "novipro_private", + } + return todo + + def _joue(self, todo, config, selection="2"): + chemin = _config(config) + self.addCleanup(os.unlink, chemin) + reponses = iter(["2", selection]) + crees = [] + vrai_makedirs = os.makedirs + vrai_rmdir = os.rmdir + + def makedirs(path, **kw): + crees.append(path) + return vrai_makedirs(path, **kw) + + retires = [] + + def rmdir(path): + retires.append(path) + return vrai_rmdir(path) + + out = io.StringIO() + with mock.patch( + "builtins.input", lambda *a: next(reponses, "") + ), mock.patch( + "os.path.expanduser", + lambda p: chemin if p.endswith("config") else p, + ), mock.patch( + "os.makedirs", makedirs + ), mock.patch( + "os.rmdir", rmdir + ): + with contextlib.redirect_stdout(out): + todo._configure_sshfs() + return out.getvalue(), crees, retires + + def test_a_failed_mount_announces_nothing_mounted(self): + """Le cœur du problème rapporté : « Monté sur … » après un code 1.""" + todo = self._todo(1) + sortie, _crees, _retires = self._joue(todo, CONFIG, "2") + self.assertNotIn(t("Mounted on: "), sortie) + self.assertNotIn("fusermount", sortie) + self.assertNotIn("nautilus", sortie) + self.assertIn(t("sshfs mount failed."), sortie) + + def test_a_failed_mount_leaves_no_empty_directory(self): + """Une tentative par jour pendant un mois laissait trente répertoires + vides dans /tmp.""" + todo = self._todo(1) + _s, crees, retires = self._joue(todo, CONFIG, "2") + self.assertEqual(crees, retires) + for chemin in retires: + self.assertFalse(os.path.exists(chemin)) + + def test_a_successful_mount_says_how_to_unmount(self): + todo = self._todo(0) + sortie, crees, retires = self._joue(todo, CONFIG, "2") + self.assertIn(t("Mounted on: "), sortie) + self.assertIn("fusermount -u", sortie) + self.assertEqual([], retires) + for chemin in crees: + self.addCleanup(lambda p=chemin: os.path.isdir(p) and os.rmdir(p)) + + def test_ssh_working_points_at_the_plus_not_at_the_network(self): + """Quand ssh joint l'hôte, ce n'est pas le réseau : c'est sshfs qui a + mal lu l'alias. Le message doit envoyer là, et donner la commande.""" + todo = self._todo(1, probe=(0, "")) + # L'alias à « + » est le troisième de la config (après le rebond seul). + sortie, _c, _r = self._joue(todo, CONFIG, "2") + self.assertIn( + t("SSH reaches this host: ~/.ssh/config is fine."), sortie + ) + + def test_ssh_failing_sends_to_the_config_or_the_server(self): + todo = self._todo( + 1, + probe=( + 255, + "ssh: connect to host x port 22: Connection timed out", + ), + ) + sortie, _c, _r = self._joue(todo, CONFIG, "2") + self.assertIn(t("no answer: is the server up and reachable?"), sortie) + self.assertIn( + t("Update ~/.ssh/config, or check the server is up."), sortie + ) + + def test_the_plus_alias_is_bypassed_before_being_run(self): + """Le vrai correctif : la commande lancée ne contient plus le « + ».""" + todo = self._todo(0) + # [1] novipro_private · [2] novipro_private+ERPLibre01 · [3] la VM + _s, _c, _r = self._joue(todo, CONFIG, "2") + lancee = todo.lances[0] + self.assertIn("mathben@192.168.122.50:/", lancee) + self.assertIn("-o ProxyJump=novipro_private", lancee) + self.assertNotIn("+ERPLibre01", lancee) + + +if __name__ == "__main__": + unittest.main(verbosity=1)