test_issue139_delai_media.py
Rama main · réplica de lectura del repositorio original.
"""Issue #139 : des clips cloud échouaient à chaque tour en « [write] TimeoutError ».
Chez MarkusKress, 19 clips sur 38 revenaient à chaque tour. blinkpy 0.25.9
demande 90 s pour un média, mais ``api.http_get`` perd ce paramètre en route :
la requête part avec les 10 s des appels JSON, et aiohttp applique ce délai
total à la lecture du corps. Tout clip dont le transfert durait plus de 10 s
était coupé. ``TimeoutError`` héritant d'``OSError``, le journal l'étiquetait
en plus « écriture ».
Le premier test rejoue ce trajet tel quel : le vrai ``Auth.query`` de blinkpy,
une vraie session aiohttp et un petit serveur local qui envoie un clip plus
lentement que le délai total d'origine, mais sans jamais se taire longtemps."""
import asyncio
import contextlib
import inspect
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import aiohttp
from aiohttp import web
from blinkpy.auth import Auth
from blinkpy.blinkpy import Blink
import blink_engine
import blink_models
import runtime
from test_blink2video_audit import MP4_STRUCTUREL, FauxBlinkHTTP
def _delai_json_par_defaut(secondes):
"""Ramène les 10 s de ``Auth.query`` à une durée de test.
``timeout=TIMEOUT`` est figé à la définition de la méthode : remplacer la
constante du module n'y changerait rien, c'est la valeur par défaut qu'il
faut remplacer."""
noms = [nom for nom, parametre in inspect.signature(Auth.query).parameters.items()
if parametre.default is not inspect.Parameter.empty]
defauts = list(Auth.query.__defaults__)
defauts[noms.index("timeout")] = secondes
return mock.patch.object(Auth.query, "__defaults__", tuple(defauts))
def _clip(numero=1):
return blink_models.CloudClip({
"id": numero, "device_name": "Hof",
"created_at": "2026-11-07T04:02:22+00:00",
"media": f"/api/v2/media/{numero}.mp4", "network_id": 7,
})
class ServeurLent:
"""Envoie MP4_STRUCTUREL en morceaux espacés, comme un clip lourd."""
def __init__(self, morceaux: int, pause: float):
self.morceaux = morceaux
self.pause = pause
async def servir(self, request):
reponse = web.StreamResponse(headers={"Content-Type": "video/mp4"})
reponse.content_length = len(MP4_STRUCTUREL)
await reponse.prepare(request)
taille = -(-len(MP4_STRUCTUREL) // self.morceaux)
for debut in range(0, len(MP4_STRUCTUREL), taille):
await reponse.write(MP4_STRUCTUREL[debut:debut + taille])
await asyncio.sleep(self.pause)
return reponse
@contextlib.asynccontextmanager
async def demarrer(self):
app = web.Application()
app.router.add_get("/api/v2/media/{nom}", self.servir)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "127.0.0.1", 0)
await site.start()
port = site._server.sockets[0].getsockname()[1]
try:
yield f"http://127.0.0.1:{port}"
finally:
await runner.cleanup()
class TransfertLentAvecBlinkpy(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.dossier = tempfile.TemporaryDirectory(prefix="blink-139-")
self.partiel = Path(self.dossier.name) / "clip.mp4.part"
async def asyncTearDown(self):
self.dossier.cleanup()
async def _telecharger(self, silence_tolere, delai_json):
"""Transfert de 1,2 s environ, jamais plus de 0,3 s sans octet."""
serveur = ServeurLent(morceaux=4, pause=0.3)
async with serveur.demarrer() as base, aiohttp.ClientSession() as session:
auth = Auth({"token": "jeton"}, no_prompt=True, session=session)
blink = SimpleNamespace(auth=auth, urls=SimpleNamespace(base_url=base))
clip = _clip()
with mock.patch.object(blink_models, "SILENCE_MEDIA_CLOUD", silence_tolere), \
_delai_json_par_defaut(delai_json):
ok = await clip.download_to(blink, self.partiel)
return ok, clip
async def test_temoin_blinkpy_coupe_le_transfert_au_delai_json(self):
# Le défaut d'origine, sur le code de blinkpy lui-même : do_http_get
# annonce 90 s mais la lecture du corps tombe au délai JSON. Si ce
# témoin casse, blinkpy a corrigé api.http_get et demander_media_cloud
# peut redevenir un simple appel à do_http_get.
serveur = ServeurLent(morceaux=4, pause=0.3)
async with serveur.demarrer() as base, aiohttp.ClientSession() as session:
auth = Auth({"token": "jeton"}, no_prompt=True, session=session)
blink = SimpleNamespace(auth=auth, urls=SimpleNamespace(base_url=base))
with _delai_json_par_defaut(0.5):
reponse = await Blink.do_http_get(blink, "/api/v2/media/1.mp4")
with self.assertRaises(asyncio.TimeoutError):
await reponse.read()
async def test_un_transfert_plus_long_que_le_delai_json_aboutit(self):
# Délai JSON ramené à 0,5 s (10 s en vrai), silence toléré 1 s (60 s) :
# avant le correctif, ce clip tombait en TimeoutError à chaque tour.
ok, clip = await self._telecharger(silence_tolere=1, delai_json=0.5)
self.assertTrue(ok, clip.download_issue)
self.assertEqual(self.partiel.read_bytes(), MP4_STRUCTUREL)
async def test_une_connexion_figee_tombe_en_reseau_pas_en_ecriture(self):
# Silence toléré plus court que les pauses du serveur : le transfert
# doit échouer, et le motif doit dire réseau, jamais écriture.
ok, clip = await self._telecharger(silence_tolere=0.1, delai_json=10)
self.assertFalse(ok)
categorie, detail = clip.download_issue
self.assertEqual(categorie, blink_models.msg("cat_reseau"))
self.assertIn("TimeoutError", detail) # SocketTimeoutError chez aiohttp
self.assertFalse(self.partiel.exists())
class DelaiTransmis(unittest.IsolatedAsyncioTestCase):
async def test_la_requete_borne_le_silence_et_non_la_duree(self):
class Reponse:
status = 200
headers = {}
async def read(self):
return MP4_STRUCTUREL
blink = FauxBlinkHTTP(Reponse())
with tempfile.TemporaryDirectory(prefix="blink-139-") as dossier:
self.assertTrue(await _clip(3).download_to(blink, Path(dossier) / "c.part"))
(requete,) = blink.requetes
self.assertEqual(requete["url"], "https://rest-test.invalid/api/v2/media/3.mp4")
self.assertFalse(requete["json_resp"])
delai = requete["timeout"]
self.assertIsNone(delai.total)
self.assertEqual(delai.sock_read, blink_models.SILENCE_MEDIA_CLOUD)
class CategorieDesErreurs(unittest.IsolatedAsyncioTestCase):
async def _motif(self, erreur):
class Flux:
async def iter_chunked(self, _taille):
yield MP4_STRUCTUREL[:10]
raise erreur
blink = FauxBlinkHTTP(SimpleNamespace(status=200, content=Flux(), headers={}))
clip = _clip()
with tempfile.TemporaryDirectory(prefix="blink-139-") as dossier:
self.assertFalse(await clip.download_to(blink, Path(dossier) / "c.part"))
return clip.download_issue[0]
async def test_delai_et_connexion_sont_du_reseau(self):
for erreur in (TimeoutError(), asyncio.TimeoutError(), ConnectionResetError(),
aiohttp.ClientPayloadError("coupé")):
with self.subTest(erreur=type(erreur).__name__):
self.assertEqual(await self._motif(erreur), blink_models.msg("cat_reseau"))
async def test_disque_plein_reste_de_l_ecriture(self):
self.assertEqual(await self._motif(OSError(28, "No space left on device")),
blink_models.msg("cat_ecriture"))
class ListeCloudVide(unittest.TestCase):
"""Une liste vide rendue par blinkpy après une requête ratée laissait un
tour muet, comme un tour sans nouveauté. Seuls les changements s'écrivent."""
def test_ecrit_le_passage_a_vide_puis_le_retour_seulement(self):
with tempfile.TemporaryDirectory(prefix="blink-139-") as dossier, \
mock.patch.object(runtime, "app_dir", return_value=Path(dossier)), \
mock.patch.object(blink_engine, "_liste_cloud_vide", None), \
mock.patch.object(runtime, "lire_langue", return_value="en"):
for lus in (40, 40, 0, 0, 0, 38, 38):
blink_engine._noter_liste_cloud(lus, 5411)
journal = Path(dossier) / blink_engine.JOURNAL_TELECHARGEMENTS
lignes = [ligne.split(" ", 1)[1] for ligne in
journal.read_text(encoding="utf-8").splitlines()]
self.assertEqual(lignes, [
"inventory cloud: empty listing (registry 5411)",
"inventory cloud: listing back, 38 clip(s)",
])
def test_un_premier_tour_vide_est_ecrit(self):
with tempfile.TemporaryDirectory(prefix="blink-139-") as dossier, \
mock.patch.object(runtime, "app_dir", return_value=Path(dossier)), \
mock.patch.object(blink_engine, "_liste_cloud_vide", None):
blink_engine._noter_liste_cloud(0, 0)
blink_engine._noter_liste_cloud(0, 0)
journal = Path(dossier) / blink_engine.JOURNAL_TELECHARGEMENTS
self.assertEqual(len(journal.read_text(encoding="utf-8").splitlines()), 1)
if __name__ == "__main__":
unittest.main()