test_service_systemd.py
Rama main · réplica de lectura del repositorio original.
"""Non-régression de l'issue #35 (2026-09-27, MarkusKress) : sous le service
systemd utilisateur posé par « autostart on », une mise à jour lancée depuis
la page, comme Appliquer dans les réglages, laissait blink2video arrêté.
La seconde étape (finaliseur de « update », « restart --finaliser ») restait
dans le cgroup de l'unité malgré start_new_session ; quand « stop » faisait
sortir le processus principal, systemd tuait tout le cgroup avec lui, avant
toute relance.
Désormais la seconde étape se relance hors de l'unité (systemd-run --user
--scope), puis relance le service par « systemctl --user start ». Ces tests
simulent /proc/self/cgroup et les commandes : ils tournent sur tous les OS,
la preuve sur un vrai systemd restant l'expérience sur runner Ubuntu."""
from __future__ import annotations
import contextlib
import subprocess
import tempfile
import types
import unittest
from pathlib import Path
from unittest import mock
import autostart
import blink_cli
import maj
UNITE = "blink2video-start.service"
ENV = {"XDG_RUNTIME_DIR": "/run/user/1000", "PATH": "/usr/bin"}
def _cgroup(texte: str) -> Path:
fichier = Path(tempfile.mkdtemp()) / "cgroup"
fichier.write_text(texte, encoding="utf-8")
return fichier
class TestsUniteSystemd(unittest.TestCase):
def _unite(self, texte, plateforme="linux"):
with mock.patch.object(autostart.sys, "platform", plateforme):
return autostart.unite_systemd(_cgroup(texte))
def test_cgroup_v2_dans_l_unite(self):
self.assertEqual(self._unite(
"0::/user.slice/user-1000.slice/user@1000.service/app.slice/"
"blink2video-start.service\n"), UNITE)
def test_cgroup_v1_lignes_nommees(self):
self.assertEqual(self._unite(
"12:pids:/user.slice/user-1000.slice/user@1000.service/"
"blink2video-serve.service\n"
"1:name=systemd:/user.slice/user-1000.slice/user@1000.service/"
"blink2video-serve.service\n"), "blink2video-serve.service")
def test_hors_d_une_unite_blink2video(self):
for texte in ("0::/user.slice/user-1000.slice/session-3.scope\n",
"0::/user.slice/user-1000.slice/user@1000.service/app.slice/"
"autre.service\n",
"0::/user.slice/user-1000.slice/user@1000.service/app.slice/"
"run-r1234.scope\n"):
with self.subTest(texte=texte):
self.assertEqual(self._unite(texte), "")
def test_cgroup_illisible_ou_autre_os(self):
with mock.patch.object(autostart.sys, "platform", "linux"):
self.assertEqual(autostart.unite_systemd(Path(tempfile.mkdtemp()) / "absent"), "")
self.assertEqual(self._unite(
"0::/user.slice/blink2video-start.service\n", plateforme="darwin"), "")
class TestsSortirDuService(unittest.TestCase):
def _sortir(self, unite=UNITE, essai=0, environ=None, erreur=None):
lancer = mock.Mock(return_value=types.SimpleNamespace(returncode=essai))
if erreur is not None:
lancer.side_effect = erreur
with mock.patch.object(autostart, "unite_systemd", return_value=unite), \
mock.patch.object(autostart.runtime, "lancer", lancer), \
mock.patch.object(autostart.runtime, "demarrer") as demarrer:
resultat = autostart.sortir_du_service(
["blink2video", "restart", "--finaliser"], environ or dict(ENV))
return resultat, lancer, demarrer
def test_dans_l_unite_relance_hors_du_cgroup(self):
resultat, lancer, demarrer = self._sortir()
self.assertTrue(resultat)
# Un essai à vide d'abord : sans lui, un systemd-run en échec après
# notre départ ne laisserait personne pour finir le travail.
self.assertEqual(lancer.call_args[0][0][-1], "true")
commande = demarrer.call_args[0][0]
self.assertEqual(commande[:5], ["systemd-run", "--user", "--scope", "--quiet", "--"])
self.assertEqual(commande[5:], ["blink2video", "restart", "--finaliser"])
self.assertEqual(demarrer.call_args[1]["env"][autostart.UNITE_ENV], UNITE)
def test_hors_d_une_unite_rien_ne_change(self):
resultat, lancer, demarrer = self._sortir(unite="")
self.assertFalse(resultat)
lancer.assert_not_called()
demarrer.assert_not_called()
def test_deja_sorti_ne_boucle_pas(self):
resultat, _, demarrer = self._sortir(
environ=dict(ENV, **{autostart.UNITE_ENV: UNITE}))
self.assertFalse(resultat)
demarrer.assert_not_called()
def test_systemd_run_en_echec_ou_absent(self):
for options in ({"essai": 1}, {"erreur": FileNotFoundError("systemd-run")},
{"erreur": subprocess.TimeoutExpired("systemd-run", 30)}):
with self.subTest(**{k: str(v) for k, v in options.items()}):
resultat, _, demarrer = self._sortir(**options)
self.assertFalse(resultat)
demarrer.assert_not_called()
class TestsRelancerService(unittest.TestCase):
def _relancer(self, environ, code=0, erreur=None):
lancer = mock.Mock(return_value=types.SimpleNamespace(returncode=code))
if erreur is not None:
lancer.side_effect = erreur
with mock.patch.object(autostart.runtime, "lancer", lancer):
return autostart.relancer_service(environ), lancer
def test_relance_l_unite_quittee(self):
unite, lancer = self._relancer(dict(ENV, **{autostart.UNITE_ENV: UNITE}))
self.assertEqual(unite, UNITE)
self.assertEqual(lancer.call_args[0][0], ["systemctl", "--user", "start", UNITE])
def test_sans_unite_quittee_rien_a_relancer(self):
unite, lancer = self._relancer(dict(ENV))
self.assertEqual(unite, "")
lancer.assert_not_called()
def test_refus_de_systemctl(self):
for options in ({"code": 5}, {"erreur": FileNotFoundError("systemctl")}):
with self.subTest(**{k: str(v) for k, v in options.items()}):
unite, _ = self._relancer(dict(ENV, **{autostart.UNITE_ENV: UNITE}), **options)
self.assertEqual(unite, "")
class TestsBranchements(unittest.TestCase):
"""La seconde étape de « update » et de « restart » passe par ces deux
fonctions, dans le bon ordre."""
def test_finaliseur_de_mise_a_jour_sort_du_service_avant_tout(self):
cible = Path(tempfile.mkdtemp())
with mock.patch.object(autostart, "sortir_du_service", return_value=True) as sortir, \
mock.patch.object(maj, "_finaliser") as finaliser:
self.assertEqual(maj.finaliser(cible), 0)
finaliser.assert_not_called()
# Figé : « blink2video update --finaliser … » ; depuis les sources,
# self_command lance directement le programme du verbe (maj.py).
self.assertEqual(sortir.call_args[0][0][-2:], ["--finaliser", str(cible)])
def test_relance_de_la_mise_a_jour_par_le_service(self):
installe = Path(tempfile.mkdtemp())
with mock.patch.object(autostart, "relancer_service", return_value=UNITE), \
mock.patch.object(maj, "_relancer") as relancer:
maj._relancer_tout(installe, [[["start"]]])
relancer.assert_not_called()
with mock.patch.object(autostart, "relancer_service", return_value=""), \
mock.patch.object(maj, "_relancer") as relancer:
maj._relancer_tout(installe, [[["start"]], [["serve"]]])
self.assertEqual(relancer.call_count, 2)
def test_restart_sort_du_service_avant_d_arreter(self):
with mock.patch.object(autostart, "sortir_du_service", return_value=True) as sortir, \
mock.patch.object(blink_cli, "_arreter_instances") as arreter:
code = blink_cli.redemarrer(["--finaliser", "--delai", "0.75"])
self.assertEqual(code, 0)
arreter.assert_not_called()
self.assertEqual(sortir.call_args[0][0][-4:],
["restart", "--finaliser", "--delai", "0.75"])
def _second_temps(self, unite_relancee):
with mock.patch.object(autostart, "sortir_du_service", return_value=False), \
mock.patch.object(autostart, "relancer_service", return_value=unite_relancee), \
mock.patch.object(blink_cli, "_arreter_instances", return_value=0), \
mock.patch.object(blink_cli.runtime, "lire_instances", return_value=[]), \
mock.patch.object(blink_cli.runtime, "verrou_controle",
return_value=contextlib.nullcontext()), \
mock.patch.object(blink_cli.runtime, "demarrer") as demarrer:
self.assertEqual(blink_cli.redemarrer(["--finaliser"]), 0)
return demarrer
def test_restart_relance_par_le_service_sinon_detache(self):
self._second_temps(UNITE).assert_not_called()
demarrer = self._second_temps("")
demarrer.assert_called_once()
self.assertIn("start", demarrer.call_args[0][0])
if __name__ == "__main__":
unittest.main()