test_runtime_reglages.py
Rama main · réplica de lectura del repositorio original.
"""Non-régression de runtime.lire_reglages / ecrire_reglages / standard
(AUDIT-2026-08-13.md, section 28.32/28.34/28.36/28.37/28.39/28.56) :
cadences USB/cloud, port, horodatage, fuseau, activation jour/semaine/mois
du merge et activation du téléchargement automatique, réglables depuis la
page web plutôt que figés dans la constante STANDARD.
Anciennement test_runtime_cadences.py : renommé avec les fonctions, le
port, l'horodatage, le fuseau puis les coches de merge ayant rejoint les
cadences dans le même fichier de réglages."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from unittest import mock
import runtime
class TestsReglages(unittest.TestCase):
def setUp(self) -> None:
self.temporaire = tempfile.TemporaryDirectory(prefix="blink_reglages_")
self.dossier = Path(self.temporaire.name)
self.patch = mock.patch.object(runtime, "app_dir", return_value=self.dossier)
self.patch.start()
def tearDown(self) -> None:
self.patch.stop()
self.temporaire.cleanup()
def test_audio_setting_roundtrip_and_older_writer_preserve_selection(self):
parameters = dict(runtime.REGLAGES_DEFAUT)
parameters.update(camera_audio_enabled=True, camera_audio_track=1)
runtime.ecrire_reglages(**parameters)
stored = runtime.lire_reglages()
self.assertTrue(stored["camera_audio_enabled"])
self.assertEqual(stored["camera_audio_track"], 1)
parameters.pop("camera_audio_enabled")
parameters.pop("camera_audio_track")
runtime.ecrire_reglages(**parameters)
self.assertTrue(runtime.lire_reglages()["camera_audio_enabled"])
self.assertEqual(runtime.lire_reglages()["camera_audio_track"], 1)
runtime.ecrire_reglages(**parameters, camera_audio_enabled=False, camera_audio_track=0)
self.assertFalse(runtime.lire_reglages()["camera_audio_enabled"])
def test_fichier_absent_rend_les_valeurs_par_defaut(self):
self.assertEqual(runtime.lire_reglages(), dict(runtime.REGLAGES_DEFAUT))
def test_fichier_illisible_rend_les_valeurs_par_defaut(self):
(self.dossier / runtime.REGLAGES).write_text("{pas du json", encoding="utf-8")
self.assertEqual(runtime.lire_reglages(), dict(runtime.REGLAGES_DEFAUT))
def test_racine_json_non_objet_rend_les_valeurs_par_defaut(self):
"""Bug #10, revue de code du 0eab463 : `[]` est du JSON valide, mais
`.get()` dessus levait AttributeError avant ce correctif (reproduit
par la revue)."""
(self.dossier / runtime.REGLAGES).write_text("[]", encoding="utf-8")
self.assertEqual(runtime.lire_reglages(), dict(runtime.REGLAGES_DEFAUT))
(self.dossier / runtime.REGLAGES).write_text('"texte"', encoding="utf-8")
self.assertEqual(runtime.lire_reglages(), dict(runtime.REGLAGES_DEFAUT))
(self.dossier / runtime.REGLAGES).write_text("42", encoding="utf-8")
self.assertEqual(runtime.lire_reglages(), dict(runtime.REGLAGES_DEFAUT))
def test_chaine_a_la_place_d_un_booleen_retombe_sur_le_defaut(self):
"""Bug #10 : bool("false") vaut True en Python (chaîne non vide) -
l'inverse de l'intention derrière une valeur écrite ainsi à la main.
Retombe désormais sur le défaut plutôt que d'inverser en silence."""
(self.dossier / runtime.REGLAGES).write_text(
'{"timestamp": "false", "merge_jour": "oui"}', encoding="utf-8")
reglages = runtime.lire_reglages()
self.assertEqual(reglages["timestamp"], runtime.REGLAGES_DEFAUT["timestamp"])
self.assertEqual(reglages["merge_jour"], runtime.REGLAGES_DEFAUT["merge_jour"])
def test_arret_auto_du_direct_relu_jusqu_au_plafond_du_serveur(self):
"""Un 300 en dur ici, resté quand le plafond est passé à 420,
relisait comme 0 tout arrêt automatique réglé entre 301 et 420 s."""
for valeur in (301, 400, runtime.DIRECT_MAX_SECONDES):
with self.subTest(valeur=valeur):
(self.dossier / runtime.REGLAGES).write_text(
f'{{"live_auto_stop_seconds": {valeur}}}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["live_auto_stop_seconds"], valeur)
(self.dossier / runtime.REGLAGES).write_text(
f'{{"live_auto_stop_seconds": {runtime.DIRECT_MAX_SECONDES + 1}}}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["live_auto_stop_seconds"], 0)
def test_port_hors_plage_retombe_sur_le_defaut(self):
(self.dossier / runtime.REGLAGES).write_text('{"port": 99999}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["port"], runtime.REGLAGES_DEFAUT["port"])
(self.dossier / runtime.REGLAGES).write_text('{"port": 0}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["port"], runtime.REGLAGES_DEFAUT["port"])
def test_cadence_non_numerique_ou_negative_retombe_sur_le_defaut(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"usb_minutes": "dix"}', encoding="utf-8")
self.assertEqual(
runtime.lire_reglages()["usb_minutes"], runtime.REGLAGES_DEFAUT["usb_minutes"])
(self.dossier / runtime.REGLAGES).write_text(
'{"cloud_minutes": -5}', encoding="utf-8")
self.assertEqual(
runtime.lire_reglages()["cloud_minutes"], runtime.REGLAGES_DEFAUT["cloud_minutes"])
def test_ecriture_ne_laisse_aucun_temporaire(self):
"""Bug #10 : ecrire_reglages() est maintenant atomique (temporaire
propre a ce processus, puis replace), meme precaution que
blink_auth.save_session (I-02)."""
runtime.ecrire_reglages(usb_minutes=5, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=True,
merge_semaine=True, merge_mois=True, download_auto=True,
live_protocol="webrtc")
self.assertEqual(list(self.dossier.glob("*.part")), [])
def test_ecrire_puis_lire_conserve_les_valeurs(self):
runtime.ecrire_reglages(usb_minutes=7, cloud_minutes=2, port=8899, timestamp=False,
timezone="America/New_York", merge_jour=True,
merge_semaine=False, merge_mois=False, download_auto=False,
live_protocol="mse", font_size=40, font_color="yellow",
box_opacity=0.3, trusted_host="100.101.194.5",
webhook_notif_url="https://exemple.invalid/notif")
self.assertEqual(
runtime.lire_reglages(),
{"usb_minutes": 7, "cloud_minutes": 2, "port": 8899, "timestamp": False,
"timezone": "America/New_York", "merge_jour": True, "merge_semaine": False,
"merge_mois": False, "download_auto": False, "live_protocol": "mse",
"camera_audio_enabled": False, "camera_audio_track": 0,
"font_size": 40, "font_color": "yellow", "box_opacity": 0.3,
"trusted_host": "100.101.194.5",
"webhook_notif_url": "https://exemple.invalid/notif",
"live_auto_stop_seconds": 0, "clock": "24h", "dossier_sorties": "",
**{champ: 0 for champ in runtime.CONSERVATION},
"header_title": "", "header_icon": ""})
def test_valeurs_partielles_completees_par_les_defauts(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"usb_minutes": 15}', encoding="utf-8")
self.assertEqual(
runtime.lire_reglages(),
{"usb_minutes": 15,
"cloud_minutes": runtime.REGLAGES_DEFAUT["cloud_minutes"],
"port": runtime.REGLAGES_DEFAUT["port"],
"timestamp": runtime.REGLAGES_DEFAUT["timestamp"],
"timezone": runtime.REGLAGES_DEFAUT["timezone"],
"merge_jour": runtime.REGLAGES_DEFAUT["merge_jour"],
"merge_semaine": runtime.REGLAGES_DEFAUT["merge_semaine"],
"merge_mois": runtime.REGLAGES_DEFAUT["merge_mois"],
"download_auto": runtime.REGLAGES_DEFAUT["download_auto"],
"live_protocol": runtime.REGLAGES_DEFAUT["live_protocol"],
"camera_audio_enabled": runtime.REGLAGES_DEFAUT["camera_audio_enabled"],
"camera_audio_track": runtime.REGLAGES_DEFAUT["camera_audio_track"],
"font_size": runtime.REGLAGES_DEFAUT["font_size"],
"font_color": runtime.REGLAGES_DEFAUT["font_color"],
"box_opacity": runtime.REGLAGES_DEFAUT["box_opacity"],
"trusted_host": runtime.REGLAGES_DEFAUT["trusted_host"],
"webhook_notif_url": runtime.REGLAGES_DEFAUT["webhook_notif_url"],
"live_auto_stop_seconds": runtime.REGLAGES_DEFAUT["live_auto_stop_seconds"],
"clock": runtime.REGLAGES_DEFAUT["clock"],
"dossier_sorties": runtime.REGLAGES_DEFAUT["dossier_sorties"],
**{champ: 0 for champ in runtime.CONSERVATION},
"header_title": "", "header_icon": ""})
def test_fuseau_vide_dans_le_fichier_retombe_sur_le_defaut(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"timezone": ""}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["timezone"], runtime.REGLAGES_DEFAUT["timezone"])
def test_taille_police_absente_ou_explicitement_nulle_vaut_auto(self):
self.assertIsNone(runtime.lire_reglages()["font_size"])
(self.dossier / runtime.REGLAGES).write_text(
'{"font_size": null}', encoding="utf-8")
self.assertIsNone(runtime.lire_reglages()["font_size"])
def test_taille_police_hors_plage_ou_non_numerique_retombe_sur_auto(self):
(self.dossier / runtime.REGLAGES).write_text('{"font_size": 999}', encoding="utf-8")
self.assertIsNone(runtime.lire_reglages()["font_size"])
(self.dossier / runtime.REGLAGES).write_text('{"font_size": 0}', encoding="utf-8")
self.assertIsNone(runtime.lire_reglages()["font_size"])
(self.dossier / runtime.REGLAGES).write_text('{"font_size": "grand"}', encoding="utf-8")
self.assertIsNone(runtime.lire_reglages()["font_size"])
def test_taille_police_valide_est_conservee(self):
(self.dossier / runtime.REGLAGES).write_text('{"font_size": 40}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["font_size"], 40)
def test_couleur_vide_retombe_sur_le_defaut(self):
(self.dossier / runtime.REGLAGES).write_text('{"font_color": ""}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["font_color"], runtime.REGLAGES_DEFAUT["font_color"])
def test_hote_de_confiance_absent_vaut_desactive(self):
self.assertEqual(runtime.lire_reglages()["trusted_host"], "")
def test_hote_de_confiance_est_recorte(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"trusted_host": " 100.101.194.5 "}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["trusted_host"], "100.101.194.5")
def test_webhook_notif_url_absente_vaut_desactive(self):
self.assertEqual(runtime.lire_reglages()["webhook_notif_url"], "")
def test_webhook_notif_url_est_recortee(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"webhook_notif_url": " https://exemple.invalid/notif "}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["webhook_notif_url"],
"https://exemple.invalid/notif")
def test_opacite_bandeau_hors_plage_ou_non_numerique_retombe_sur_le_defaut(self):
for valeur in ('{"box_opacity": 2.0}', '{"box_opacity": -0.5}',
'{"box_opacity": "opaque"}'):
with self.subTest(valeur=valeur):
(self.dossier / runtime.REGLAGES).write_text(valeur, encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["box_opacity"],
runtime.REGLAGES_DEFAUT["box_opacity"])
def test_protocole_live_inconnu_retombe_sur_le_defaut(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"live_protocol": "mjpeg"}', encoding="utf-8")
self.assertEqual(
runtime.lire_reglages()["live_protocol"], runtime.REGLAGES_DEFAUT["live_protocol"])
def test_protocole_live_mse_est_respecte(self):
(self.dossier / runtime.REGLAGES).write_text(
'{"live_protocol": "mse"}', encoding="utf-8")
self.assertEqual(runtime.lire_reglages()["live_protocol"], "mse")
def test_standard_reprend_les_verbes_historiques_avec_reglages_par_defaut(self):
# Hebdo et mensuel sont désactivés par défaut (chacun ré-encode la
# même matière que le quotidien) : une installation neuve n'a que
# --no-weekly --no-monthly à ajouter ici, jamais les deux réglages.
# L'horodatage est lui aussi désactivé par défaut (ré-encodage ffmpeg
# lourd, plutôt qu'un copy rapide sans lui) : --no-timestamp aussi.
self.assertEqual(
runtime.standard(),
("serve", "--port", "8765", "--timezone", "Europe/Paris",
"--trusted-host", "",
"watch", "--loop", "10",
"download", "--from", "all", "--usb-loop", "10",
"--cloud-loop", "1",
"merge", "--loop", "5", "--timezone", "Europe/Paris",
"--no-timestamp", "--no-weekly", "--no-monthly"))
def test_standard_reflete_les_reglages_enregistres(self):
runtime.ecrire_reglages(usb_minutes=20, cloud_minutes=3, port=9090, timestamp=True,
timezone="Asia/Tokyo", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc",
trusted_host="100.101.194.5")
composition = runtime.standard()
self.assertEqual(
composition,
("serve", "--port", "9090", "--timezone", "Asia/Tokyo",
"--trusted-host", "100.101.194.5",
"watch", "--loop", "10",
"download", "--from", "all", "--usb-loop", "20",
"--cloud-loop", "3",
"merge", "--loop", "5", "--timezone", "Asia/Tokyo"))
def test_standard_n_a_qu_un_worker_de_telechargement_global(self):
groupes = runtime.decouper_verbes(runtime.standard())
downloads = [groupe for groupe in groupes if groupe[0] == "download"]
self.assertEqual(downloads, [[
"download", "--from", "all", "--usb-loop", "10",
"--cloud-loop", "1",
]])
def test_standard_ajoute_no_timestamp_quand_desactive(self):
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=False,
timezone="Europe/Paris", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc")
composition = runtime.standard()
self.assertEqual(composition[-6:],
("merge", "--loop", "5", "--timezone", "Europe/Paris",
"--no-timestamp"))
def test_standard_omet_merge_quand_journaliere_desactivee(self):
# Semaine et mois derivent des videos journalieres : les couper
# toutes les trois revient a ne plus lancer merge du tout, plutot
# que de le lancer pour rien.
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=False, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc")
composition = runtime.standard()
self.assertNotIn("merge", composition)
def test_standard_omet_download_quand_desactive(self):
# Coche pensee pour qui ne veut que le direct (voir BACKLOG.md) :
# le worker global "download" ne doit pas rester ; watch/serve/merge
# continuent de tourner normalement.
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=False, live_protocol="webrtc")
composition = runtime.standard()
self.assertNotIn("download", composition)
self.assertIn("serve", composition)
self.assertIn("watch", composition)
self.assertIn("merge", composition)
def test_standard_ajoute_no_weekly_et_no_monthly_selon_les_reglages(self):
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=True, merge_semaine=False,
merge_mois=False, download_auto=True, live_protocol="webrtc")
composition = runtime.standard()
self.assertEqual(composition[-7:],
("merge", "--loop", "5", "--timezone", "Europe/Paris",
"--no-weekly", "--no-monthly"))
def test_standard_ajoute_taille_couleur_et_opacite_quand_differentes_du_defaut(self):
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc",
font_size=40, font_color="yellow", box_opacity=0.3)
composition = runtime.standard()
self.assertEqual(composition[-11:],
("merge", "--loop", "5", "--timezone", "Europe/Paris",
"--font-size", "40", "--font-color", "yellow",
"--box-opacity", "0.3"))
def test_standard_n_ajoute_que_l_option_de_style_ecartee_du_defaut(self):
# Une seule des trois change : les deux autres, restées au défaut de
# merge_daily, ne doivent pas apparaître non plus.
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc",
font_size=40)
composition = runtime.standard()
self.assertEqual(composition[-7:],
("merge", "--loop", "5", "--timezone", "Europe/Paris",
"--font-size", "40"))
def test_standard_omet_le_style_quand_l_horodatage_est_desactive(self):
# --no-timestamp désactive tout l'habillage : pas la peine d'y
# ajouter une couleur ou une opacité qui ne serviront à rien.
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=False,
timezone="Europe/Paris", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc",
font_size=40, font_color="yellow", box_opacity=0.3)
composition = runtime.standard()
self.assertEqual(composition[-6:],
("merge", "--loop", "5", "--timezone", "Europe/Paris",
"--no-timestamp"))
def test_les_elements_du_bloc_serve_sont_fixes(self):
# blink_cli.py greffe le supplément de « start » juste après ce bloc
# (composition[:LONGUEUR_BLOC_SERVE]) : un --port, --timezone ou
# --trusted-host tapé à la main arrive donc après celui-ci et
# l'emporte (argparse retient la dernière occurrence).
composition = runtime.standard()
n = runtime.LONGUEUR_BLOC_SERVE
self.assertEqual(composition[:n],
("serve", "--port", "8765", "--timezone", "Europe/Paris",
"--trusted-host", ""))
self.assertEqual(composition[n], "watch")
def test_hote_de_confiance_transite_par_standard(self):
runtime.ecrire_reglages(usb_minutes=10, cloud_minutes=1, port=8765, timestamp=True,
timezone="Europe/Paris", merge_jour=True, merge_semaine=True,
merge_mois=True, download_auto=True, live_protocol="webrtc",
trusted_host="100.101.194.5")
composition = runtime.standard()
n = runtime.LONGUEUR_BLOC_SERVE
self.assertEqual(composition[:n],
("serve", "--port", "8765", "--timezone", "Europe/Paris",
"--trusted-host", "100.101.194.5"))
def test_un_port_explicite_dans_le_supplement_arrive_apres_le_defaut(self):
composition = runtime.standard()
n = runtime.LONGUEUR_BLOC_SERVE
supplement = ["--port", "9999"]
assemblee = [*composition[:n], *supplement, *composition[n:]]
indice_defaut = assemblee.index("8765")
indice_explicite = assemblee.index("9999")
self.assertLess(indice_defaut, indice_explicite,
"le port explicite doit venir apres le defaut pour l'emporter")
if __name__ == "__main__":
unittest.main()