test_dossier_sorties_absent.py
Rama main · réplica de lectura del repositorio original.
"""Issue #134 : un dossier des vidéos choisi sur un disque à part, encore
démonté, ne renaît pas sur le disque système.
Le cas décisif n'est pas la boucle de téléchargement mais la vignette : un seul
mkdir(parents=True) direct recréait le dossier, et tout repartait au mauvais
endroit. D'où le balayage du source en fin de fichier."""
from __future__ import annotations
import asyncio
import io
import os
import re
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
os.environ["BLINK_BOOTSTRAP"] = "none"
_TEST_HOME = tempfile.TemporaryDirectory(prefix="blink-sorties-absent-")
os.environ["BLINK_HOME"] = _TEST_HOME.name
import blink_engine # noqa: E402
import blink_registre # noqa: E402
import merge_daily # noqa: E402
import runtime # noqa: E402
import serve # noqa: E402
from test_runtime_dossier_stockage import BaseDossiers # noqa: E402
class BaseDisqueDemonte(BaseDossiers):
"""Un dossier choisi dans les réglages, dont le disque n'est pas monté."""
def setUp(self) -> None:
super().setUp()
self.disque = self.racine / "export" / "cam" / "blink2video"
runtime.ecrire_dossier_stockage(str(self.disque))
self.demonter()
def demonter(self) -> None:
self.disque.rmdir()
self.disque.parent.rmdir()
self.assertFalse(self.disque.parent.exists())
def monter(self) -> None:
self.disque.mkdir(parents=True)
class TestsDossierManquant(BaseDisqueDemonte):
def test_dossier_choisi_absent_signale(self):
self.assertEqual(runtime.dossier_sorties_manquant(), self.disque)
def test_dossier_revenu_n_est_plus_signale(self):
self.monter()
self.assertIsNone(runtime.dossier_sorties_manquant())
def test_dossier_par_defaut_jamais_signale(self):
runtime.ecrire_dossier_stockage("")
self.defaut_sorties.rmdir()
self.assertIsNone(runtime.dossier_sorties_manquant())
runtime.creer_sous_sorties(self.defaut_sorties / "Blink_Clips")
self.assertTrue((self.defaut_sorties / "Blink_Clips").is_dir())
class TestsCreationGardee(BaseDisqueDemonte):
def test_refuse_sans_rien_creer(self):
for cible in (self.disque, self.disque / ".blink_thumbs" / "cameras"):
with self.assertRaises(runtime.DossierSortiesAbsent) as contexte:
runtime.creer_sous_sorties(cible)
self.assertIn(str(self.disque), str(contexte.exception))
self.assertFalse(self.disque.parent.exists())
def test_est_un_oserror_comme_un_disque_plein(self):
self.assertTrue(issubclass(runtime.DossierSortiesAbsent, OSError))
def test_chemin_ailleurs_reste_libre(self):
ailleurs = self.racine / "autre" / "sortie"
runtime.creer_sous_sorties(ailleurs)
self.assertTrue(ailleurs.is_dir())
def test_nom_voisin_n_est_pas_sous_le_dossier(self):
voisin = self.disque.with_name(self.disque.name + "-bis")
runtime.creer_sous_sorties(voisin)
self.assertTrue(voisin.is_dir())
def test_disque_revenu_la_creation_reprend(self):
self.monter()
runtime.creer_sous_sorties(self.disque / "Blink_Clips")
self.assertTrue((self.disque / "Blink_Clips").is_dir())
def test_registre_ne_recree_pas_le_dossier(self):
with self.assertRaises(runtime.DossierSortiesAbsent):
blink_registre.save_download_state(
self.disque / "Blink_Clips", {"version": 2, "clips": {}})
self.assertFalse(self.disque.parent.exists())
class TestsReglages(BaseDisqueDemonte):
def test_changer_un_autre_reglage_ne_recree_pas_le_dossier(self):
serve._verifier_dossier_reglages(str(self.disque))
runtime.ecrire_dossier_stockage(
str(self.disque), reglages={**runtime.lire_reglages(), "port": 8899})
self.assertFalse(self.disque.parent.exists())
self.assertEqual(runtime.lire_reglages()["port"], 8899)
self.assertEqual(runtime.lire_reglages()["dossier_sorties"], str(self.disque))
def test_choisir_un_autre_dossier_le_cree(self):
autre = self.racine / "autre"
serve._verifier_dossier_reglages(str(autre))
runtime.ecrire_dossier_stockage(str(autre))
self.assertTrue(autre.is_dir())
self.assertIsNone(runtime.dossier_sorties_manquant())
def test_revenir_au_defaut_le_cree(self):
runtime.ecrire_dossier_stockage("")
self.assertTrue(self.defaut_sorties.is_dir())
class TestsPassage(BaseDisqueDemonte):
def args(self, commande: str = "download"):
return SimpleNamespace(command=commande,
output=self.disque / "Blink_Clips")
def passage(self, args, repetition: bool, **correctifs):
un_passage = mock.AsyncMock(return_value=0)
with mock.patch.object(blink_engine, "un_passage", un_passage), \
mock.patch("builtins.print") as sortie:
resultat = asyncio.run(blink_engine._faire_passage(
object(), args, [], repetition))
return resultat, un_passage, [str(a.args[0]) for a in sortie.call_args_list]
def test_passage_unique_echoue_et_le_dit(self):
resultat, un_passage, lignes = self.passage(self.args(), False)
self.assertEqual(resultat, blink_engine._ResultatPassage(1, False))
un_passage.assert_not_awaited()
self.assertTrue(any(str(self.disque) in ligne for ligne in lignes))
def test_list_n_a_pas_besoin_du_dossier(self):
resultat, un_passage, _ = self.passage(self.args("list"), False)
self.assertEqual(resultat, blink_engine._ResultatPassage(0, True))
un_passage.assert_awaited_once()
def test_boucle_attend_le_montage_puis_part_aussitot(self):
regards = []
async def dormir(_secondes):
regards.append(1)
if len(regards) == 3:
self.monter()
with mock.patch.object(blink_engine.asyncio, "sleep", dormir):
resultat, un_passage, lignes = self.passage(self.args(), True)
self.assertEqual(len(regards), 3)
self.assertEqual(resultat, blink_engine._ResultatPassage(0, True))
un_passage.assert_awaited_once()
# Trois regards, une seule ligne : pas de journal rempli toute la nuit.
self.assertEqual(sum(str(self.disque) in ligne for ligne in lignes), 1)
self.assertIn(blink_engine.msg("dossier_sorties_retrouve"), lignes)
def test_arret_demande_pendant_l_attente(self):
async def dormir(_secondes):
runtime_arret.return_value = True
with mock.patch.object(runtime, "arret_demande",
return_value=False) as runtime_arret, \
mock.patch.object(blink_engine.asyncio, "sleep", dormir):
resultat, un_passage, _ = self.passage(self.args(), True)
self.assertEqual(resultat, blink_engine._ResultatPassage(0, False))
un_passage.assert_not_awaited()
self.assertFalse(self.disque.parent.exists())
class TestsAssemblage(BaseDisqueDemonte):
def lancer(self, entree: Path) -> tuple:
args = SimpleNamespace(loop=None, input=entree)
with mock.patch.object(merge_daily, "parse_args", return_value=args), \
mock.patch.object(merge_daily, "_executer", return_value=0) as executer, \
mock.patch.object(merge_daily, "journal") as journal, \
mock.patch("builtins.print"):
return merge_daily.main(), executer, journal
def test_le_tour_est_saute_et_journalise(self):
code, executer, journal = self.lancer(self.disque / "Blink_Clips")
self.assertEqual(code, 1)
executer.assert_not_called()
self.assertIn(str(self.disque), journal.call_args.args[0])
self.assertFalse(self.disque.parent.exists())
def test_entree_explicite_ailleurs_reste_servie(self):
code, executer, _ = self.lancer(self.racine / "ailleurs")
self.assertEqual(code, 0)
executer.assert_called_once()
class TestsServe(BaseDisqueDemonte):
def test_sonde_rapide_nomme_le_dossier(self):
reponses = []
faux = SimpleNamespace(
path="/api/travail", hote_autorise=lambda: True,
jeton_valide=lambda: True,
send_json=lambda corps, *_args: reponses.append(corps))
serve.Handler.do_GET(faux)
self.assertEqual(reponses[-1]["dossier_absent"], str(self.disque))
self.monter()
serve.Handler.do_GET(faux)
self.assertEqual(reponses[-1]["dossier_absent"], "")
def test_vignette_affichee_sans_recreer_le_dossier(self):
handler = serve.Handler.__new__(serve.Handler)
handler.paths = {"thumbs": self.disque / ".blink_thumbs"}
handler.wfile = io.BytesIO()
handler.send_response = mock.Mock()
handler.send_header = mock.Mock()
handler.end_headers = mock.Mock()
with mock.patch.object(serve.BLINK, "call", return_value=b"image"):
handler.send_camera_thumb("jardin")
self.assertEqual(handler.wfile.getvalue(), b"image")
self.assertFalse(self.disque.parent.exists())
def test_bandeau_traduit_et_present_dans_la_page(self):
code = Path(serve.__file__).with_name("serve_app.js").read_text(encoding="utf-8")
self.assertEqual(code.count('"dossier.absent":'), 2)
self.assertEqual(code.count('$("dossierAbsent")'), 1)
self.assertIn('id="dossierAbsent"', serve.PAGE)
class TestsBalayage(unittest.TestCase):
MODULES = ("blink_engine", "blink_models", "blink_registre", "blink_webrtc",
"diagnostic_xr", "merge_daily", "serve")
def test_aucun_mkdir_direct_sous_les_sorties(self):
"""Tout ce que ces modules écrivent est sous dossier_sorties()."""
racine = Path(__file__).resolve().parent
trouves = {}
for module in self.MODULES:
texte = (racine / f"{module}.py").read_text(encoding="utf-8")
for ligne in texte.splitlines():
if re.search(r"\.mkdir\(|os\.makedirs\(", ligne):
trouves.setdefault(module, []).append(ligne.strip())
# Seule la voie du choix dans les réglages crée le dossier.
self.assertEqual(trouves, {
"serve": ["candidat.mkdir(parents=True, exist_ok=True)"]})
if __name__ == "__main__":
unittest.main()