tests/test_local_poster.py
Rama master · réplica de lectura del repositorio original.
"""同目录封面的隔离读取、超时回收与独立包入口。"""
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch
from PIL import Image
from peach.local_poster import read_jpeg
class LocalPosterTests(unittest.TestCase):
def test_generated_sidecar_poster_is_signed_served_and_rebuilt_when_source_changes(self):
from types import SimpleNamespace
from peach import artifact_rules, upgrade_rules
from peach.library_processing import _read_local_posters, _queue_local_poster
from peach.previews import PreviewService, PreviewUnavailable
from peach.repository import MediaAsset
generated = self.root / 'posters'
jobs = []
enqueue = lambda source, code, target: jobs.append((dict(id=1, name='video.mp4'), source, code, target))
def generate():
jobs.clear()
_queue_local_poster(self.root / 'video.mp4', '1_4', generated, posters=[self.source],
enqueue=enqueue, generated=True)
return _read_local_posters(jobs, dict(speed_mode='standard', covers=0), update=lambda **_: None,
issue=lambda *_args, **_kwargs: self.fail('配套图片应可读取'), generated_root=generated)
self.assertEqual(generate(), set())
poster = generated / '1_4.jpg'
self.assertTrue(artifact_rules.current(poster, upgrade_rules.LOCAL_POSTERS, source_required=True))
service = PreviewService(SimpleNamespace(media_asset=lambda _: MediaAsset(id=1, path=None, snapshot_path=None)),
None, self.root / 'snapshots', generated, self.root / 'avatars', self.root / 'logos')
self.assertEqual(service.poster(1), poster)
generate()
self.assertEqual(jobs, [])
Image.new('RGB', (24, 36), 'red').save(self.source)
with self.assertRaises(PreviewUnavailable):
service.poster(1)
generate()
self.assertEqual(len(jobs), 1)
self.assertEqual(service.poster(1), poster)
self.assertTrue(self.source.is_file())
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.source = self.root / '示例封面.png'
Image.new('RGB', (24, 36), 'teal').save(self.source)
def test_child_reads_a_unicode_path_and_returns_jpeg(self):
data, size = read_jpeg(self.source)
self.assertEqual(size, (24, 36))
self.assertTrue(data.startswith(b'\xff\xd8'))
def test_unreadable_image_returns_a_retryable_error(self):
self.source.write_bytes(b'not an image')
with self.assertRaisesRegex(ValueError, '无法读取'):
read_jpeg(self.source)
def test_local_disk_picture_decodes_without_starting_a_process(self):
expected = read_jpeg(self.source)
with patch('peach.local_poster.subprocess.run', side_effect=AssertionError('本地图片不启动子进程')):
self.assertEqual(read_jpeg(self.source, isolated=False), expected)
def test_timeout_terminates_and_reaps_the_child(self):
from peach import local_poster
original_run = subprocess.run
original_popen = subprocess.Popen
children = []
def spawn(*args, **kwargs):
child = original_popen(*args, **kwargs)
children.append(child)
return child
def blocked(command, **options):
return original_run([sys.executable, '-c', 'import time; time.sleep(60)'], **options)
with patch.object(local_poster, 'TIMEOUT_SECONDS', 0.2), \
patch.object(local_poster.subprocess, 'run', side_effect=blocked), \
patch.object(local_poster.subprocess, 'Popen', side_effect=spawn), \
self.assertRaises(subprocess.TimeoutExpired):
read_jpeg(self.source)
self.assertEqual(len(children), 1)
self.assertIsNotNone(children[0].poll())
def test_packaged_command_uses_the_cli_worker(self):
from peach.cli import main
output = self.root / 'worker.jpg'
self.assertEqual(main(['local-image', str(self.source), str(output)]), 0)
with Image.open(output) as image:
self.assertEqual(image.size, (24, 36))
def test_picture_batches_use_the_selected_concurrency_and_one_writer(self):
from peach.library_processing import _read_local_posters
data, size = read_jpeg(self.source)
for mode, workers in (('low', 1), ('standard', 2), ('high', 4)):
with self.subTest(mode=mode):
barrier = threading.Barrier(workers, timeout=3)
readers, writers = set(), set()
jobs = [(dict(id=index, name=f'{index}.mp4'), self.source, str(index),
self.root / mode / f'{index}.jpg') for index in range(workers)]
def read(_source, *, readers=readers, barrier=barrier, **_options):
readers.add(threading.get_ident())
barrier.wait()
return data, size
state = dict(speed_mode=mode, covers=0)
with patch('peach.local_poster.read_jpeg', side_effect=read):
failed = _read_local_posters(jobs, state,
update=lambda writers=writers, **_: writers.add(threading.get_ident()),
issue=lambda *_args, **_kwargs: self.fail('图片应可读取'))
self.assertEqual(len(readers), workers)
self.assertEqual(writers, {threading.get_ident()})
self.assertEqual(state['covers'], workers)
self.assertEqual(failed, set())
def test_stop_does_not_start_the_next_picture_batch(self):
from peach.library_processing import _read_local_posters
data, size = read_jpeg(self.source)
jobs = [(dict(id=index, name=f'{index}.mp4'), self.source, str(index),
self.root / f'{index}.jpg') for index in range(5)]
def update(**values):
if values.get('checked') == 1:
raise InterruptedError('停止')
with patch('peach.local_poster.read_jpeg', return_value=(data, size)) as read, \
self.assertRaises(InterruptedError):
_read_local_posters(jobs, dict(speed_mode='standard', covers=0),
update=update, issue=lambda *_args, **_kwargs: None)
self.assertEqual(read.call_count, 2)
def test_idle_reader_takes_another_picture_while_one_is_waiting(self):
from peach.library_processing import _read_local_posters
data, size = read_jpeg(self.source)
third_started = threading.Event()
jobs = [(dict(id=index, name=f'{index}.mp4'), Path(str(index)), str(index),
self.root / f'{index}.jpg') for index in range(3)]
def read(source, **_options):
if str(source) == '0':
self.assertTrue(third_started.wait(3))
if str(source) == '2':
third_started.set()
return data, size
with patch('peach.local_poster.read_jpeg', side_effect=read):
failed = _read_local_posters(jobs, dict(speed_mode='standard', covers=0),
update=lambda **_: None, issue=lambda *_args, **_kwargs: self.fail('图片应可读取'))
self.assertEqual(failed, set())
def test_same_release_installs_one_shared_cover(self):
from peach.library_processing import _read_local_posters
data, size = read_jpeg(self.source)
target = self.root / 'covers' / 'ABW-101.jpg'
jobs = [(dict(id=index, name=f'ABW-101-{index}.mp4'), self.source, 'ABW-101', target)
for index in range(2)]
state = dict(speed_mode='standard', covers=0)
with patch('peach.local_poster.read_jpeg', return_value=(data, size)):
failed = _read_local_posters(jobs, state, update=lambda **_: None,
issue=lambda *_args, **_kwargs: self.fail('图片应可读取'))
self.assertEqual((state['covers'], failed), (1, set()))
def test_mixed_sources_isolate_every_picture(self):
from peach.library_processing import _read_local_posters
jobs = [(dict(id=index, name=f'{index}.mp4', location=location), self.source, str(index),
self.root / 'covers' / f'{index}.jpg') for index, location in enumerate(('local', 'pikpak'))]
state = dict(speed_mode='standard', covers=0)
with patch('peach.local_poster.subprocess.run', wraps=subprocess.run) as run:
failed = _read_local_posters(jobs, state, update=lambda **_: None,
issue=lambda *_args, **_kwargs: self.fail('图片应可读取'))
self.assertEqual((state['covers'], failed), (2, set()))
self.assertEqual(run.call_count, 2)