test_download_diagnostic.py
Rama main · réplica de lectura del repositorio original.
"""Diagnostics bornés, sans exposer le média ou le texte d'une exception."""
import contextlib
import io
import os
import datetime as dt
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import blink_engine
import blink_models
from test_download_progression import FauxSync, FauxClip
class TestsDiagnosticAcquisition(unittest.IsolatedAsyncioTestCase):
async def test_failure_stages_preserve_calls_and_cleanup(self):
for stage in ('preparation', 'transfer', 'validation'):
with self.subTest(stage=stage), tempfile.TemporaryDirectory() as folder:
target = Path(folder) / 'private-name.mp4'
async def transfer(_blink, path):
if stage == 'validation':
Path(path).write_bytes(b'invalid media')
return stage == 'validation'
clip = SimpleNamespace(
prepare_download=mock.AsyncMock(return_value=stage != 'preparation'),
download_video=mock.AsyncMock(side_effect=transfer))
text = io.StringIO()
with mock.patch.object(blink_engine.md, 'valid_mp4_complet', return_value=False), contextlib.redirect_stdout(text):
outcome = await blink_engine.download_clip(object(), clip, target, False)
self.assertEqual(outcome, 'failed')
self.assertEqual(clip.prepare_download.await_count, 1)
self.assertEqual(clip.download_video.await_count, 0 if stage == 'preparation' else 1)
self.assertIn('='+stage, text.getvalue())
self.assertNotIn('private-name', text.getvalue())
self.assertFalse(target.exists())
self.assertFalse(target.with_suffix('.mp4.part').exists())
async def test_exception_propagates_with_only_safe_status(self):
class PrivateError(Exception):
status = 429
error = PrivateError('https://private.example/token?account=123 private-name.mp4')
for stage in ('preparation', 'transfer'):
with self.subTest(stage=stage), tempfile.TemporaryDirectory() as folder:
clip = SimpleNamespace(prepare_download=mock.AsyncMock(return_value=True),
download_video=mock.AsyncMock(return_value=True))
getattr(clip, 'prepare_download' if stage == 'preparation' else 'download_video').side_effect = error
text = io.StringIO()
with contextlib.redirect_stdout(text), self.assertRaises(PrivateError) as caught:
await blink_engine.download_clip(object(), clip, Path(folder)/'private-name.mp4', False)
self.assertIs(caught.exception, error)
self.assertIn('='+stage, text.getvalue())
self.assertIn('429', text.getvalue())
for secret in ('https:', 'token', 'account', '123', 'private-name'):
self.assertNotIn(secret, text.getvalue())
async def test_success_and_skip_do_not_emit_failure(self):
with tempfile.TemporaryDirectory() as folder:
target = Path(folder)/'private-name.mp4'
async def transfer(_blink, path):
Path(path).write_bytes(b'synthetic valid media')
return True
clip = SimpleNamespace(prepare_download=mock.AsyncMock(return_value=True),
download_video=mock.AsyncMock(side_effect=transfer))
text = io.StringIO()
with mock.patch.object(blink_engine.md, 'valid_mp4_complet', return_value=True), contextlib.redirect_stdout(text):
self.assertEqual(await blink_engine.download_clip(object(), clip, target, False), 'downloaded')
self.assertEqual(await blink_engine.download_clip(object(), clip, target, False), 'skipped')
self.assertEqual(clip.prepare_download.await_count, 1)
self.assertEqual(clip.download_video.await_count, 1)
self.assertEqual(text.getvalue(), '')
def test_unknown_status_is_not_printed_verbatim(self):
for status in ('https://private.example/token', 123456789, None):
text = io.StringIO()
with contextlib.redirect_stdout(text):
blink_engine._signaler_echec_acquisition('transfer', SimpleNamespace(status=status))
self.assertIn('unknown', text.getvalue())
self.assertNotIn('private', text.getvalue())
self.assertNotIn('123456789', text.getvalue())
def test_final_and_exception_messages_do_not_claim_retries_or_print_errors(self):
for language in ('fr', 'en'):
with mock.patch.object(blink_engine.runtime, 'lire_langue', return_value=language):
final = blink_engine.msg('usb_echec_final')
exception = blink_engine.msg('usb_echec', type='RuntimeError')
gabarit = blink_engine.LIBELLES[language]['usb_echec']
self.assertNotIn('attempt', final)
self.assertNotIn('tentative', final)
# Seul le nom de la classe est imprimé, jamais le texte de l'erreur.
self.assertIn('RuntimeError', exception)
self.assertNotIn('{', exception)
self.assertNotIn('{erreur}', gabarit)
async def test_usb_exception_log_does_not_copy_private_exception_text(self):
with tempfile.TemporaryDirectory() as folder, mock.patch.dict(os.environ, {"BLINK_HOME": folder}):
clip = FauxClip("synthetic", "Test", dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc), 7, "test")
args = SimpleNamespace(since=None, camera=None, command="download", output=Path(folder)/"clips",
hub=None, overwrite=False, source="usb", loop=None)
text = io.StringIO()
with mock.patch.object(blink_models, "read_local_manifest", new=mock.AsyncMock(return_value=[clip])), \
mock.patch.object(blink_engine, "download_clip", new=mock.AsyncMock(side_effect=RuntimeError("https://private.example/token?account=123"))), \
mock.patch.object(blink_engine.runtime, "travail"), \
mock.patch.object(blink_engine.runtime, "lire_suppression_auto", return_value=set()), \
mock.patch.object(blink_engine.runtime, "marquer"), contextlib.redirect_stdout(text):
code = await blink_engine.un_passage(object(), args, [("Test", FauxSync(10, 7))])
self.assertEqual(code, 1)
for secret in ("https:", "private.example", "token", "account=123"):
self.assertNotIn(secret, text.getvalue())
async def test_directory_and_initial_unlink_errors_are_local(self):
for operation in ('mkdir', 'unlink'):
with self.subTest(operation=operation), tempfile.TemporaryDirectory() as folder:
target = Path(folder)/'private-name.mp4'
error = PermissionError('https://private.example/token?account=123')
clip = SimpleNamespace(prepare_download=mock.AsyncMock(), download_video=mock.AsyncMock())
text = io.StringIO()
with mock.patch.object(Path, operation, side_effect=error) as operation_mock, contextlib.redirect_stdout(text):
with self.assertRaises(PermissionError) as caught:
await blink_engine.download_clip(object(), clip, target, False)
self.assertIs(caught.exception, error)
self.assertEqual(operation_mock.call_count, 1)
self.assertEqual(clip.prepare_download.await_count, 0)
self.assertEqual(clip.download_video.await_count, 0)
self.assertEqual(text.getvalue().count('=local'), 1)
for secret in ('https:', 'token', 'account', '123', 'private-name'):
self.assertNotIn(secret, text.getvalue())
async def test_cleanup_error_retains_error_and_reports_local(self):
for primary in ('false-transfer', 'exception', 'success'):
with self.subTest(primary=primary), tempfile.TemporaryDirectory() as folder:
target = Path(folder)/'private-name.mp4'
original_unlink = Path.unlink
cleanup_error = PermissionError('https://private.example/cleanup-token')
primary_error = RuntimeError('https://private.example/primary-token')
calls = []
def unlink(path, *args, **kwargs):
calls.append(path)
if len(calls) == 2:
raise cleanup_error
return original_unlink(path, *args, **kwargs)
async def transfer(_blink, path):
if primary == 'exception':
raise primary_error
if primary == 'success':
Path(path).write_bytes(b'synthetic valid media')
return True
return False
clip = SimpleNamespace(prepare_download=mock.AsyncMock(return_value=True),
download_video=mock.AsyncMock(side_effect=transfer))
text = io.StringIO()
with mock.patch.object(Path, 'unlink', new=unlink), \
mock.patch.object(blink_engine.md, 'valid_mp4_complet', return_value=True), \
contextlib.redirect_stdout(text), self.assertRaises(PermissionError) as caught:
await blink_engine.download_clip(object(), clip, target, False)
self.assertIs(caught.exception, cleanup_error)
if primary == 'exception':
self.assertIs(cleanup_error.__context__, primary_error)
self.assertEqual(len(calls), 2)
self.assertEqual(clip.prepare_download.await_count, 1)
self.assertEqual(clip.download_video.await_count, 1)
self.assertEqual(target.exists(), primary == 'success')
self.assertEqual(text.getvalue().count('=local'), 1)
self.assertEqual(text.getvalue().count('=transfer'), 0 if primary == 'success' else 1)
for secret in ('https:', 'token', 'private-name'):
self.assertNotIn(secret, text.getvalue())