"""Caractérise la validation HTTP des réglages, sans stockage ni réseau réels.""" import io import json import os import tempfile import unittest from unittest import mock _IMPORT_HOME = tempfile.TemporaryDirectory(prefix="blink-reglages-validation-import-") with mock.patch.dict(os.environ, {"BLINK_BOOTSTRAP": "none", "BLINK_HOME": _IMPORT_HOME.name}): import serve ERREUR_NOMBRES = ( "Les cadences doivent être des nombres de minutes d'au " "moins 1, et le port un nombre entre 1 et 65535.") DEFAUTS = { "usb_minutes": 5, "cloud_minutes": 10, "port": 8765, "timestamp": False, "timezone": "UTC", "merge_jour": True, "merge_semaine": True, "merge_mois": True, "download_auto": True, "live_protocol": "webrtc", "camera_audio_enabled": False, "camera_audio_track": 0, "font_size": None, "font_color": "white", "box_opacity": 0.55, "trusted_host": "", "webhook_notif_url": "", "live_auto_stop_seconds": 0, "clock": "24h", "conservation_clips": 0, "conservation_jour": 0, "conservation_semaine": 0, "conservation_mois": 0, "conservation_directs": 0, "conservation_photos": 0, "header_title": "", "header_icon": "", } CHAMPS_BOOLEENS = ( "timestamp", "merge_jour", "merge_semaine", "merge_mois", "download_auto") class TestsValidationReglagesHttp(unittest.TestCase): def setUp(self): settings = mock.patch.object(serve.runtime, "lire_reglages", return_value=dict(serve.runtime.REGLAGES_DEFAUT)) settings.start() self.addCleanup(settings.stop) self.observations = [] def remplacer(objet, nom, **options): patch = mock.patch.object(objet, nom, **options) remplacement = patch.start() self.addCleanup(patch.stop) self.observations.append(remplacement) return remplacement self.stockage = remplacer(serve.runtime, "ecrire_dossier_stockage") self.verrou = remplacer(serve.runtime, "verrou_configuration") # La sonde fait partie du comportement testé, mais reste entièrement # simulée, y compris lorsque le payload désigne un chemin arbitraire. self.chemin = remplacer(serve, "Path") self.dossier = self.chemin.return_value.expanduser.return_value self.creer_sonde = remplacer(serve.tempfile, "NamedTemporaryFile") self.contexte_sonde = self.creer_sonde.return_value self.sonde = self.contexte_sonde.__enter__.return_value self.contexte_sonde.__exit__.return_value = False self.interdits = [] for objet, nom in ( (serve.runtime, "lancer"), (serve.runtime, "demarrer"), (serve.BLINK, "call"), (serve.threading, "Thread"), ): self.interdits.append(remplacer( objet, nom, side_effect=AssertionError("Effet réel interdit : " + nom))) def tearDown(self): for interdit in self.interdits: interdit.assert_not_called() def payload(self, **changements): return {"usb_minutes": 5, "cloud_minutes": 10, "port": 8765, "timezone": "UTC", **changements} def requete(self, payload): for observation in self.observations: observation.reset_mock() corps = json.dumps(payload, ensure_ascii=False).encode("utf-8") handler = object.__new__(serve.Handler) handler.path = "/api/reglages" handler.paths = {} handler.initial_setup = False handler.client_address = ("127.0.0.1", 12345) handler.headers = { "Host": "127.0.0.1:8765", "X-Blink-Token": serve.TOKEN, "Content-Length": str(len(corps)), } handler.rfile = io.BytesIO(corps) handler.send_json = mock.Mock() handler.send_error = mock.Mock() handler.repondre_puis_redemarrer = mock.Mock() handler.do_POST() handler.send_error.assert_not_called() return handler def assert_enregistre(self, handler, reglages, dossier=""): self.verrou.assert_called_once_with() self.stockage.assert_called_once_with( dossier, reglages=reglages, configuration_initiale=False) handler.repondre_puis_redemarrer.assert_called_once_with(["restart"]) handler.send_json.assert_not_called() def assert_refuse(self, handler, message): handler.send_json.assert_called_once_with({"error": message}, 400) self.verrou.assert_not_called() self.stockage.assert_not_called() handler.repondre_puis_redemarrer.assert_not_called() def test_en_tete_nettoye_borne_et_garde_pour_une_page_ancienne(self): # Issue #40 : titre et icône de l'en-tête. Contrôles retirés, bords # nettoyés, liant des émojis composés (U+200D) gardé. famille = chr(0x200D).join(["\U0001F468", "\U0001F469", "\U0001F467"]) self.assert_enregistre( self.requete(self.payload(header_title=" Maison\nde Marc ", header_icon=famille)), dict(DEFAUTS, header_title="Maisonde Marc", header_icon=famille)) # Une page restée ouverte depuis une version sans ce réglage ne # l'envoie pas : celui enregistré reste. actuel = dict(serve.runtime.REGLAGES_DEFAUT, header_title="Garage", header_icon="🚗") with mock.patch.object(serve.runtime, "lire_reglages", return_value=actuel): self.assert_enregistre(self.requete(self.payload()), dict(DEFAUTS, header_title="Garage", header_icon="🚗")) message = "Le titre de l'en-tête tient en 60 caractères au plus, l'icône en 16." self.assert_refuse(self.requete(self.payload(header_title="x" * 61)), message) self.assert_refuse(self.requete(self.payload(header_icon="x" * 17)), message) # Autre chose qu'un texte : vide, comme un réglage absent du fichier. self.assert_enregistre(self.requete(self.payload(header_title=42)), DEFAUTS) def test_audio_settings_are_validated_and_preserved_for_older_clients(self): current = dict(serve.runtime.REGLAGES_DEFAUT, camera_audio_enabled=True, camera_audio_track=1) with mock.patch.object(serve.runtime, "lire_reglages", return_value=current): self.assert_enregistre(self.requete(self.payload()), dict(DEFAUTS, camera_audio_enabled=True, camera_audio_track=1)) self.assert_enregistre(self.requete(self.payload(camera_audio_enabled=True, camera_audio_track=1)), dict(DEFAUTS, camera_audio_enabled=True, camera_audio_track=1)) for invalid in ("false", 1, None): with self.subTest(enabled=invalid): self.assert_refuse(self.requete(self.payload(camera_audio_enabled=invalid)), "camera_audio_enabled : booléen attendu") for invalid in (-1, 2, True, "1", 0.5, None): with self.subTest(track=invalid): self.assert_refuse(self.requete(self.payload(camera_audio_track=invalid)), "camera_audio_track : canal 0 ou 1 attendu") def test_horloge_validee_et_conservee_pour_une_page_ancienne(self): self.assert_enregistre(self.requete(self.payload(clock="12h")), dict(DEFAUTS, clock="12h")) actuel = dict(serve.runtime.REGLAGES_DEFAUT, clock="12h") with mock.patch.object(serve.runtime, "lire_reglages", return_value=actuel): self.assert_enregistre(self.requete(self.payload()), dict(DEFAUTS, clock="12h")) for invalide in ("12", "am/pm", "", None, 12): with self.subTest(clock=invalide): self.assert_refuse(self.requete(self.payload(clock=invalide)), f"Format d'heure inconnu : « {invalide} » (24h ou 12h).") def test_valeurs_par_defaut_et_absence_de_sonde_sans_dossier(self): handler = self.requete(self.payload()) self.assert_enregistre(handler, DEFAUTS) self.chemin.assert_not_called() def test_normalisation_des_nombres_booleens_et_textes(self): handler = self.requete(self.payload( usb_minutes=" 3 ", cloud_minutes=12.9, port=" 65535 ", storage_dir=" archives locales ", timezone=" UTC \t", timestamp="false", merge_jour=0, merge_semaine=None, merge_mois="", download_auto=[False], live_protocol="mse")) self.assert_enregistre(handler, { "usb_minutes": 3, "cloud_minutes": 12, "port": 65535, "timestamp": True, "timezone": "UTC", "merge_jour": False, "merge_semaine": False, "merge_mois": False, "download_auto": True, "live_protocol": "mse", "camera_audio_enabled": False, "camera_audio_track": 0, "font_size": None, "font_color": "white", "box_opacity": 0.55, "trusted_host": "", "webhook_notif_url": "", "live_auto_stop_seconds": 0, "clock": "24h", "conservation_clips": 0, "conservation_jour": 0, "conservation_semaine": 0, "conservation_mois": 0, "conservation_directs": 0, "conservation_photos": 0, "header_title": "", "header_icon": "", }, "archives locales") self.chemin.assert_called_once_with("archives locales") def test_booleens_conservent_la_conversion_bool_du_payload(self): for valeur, attendu in ( (False, False), (None, False), (0, False), ("", False), ([], False), ({}, False), (True, True), (1, True), ("false", True), ([0], True), ({"actif": False}, True), ): with self.subTest(valeur=valeur): champs = dict.fromkeys(CHAMPS_BOOLEENS, valeur) handler = self.requete(self.payload(**champs)) self.assert_enregistre( handler, {**DEFAUTS, **dict.fromkeys(CHAMPS_BOOLEENS, attendu)}) def test_bornes_et_conversions_numeriques_acceptees(self): for valeurs, attendues in ( ((1, 1, 1), (1, 1, 1)), ((1000000, 1000000, 65535), (1000000, 1000000, 65535)), ((True, True, True), (1, 1, 1)), ((" 2 ", "3", "4"), (2, 3, 4)), ((2.9, 3.9, 4.9), (2, 3, 4)), ): with self.subTest(valeurs=valeurs): champs = ("usb_minutes", "cloud_minutes", "port") handler = self.requete(self.payload(**dict(zip(champs, valeurs)))) self.assert_enregistre(handler, {**DEFAUTS, **dict(zip(champs, attendues))}) def test_nombres_absents_ou_invalides_refuses_avant_sonde(self): for champ in ("usb_minutes", "cloud_minutes", "port"): for valeur in (None, "", "invalide", [], {}, 0, -1, False, 0.9): with self.subTest(champ=champ, valeur=valeur): handler = self.requete(self.payload( **{champ: valeur, "storage_dir": "archives"})) self.assert_refuse(handler, ERREUR_NOMBRES) self.chemin.assert_not_called() with self.subTest(champ_absent=champ): payload = self.payload() del payload[champ] self.assert_refuse(self.requete(payload), ERREUR_NOMBRES) def test_port_superieur_a_65535_refuse(self): for valeur in (65536, "65536", 99999999): with self.subTest(valeur=valeur): self.assert_refuse(self.requete(self.payload(port=valeur)), ERREUR_NOMBRES) def test_debordement_numerique_ne_devient_pas_un_refus_de_formulaire(self): # La refactorisation ne doit pas élargir les exceptions absorbées par # le validateur : int(inf) lève historiquement une OverflowError. for champ in ("usb_minutes", "cloud_minutes", "port"): with self.subTest(champ=champ): with self.assertRaises(OverflowError): self.requete(self.payload(**{champ: float("inf")})) self.chemin.assert_not_called() self.verrou.assert_not_called() self.stockage.assert_not_called() def test_erreurs_inattendues_de_sonde_restent_propagees(self): # Seules les erreurs d'accès OSError sont des refus 400 connus. Une # exception métier dédiée ne doit pas capturer tout ValueError. for erreur in (ValueError("chemin invalide"), RuntimeError("profil absent")): with self.subTest(erreur=type(erreur).__name__): self.chemin.return_value.expanduser.side_effect = erreur with self.assertRaises(type(erreur)) as capture: self.requete(self.payload(storage_dir="archives")) self.assertIs(capture.exception, erreur) self.dossier.mkdir.assert_not_called() self.verrou.assert_not_called() self.stockage.assert_not_called() def test_dossier_vide_ou_blanc_ne_declenche_pas_de_sonde(self): for valeur in ("", " \t \n"): with self.subTest(valeur=valeur): handler = self.requete(self.payload(storage_dir=valeur)) self.assert_enregistre(handler, DEFAUTS) self.chemin.assert_not_called() def test_dossier_converti_en_texte_avant_sonde(self): for valeur, attendu in ((None, "None"), (123, "123")): with self.subTest(valeur=valeur): handler = self.requete(self.payload(storage_dir=valeur)) self.assert_enregistre(handler, DEFAUTS, attendu) self.chemin.assert_called_once_with(attendu) def test_sonde_cree_ecrit_puis_efface_avant_enregistrement(self): operations = mock.Mock() operations.attach_mock(self.dossier.mkdir, "mkdir") operations.attach_mock(self.creer_sonde, "creer_sonde") operations.attach_mock(self.sonde.write, "write") operations.attach_mock(self.sonde.flush, "flush") operations.attach_mock(self.contexte_sonde.__exit__, "fermer_et_effacer") operations.attach_mock(self.stockage, "enregistrer") handler = self.requete(self.payload(storage_dir="archives")) self.assert_enregistre(handler, DEFAUTS, "archives") self.chemin.return_value.expanduser.assert_called_once_with() self.assertEqual(operations.mock_calls, [ mock.call.mkdir(parents=True, exist_ok=True), mock.call.creer_sonde(mode="w", encoding="utf-8", prefix=".blink_ecriture_test-", dir=self.dossier), mock.call.creer_sonde().__enter__(), mock.call.write("blink2video"), mock.call.flush(), mock.call.fermer_et_effacer(None, None, None), mock.call.enregistrer("archives", reglages=DEFAUTS, configuration_initiale=False), ]) def test_erreur_creation_dossier_renvoie_400_sans_ecriture_sonde(self): self.dossier.mkdir.side_effect = OSError("création refusée") handler = self.requete(self.payload(storage_dir="archives")) self.assert_refuse(handler, "Dossier de stockage inaccessible : création refusée") self.creer_sonde.assert_not_called() def test_erreur_creation_sonde_renvoie_400_sans_enregistrement(self): self.creer_sonde.side_effect = OSError("création sonde refusée") handler = self.requete(self.payload(storage_dir="archives")) self.assert_refuse(handler, "Dossier de stockage inaccessible : création sonde refusée") self.sonde.write.assert_not_called() self.contexte_sonde.__exit__.assert_not_called() def test_erreur_ecriture_sonde_renvoie_400_et_nettoie(self): self.sonde.write.side_effect = OSError("écriture refusée") handler = self.requete(self.payload(storage_dir="archives")) self.assert_refuse(handler, "Dossier de stockage inaccessible : écriture refusée") self.dossier.mkdir.assert_called_once_with(parents=True, exist_ok=True) self.contexte_sonde.__exit__.assert_called_once() def test_erreur_vidage_sonde_renvoie_400_et_nettoie(self): self.sonde.flush.side_effect = OSError("disque plein") handler = self.requete(self.payload(storage_dir="archives")) self.assert_refuse(handler, "Dossier de stockage inaccessible : disque plein") self.contexte_sonde.__exit__.assert_called_once() def test_erreur_effacement_sonde_renvoie_400_sans_enregistrement(self): self.contexte_sonde.__exit__.side_effect = OSError("effacement refusé") handler = self.requete(self.payload(storage_dir="archives")) self.assert_refuse(handler, "Dossier de stockage inaccessible : effacement refusé") self.sonde.write.assert_called_once_with("blink2video") def test_fuseaux_absents_invalides_ou_vides_refuses(self): for valeur, texte in ( ("", ""), (" ", ""), (None, "None"), (" Zone/Inexistante ", "Zone/Inexistante"), ("../UTC", "../UTC"), ): with self.subTest(valeur=valeur): self.assert_refuse(self.requete(self.payload(timezone=valeur)), f"Fuseau horaire inconnu : « {texte} ».") payload = self.payload() del payload["timezone"] self.assert_refuse(self.requete(payload), "Fuseau horaire inconnu : « ».") def test_protocoles_webrtc_et_mse_acceptes(self): for protocole in ("webrtc", "mse"): with self.subTest(protocole=protocole): handler = self.requete(self.payload(live_protocol=protocole)) self.assert_enregistre(handler, {**DEFAUTS, "live_protocol": protocole}) def test_protocoles_invalides_refuses_sans_normalisation(self): for valeur, texte in ( (None, "None"), ("", ""), ("MSE", "MSE"), ("WebRTC", "WebRTC"), (" mse ", " mse "), ("baseline", "baseline"), (123, "123"), ): with self.subTest(valeur=valeur): self.assert_refuse(self.requete(self.payload(live_protocol=valeur)), f"Protocole de direct inconnu : « {texte} ».") def test_taille_couleur_opacite_par_defaut_quand_absentes(self): handler = self.requete(self.payload()) self.assert_enregistre(handler, DEFAUTS) def test_normalisation_taille_couleur_opacite(self): handler = self.requete(self.payload( font_size="40", font_color=" yellow ", box_opacity="0.3")) self.assert_enregistre(handler, { **DEFAUTS, "font_size": 40, "font_color": "yellow", "box_opacity": 0.3, }) def test_taille_police_absente_nulle_ou_vide_vaut_auto(self): handler = self.requete(self.payload()) self.assert_enregistre(handler, DEFAUTS) for valeur in (None, "", " "): with self.subTest(valeur=valeur): handler = self.requete(self.payload(font_size=valeur)) self.assert_enregistre(handler, DEFAUTS) def test_taille_police_hors_plage_ou_invalide_refusee(self): message = ("La taille de police doit être un nombre entre 8 et 500, " "ou vide pour la taille automatique.") for valeur in (7, 501, 0, -1, "grande", [], {}): with self.subTest(valeur=valeur): self.assert_refuse(self.requete(self.payload(font_size=valeur)), message) def test_couleur_vide_ou_blanche_retombe_sur_white(self): for valeur in ("", " "): with self.subTest(valeur=valeur): handler = self.requete(self.payload(font_color=valeur)) self.assert_enregistre(handler, DEFAUTS) def test_couleur_invalide_refusee(self): for valeur in ("red:evil", "not a color!", "white@2.0", "white@-0.1", "rouge,vert"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(font_color=valeur)), f"Couleur d'horodatage invalide : « {valeur} ».") def test_couleur_hex_et_avec_opacite_acceptees(self): for valeur in ("#FF0000", "0xFF0000", "white@0.8", "yellow"): with self.subTest(valeur=valeur): handler = self.requete(self.payload(font_color=valeur)) self.assert_enregistre(handler, {**DEFAUTS, "font_color": valeur}) def test_opacite_bandeau_hors_plage_ou_invalide_refusee(self): message = "L'opacité du bandeau doit être un nombre entre 0.0 et 1.0." for valeur in (-0.1, 1.1, "opaque", [], {}): with self.subTest(valeur=valeur): self.assert_refuse(self.requete(self.payload(box_opacity=valeur)), message) def test_opacite_bandeau_bornes_incluses_acceptees(self): for valeur in (0.0, 1.0): with self.subTest(valeur=valeur): handler = self.requete(self.payload(box_opacity=valeur)) self.assert_enregistre(handler, {**DEFAUTS, "box_opacity": valeur}) def test_hote_de_confiance_absent_ou_vide_vaut_desactive(self): for valeur in (None, "", " "): with self.subTest(valeur=valeur): payload = self.payload() if valeur is not None: payload["trusted_host"] = valeur handler = self.requete(payload) self.assert_enregistre(handler, DEFAUTS) def test_hote_de_confiance_valide_est_recorte(self): handler = self.requete(self.payload(trusted_host=" 100.101.194.5 ")) self.assert_enregistre(handler, {**DEFAUTS, "trusted_host": "100.101.194.5"}) def _message_hote_invalide(self, entree): return (f"Hôte de confiance invalide : « {entree} ». Un ou plusieurs " "noms d'hôte, adresses IP ou sous-réseaux CIDR (192.168.1.0/24), " "séparés par des virgules.") def test_hote_de_confiance_avec_espace_refuse(self): for valeur in ("hote invalide", "100 .101.194.5"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(trusted_host=valeur)), self._message_hote_invalide(valeur.strip())) def test_hote_de_confiance_caractere_interdit_refuse(self): # Chaque entrée finit aussi dans l'en-tête CSP (frame-ancestors) : un # « ; » y ouvrirait une directive de plus. for valeur in ("hote;script-src", "hote'x", "hôte"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(trusted_host=valeur)), self._message_hote_invalide(valeur)) def test_hote_de_confiance_commencant_par_un_tiret_refuse(self): # argparse lirait « --trusted-host -x » comme une option sans valeur : # enregistré tel quel, serve refusait de démarrer au lancement suivant. for valeur in ("-x", "--loop", "192.168.1.10,-mon_pc"): with self.subTest(valeur=valeur): entree = valeur.split(",")[-1] self.assert_refuse( self.requete(self.payload(trusted_host=valeur)), self._message_hote_invalide(entree)) def test_hote_de_confiance_liste_normalisee_sans_espaces_ni_vides(self): # Issue #13 : plusieurs noms, adresses ou sous-réseaux à la fois. handler = self.requete(self.payload( trusted_host=" 192.168.1.10 , 192.168.1.0/24,,mon_pc, ")) self.assert_enregistre( handler, {**DEFAUTS, "trusted_host": "192.168.1.10,192.168.1.0/24,mon_pc"}) def test_hote_de_confiance_liste_refusee_sur_la_premiere_entree_invalide(self): self.assert_refuse( self.requete(self.payload(trusted_host="192.168.1.10, 10.0.0.0/99")), "Sous-réseau invalide : « 10.0.0.0/99 ». Format attendu : 192.168.1.0/24.") def test_hote_de_confiance_avec_barre_mais_pas_un_cidr_refuse(self): for valeur in ("100.101.194.5/etc", "http://100.101.194.5"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(trusted_host=valeur)), f"Sous-réseau invalide : « {valeur.strip()} ». Format " "attendu : 192.168.1.0/24.") def test_hote_de_confiance_cidr_valide_est_accepte(self): # Issue #10 : un client Windows en DHCP n'a pas d'adresse fixe a # mettre dans un champ IP unique - accepter un sous-reseau entier. # strict=False cote hote_autorise() tolere aussi l'IP d'une machine # du reseau (192.168.1.5/24) plutot que l'adresse reseau elle-meme, # mais la valeur est enregistree telle quelle, sans normalisation. for valeur in ("192.168.1.0/24", "192.168.1.5/24"): with self.subTest(valeur=valeur): handler = self.requete(self.payload(trusted_host=valeur)) self.assert_enregistre(handler, {**DEFAUTS, "trusted_host": valeur}) def test_webhook_notif_url_absent_ou_vide_vaut_desactive(self): for valeur in (None, "", " "): with self.subTest(valeur=valeur): payload = self.payload() if valeur is not None: payload["webhook_notif_url"] = valeur handler = self.requete(payload) self.assert_enregistre(handler, DEFAUTS) def test_webhook_notif_url_valide_est_recortee(self): handler = self.requete( self.payload(webhook_notif_url=" https://exemple.invalid/notif ")) self.assert_enregistre( handler, {**DEFAUTS, "webhook_notif_url": "https://exemple.invalid/notif"}) def test_webhook_notif_url_sans_schema_http_refusee(self): for valeur in ("ftp://exemple.invalid/notif", "exemple.invalid/notif", "non une url", # urlparse lève ValueError sur celle-ci (IPv6 mal fermé) # plutôt que de rendre un schéma vide comme les autres : # doit rester un refus propre, pas une exception non # rattrapée qui remonterait jusqu'à do_POST. "http://[invalid"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(webhook_notif_url=valeur)), f"URL de notification invalide : « {valeur} ». Une " "adresse http:// ou https:// complète, ou vide pour désactiver.") def test_arret_auto_direct_absent_vaut_desactive(self): for valeur in (None, "", 0): with self.subTest(valeur=valeur): payload = self.payload() if valeur is not None: payload["live_auto_stop_seconds"] = valeur handler = self.requete(payload) self.assert_enregistre(handler, DEFAUTS) def test_arret_auto_direct_valide_est_converti_en_entier(self): handler = self.requete(self.payload(live_auto_stop_seconds="120")) self.assert_enregistre(handler, {**DEFAUTS, "live_auto_stop_seconds": 120}) def test_arret_auto_direct_hors_plage_ou_invalide_refuse(self): # Au-delà de LIVE_MAX_SECONDS le réglage serait inopérant : le serveur # coupe de toute façon chaque direct à ce plafond (issue #15). message = ("L'arrêt automatique du direct doit être un nombre de secondes " f"entre 0 (désactivé) et {serve.LIVE_MAX_SECONDS} (le plafond " "de tout direct).") for valeur in (-1, serve.LIVE_MAX_SECONDS + 1, 86400, "beaucoup"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(live_auto_stop_seconds=valeur)), message) def test_conservation_hors_plage_refusee(self): maximum = serve.runtime.CONSERVATION_MAX_JOURS message = serve.msg("conservation_invalide", maximum=maximum) for valeur in (-1, maximum + 1, "longtemps"): with self.subTest(valeur=valeur): self.assert_refuse( self.requete(self.payload(conservation_photos=valeur)), message) def test_conservation_dans_le_desordre_refusee(self): """Issue #120 : une journalière gardée moins longtemps que ses bruts serait refaite à chaque passage.""" message = serve.msg("conservation_ordre", niveau=serve.msg("libelle_conservation_jour"), source=serve.msg("libelle_conservation_clips")) for clips in (0, 90): with self.subTest(clips=clips): self.assert_refuse(self.requete(self.payload( conservation_clips=clips, conservation_jour=30)), message) def test_conservation_dans_l_ordre_enregistree(self): valeurs = {"conservation_clips": 30, "conservation_jour": 90, "conservation_mois": 365, "conservation_photos": 7} handler = self.requete(self.payload(**valeurs)) self.assert_enregistre(handler, {**DEFAUTS, **valeurs}) def test_arret_auto_direct_bornes_incluses_acceptees(self): for valeur in (0, serve.LIVE_MAX_SECONDS): with self.subTest(valeur=valeur): handler = self.requete(self.payload(live_auto_stop_seconds=valeur)) self.assert_enregistre(handler, {**DEFAUTS, "live_auto_stop_seconds": valeur}) def test_ordre_des_refus_nombres_dossier_fuseau_puis_protocole(self): payload = self.payload(usb_minutes=0, storage_dir="archives", timezone="Zone/Inexistante", live_protocol="inconnu") self.dossier.mkdir.side_effect = OSError("sonde refusée") self.assert_refuse(self.requete(payload), ERREUR_NOMBRES) self.chemin.assert_not_called() payload["usb_minutes"] = 1 self.assert_refuse(self.requete(payload), "Dossier de stockage inaccessible : sonde refusée") self.dossier.mkdir.side_effect = None self.assert_refuse(self.requete(payload), "Fuseau horaire inconnu : « Zone/Inexistante ».") self.contexte_sonde.__exit__.assert_called_once_with(None, None, None) payload["timezone"] = "UTC" self.assert_refuse(self.requete(payload), "Protocole de direct inconnu : « inconnu ».") self.contexte_sonde.__exit__.assert_called_once_with(None, None, None) if __name__ == "__main__": unittest.main()