tests/test_downloads_postprocess.py
Rama master · réplica de lectura del repositorio original.
"""下载后处理在独立账本和临时媒体上的失败与重试。"""
import os
import tempfile
import time
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch
from peach import download_postprocess as post
from peach import ledger_backups
from peach.downloads import DownloadStore, DownloadTask, INGESTED, POSTPROCESS_PENDING
from peach.ingest_notices import IngestNotices
from peach.jobs import JobDeferred
from peach.repository import LedgerDatabase
from support.ledger import fresh_ledger
class DownloadPostprocessTests(unittest.TestCase):
def test_isolated_ads_record_the_junk_reason(self):
with self.database.write_transaction() as connection:
connection.execute("UPDATE asset SET medium='other',name='推广.url',path=? WHERE id=?",
(str(self.root / '推广.url'), self.asset_id))
self.assertEqual(post._isolate_ads(self.contract, -1, [self.asset_id]), 1)
with self.database.read_connection() as connection:
self.assertEqual(tuple(connection.execute("SELECT disposal,disposal_reason FROM asset WHERE id=?",
(self.asset_id,)).fetchone()), ("trash", "junk"))
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.database = LedgerDatabase(fresh_ledger(self.root))
self.source = self.root / 'incoming' / 'ABC-123.mp4'
self.target = self.root / 'organized' / self.source.name
self.source.parent.mkdir()
self.source.write_bytes(b'main video')
with self.database.write_transaction() as connection:
self.asset_id = connection.execute(
'INSERT INTO asset(location,path,name,medium,size,code) VALUES(?,?,?,?,?,?)',
('115', str(self.source), self.source.name, 'video', 10, 'ABC-123')).lastrowid
self.store = DownloadStore(self.database)
self.task = self.store.save(DownloadTask(
id=0, info_hash='a' * 40, provider='115', source_uri='', display_name='ABC-123',
target='/115/incoming', state=INGESTED, code='ABC-123',
ledger_path=str(self.source.parent), asset_id=self.asset_id,
postprocess_state=POSTPROCESS_PENDING))
self.contract = SimpleNamespace(
database=self.database, db_path=self.database.db_path,
downloads=SimpleNamespace(wake=Mock(), clock=time.time),
candidate_root=self.root / 'generated',
download_organize_templates=lambda: {'115': {'file': '{code}'}}, cache_bust=lambda: None)
self.handle = SimpleNamespace(progress=lambda **_: None)
plan = {'rows': [{'asset_id': self.asset_id, 'action': 'move',
'current_path': str(self.source), 'target_path': str(self.target)}], 'reasons': {}}
self.plan = patch.object(post.organize, 'build_plan', return_value=plan)
self.plan.start()
self.addCleanup(self.plan.stop)
self.collect_source = post._collect
collect = patch.object(post, '_collect', return_value={'candidates': 0})
collect.start()
self.addCleanup(collect.stop)
probe = patch.object(post, '_probe_folder', return_value=0)
probe.start()
self.addCleanup(probe.stop)
def run_task(self):
return post.run(self.contract, post.followup_key(self.task.id), self.handle)
def test_temporary_failure_is_scheduled_and_only_three_retries_are_allowed(self):
from peach.downloads import parse_iso
from peach.task_retry import RETRY_DELAYS
started = 1_800_000_000.0
self.contract.downloads.clock = lambda: started
with patch.object(post, '_collect', side_effect=TimeoutError('来源连接超时')):
for attempt, delay in enumerate(RETRY_DELAYS, 1):
with self.assertRaises(TimeoutError):
self.run_task()
task = self.store.get(self.task.id)
self.assertEqual((task.postprocess_state, task.postprocess_retries), ('pending', attempt))
self.assertEqual(parse_iso(task.next_check_at), started + delay)
self.assertEqual(self.store.pending_postprocess(started), [])
self.assertEqual([row.id for row in self.store.pending_postprocess(parse_iso(task.next_check_at))], [task.id])
started += delay
with self.assertRaises(TimeoutError):
self.run_task()
task = self.store.get(self.task.id)
self.assertEqual(task.postprocess_state, 'failed')
self.assertIsNone(task.next_check_at)
self.assertIn('自动重试 3 次', task.postprocess_detail)
def test_offline_mount_retries_but_target_collision_stops_immediately(self):
with patch.object(post.organize, 'build_plan', return_value={
'rows': [{'action': 'skip', 'reason': post.organize.SKIP_OFFLINE}],
'reasons': {post.organize.SKIP_OFFLINE: 1}}), self.assertRaises(RuntimeError):
self.run_task()
self.assertEqual(self.store.get(self.task.id).postprocess_state, 'pending')
with patch.object(post.organize, 'build_plan', return_value={
'rows': [{'action': 'skip', 'reason': post.organize.SKIP_TARGET_EXISTS}],
'reasons': {post.organize.SKIP_TARGET_EXISTS: 1}}), self.assertRaises(RuntimeError):
self.run_task()
self.assertEqual(self.store.get(self.task.id).postprocess_state, 'failed')
self.assertIsNone(self.store.get(self.task.id).next_check_at)
def test_collection_retry_uses_structured_failure_instead_of_the_error_message(self):
log = self.root / 'issues.jsonl'
import json
state = {'issues_log': str(log), 'issue_count': 1}
issue = {'asset_id': self.asset_id, 'severity': 'error', 'retryable': True,
'diagnostics': {'failures': [{'automatic_retry_delay': 900.0}]}}
log.write_text(json.dumps(issue) + '\n', encoding='utf-8')
self.assertEqual(post._collection_retry_delay(state, self.asset_id), 900)
issue['diagnostics'] = {'failures': [{'automatic_retry_delay': None}]}
log.write_text(json.dumps(issue) + '\n', encoding='utf-8')
self.assertIsNone(post._collection_retry_delay(state, self.asset_id))
def test_existing_unrelated_foreign_key_violation_is_reported_without_failing_the_move(self):
with self.database.write_transaction() as connection:
connection.execute("INSERT INTO media_binding(asset_id,backend,external_id) VALUES(99999,'test','x')")
result = self.run_task()
self.assertEqual((result['moved'], result['existing_foreign_keys']), (1, 1))
self.assertEqual(self.store.get(self.task.id).postprocess_state, 'complete')
self.assertTrue(self.target.exists())
def test_organize_backup_lands_in_the_organize_folder_and_old_ones_are_pruned(self):
folder = ledger_backups.organize_dir(self.contract.candidate_root)
folder.mkdir(parents=True)
old = []
for index in range(ledger_backups.ORGANIZE_KEEP_RECENT + 2):
path = folder / f'ledger.pre-download-{index}-20260801-000000.db'
path.write_bytes(b'backup')
moment = time.time() - 30 * 86400 + index
os.utime(path, (moment, moment))
old.append(path)
result = post._organize(self.contract, self.task, self.asset_id)
backup = Path(result['backup'])
self.assertEqual(backup.parent, folder)
self.assertEqual(ledger_backups.list_backups(folder, ledger_backups.ORGANIZE_PATTERN),
[*old[3:], backup])
def test_retry_uses_the_bound_asset_after_a_move(self):
post._organize(self.contract, self.task, self.asset_id)
main, path, _ = post._bind_main(self.contract, self.store.get(self.task.id))
self.assertEqual((main, path), (self.asset_id, str(self.target)))
def test_cloud_library_channel_binds_assets_by_media_source(self):
self.task.provider = 'clouddrive:guangya:test-library'
with self.database.write_transaction() as connection:
connection.execute("UPDATE asset SET location='guangya' WHERE id=?", (self.asset_id,))
main, path, ids = post._bind_main(self.contract, self.task)
self.assertEqual((main, path, ids), (self.asset_id, str(self.source), [self.asset_id]))
self.assertEqual(self.task.payload()['provider_label'], '光鸭云盘')
def test_occupied_target_is_a_retryable_failure(self):
self.target.parent.mkdir()
self.target.write_bytes(b'existing file')
with self.assertRaisesRegex(RuntimeError, '整理'):
self.run_task()
self.assertEqual(self.store.get(self.task.id).postprocess_state, 'failed')
self.assertTrue(self.source.exists())
self.assertEqual(self.target.read_bytes(), b'existing file')
def test_new_foreign_key_violation_is_rejected(self):
original = post.organize.apply_plan
def corrupt(connection, rows, **options):
result = original(connection, rows, **options)
connection.execute("INSERT INTO media_binding(asset_id,backend,external_id) VALUES(99999,'test','x')")
connection.commit()
return result
with patch.object(post.organize, 'apply_plan', side_effect=corrupt), \
self.assertRaisesRegex(RuntimeError, '校验'):
self.run_task()
self.assertEqual(self.store.get(self.task.id).postprocess_state, 'failed')
def test_a_collision_found_in_the_plan_is_a_failure(self):
with patch.object(post.organize, 'build_plan', return_value={
'rows': [{'action': 'skip', 'reason': post.organize.SKIP_TARGET_EXISTS}],
'reasons': {post.organize.SKIP_TARGET_EXISTS: 1}}), \
self.assertRaisesRegex(RuntimeError, '整理'):
self.run_task()
self.assertTrue(self.source.exists())
def test_followups_are_declared_only_after_successful_organization(self):
events = []
self.handle.run_id = 123
self.contract.task_runs = Mock()
rows = [{'task_key': 'avatar', 'followup_key': 'asset:1'}]
self.contract.task_runs.enqueue_followups.side_effect = lambda *_: events.append('followup')
original = post._organize
def organize(*args):
events.append('organize')
return original(*args)
notices = IngestNotices()
with patch.object(post, '_isolate_ads', side_effect=lambda *_: events.append('check') or 0), \
patch.object(post, '_probe_folder', side_effect=lambda *_, **options: events.append('probe') or 1), \
patch.object(post, '_collect', side_effect=lambda *_: events.append('collect') or {'followups': rows}), \
patch.object(post, '_organize', side_effect=organize), \
patch.object(post, 'declared_rows', return_value=rows), patch.object(post, 'NOTICES', notices):
result = self.run_task()
self.assertEqual(self.run_task(), {'task_id': self.task.id, 'already_complete': True})
self.assertEqual(events, ['check', 'probe', 'check', 'collect', 'check', 'organize', 'followup'])
self.assertEqual(result['probed'], 1)
self.contract.task_runs.enqueue_followups.assert_called_once()
# 整理完成记一条入库通知，页面据此弹 toast；已完成的再跑一次不重复记。
[notice] = notices.snapshot()['items']
self.assertEqual((notice['kind'], notice['asset_id']), ('download', result['asset_id']))
def test_collection_that_steps_aside_stays_queued_instead_of_failed(self):
self.handle.run_id = 123
self.contract.task_runs = Mock()
with patch.object(post, '_collect', side_effect=JobDeferred('等待扫描与采集完成后继续')):
with self.assertRaises(JobDeferred):
self.run_task()
task = self.store.get(self.task.id)
self.assertEqual((task.postprocess_state, task.postprocess_detail), ('queued', '等待扫描与采集完成后继续'))
self.contract.task_runs.enqueue_followups.assert_not_called()
def test_collection_held_only_by_paused_sources_still_organizes_the_file(self):
"""MIAB-086：其余资料已落库，只差一站拒绝访问；文件照常整理，详情写明哪项等来源恢复。"""
held = {'status': 'failed', 'error': '1 项都卡在来源限流上，等一会儿再跑一次。',
'issue_count': 1, 'paused_count': 1,
'issue_preview': [{'message': 'JavDB 标签未取得：来源拒绝访问'}]}
# 账本就在临时根下，配置的各目录都指回它。
config = SimpleNamespace(configured=True, directory=lambda name: self.root)
self.contract.cover_root = self.root / 'covers'
with patch.object(post.settings_file, 'active', return_value=config), \
patch.object(post, 'process_library', return_value=held):
state = self.collect_source(self.contract, self.asset_id, self.handle)
self.assertEqual(state['paused_pending'], ['JavDB 标签未取得：来源拒绝访问'])
with patch.object(post, '_collect', return_value=state):
result = self.run_task()
self.assertEqual(result['moved'], 1)
task = self.store.get(self.task.id)
self.assertEqual(task.postprocess_state, 'complete')
self.assertIn('JavDB 标签未取得：来源拒绝访问', task.postprocess_detail)
# 混有来源限流以外的问题时照旧失败，等人处理。
with patch.object(post.settings_file, 'active', return_value=config), \
patch.object(post, 'process_library', return_value={**held, 'issue_count': 2}):
with self.assertRaisesRegex(RuntimeError, '来源限流'):
self.collect_source(self.contract, self.asset_id, self.handle)
def test_metadata_or_organization_failure_never_dispatches_followups(self):
self.handle.run_id = 123
self.contract.task_runs = Mock()
notices = IngestNotices()
for failed_stage in ('_collect', '_organize'):
with self.subTest(stage=failed_stage), patch.object(post, failed_stage, side_effect=RuntimeError('模拟阶段失败')), \
patch.object(post, 'NOTICES', notices):
with self.assertRaisesRegex(RuntimeError, '模拟阶段失败'):
self.run_task()
self.assertEqual(self.store.get(self.task.id).postprocess_state, 'failed')
self.contract.task_runs.enqueue_followups.assert_not_called()
self.assertEqual(notices.snapshot()['items'], [])