erplibre/test/test_todo_sshfs.py

#!/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)