tests/test_processing_inventory.py
Rama master · réplica de lectura del repositorio original.
"""首次目录的只读分批预检与请求生命周期。"""
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from unittest.mock import patch
from peach.processing_inventory import preview, recent_rows, _SESSIONS, _LATEST
from peach.processing_traffic import ProcessingTraffic
from peach.settings_file import PeachConfig
from support.ledger import fresh_ledger
class InventoryTests(unittest.TestCase):
def test_library_inventory_excludes_other_library_assets_and_roots(self):
from dataclasses import replace
from peach.media_libraries import libraries
from peach.library_processing_scope import scoped_config
config = replace(self.config, locations={'mounted': ('A:\\Cloud', 'B:\\Other')})
with closing(sqlite3.connect(self.db)) as connection, connection:
for location in ('A:\\Cloud', 'B:\\Other'):
connection.execute('INSERT INTO asset(location,path,name,medium) VALUES(?,?,?,?)',
('mounted', location + '\\ABW-001.mp4', 'ABW-001.mp4', 'video'))
library_id = libraries(config)[0]['id']
selected = scoped_config(config, library_id)
result = preview(selected, self.db, 'collect', None, library_id=library_id)
self.assertEqual(result['workload']['video_files'], 1)
self.assertEqual(selected.locations, {'mounted': ('A:\\Cloud',)})
with self.assertRaisesRegex(ValueError, '媒体库'):
scoped_config(config, 'missing')
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name).resolve()
self.media = self.root / 'media'
self.media.mkdir()
self.config = PeachConfig(self.root, self.root / 'config.toml', present=True,
locations={'mounted': ('A:\\Cloud',)}, media_costs={'A:\\Cloud': 'metered'})
self.db = fresh_ledger(self.root)
self.addCleanup(_SESSIONS.clear)
self.addCleanup(_LATEST.clear)
mount = patch('peach.scan.walk_root_for', return_value=self.media)
self.walk = mount.start()
self.addCleanup(mount.stop)
def test_empty_ledger_recognizes_names_without_opening_media_or_writing_ledger(self):
for name in ('ABW-001.mp4', 'ABW-001-CD2.MKV', '日常.mov', 'ABW-001.nfo', 'poster.jpg'):
(self.media / name).write_bytes(b'')
(self.media / '#recycle').mkdir()
(self.media / '#recycle' / 'ABW-002.mp4').touch()
before = self.db.read_bytes()
original = Path.open
def guarded(path, *args, **kwargs):
self.assertFalse(path.is_relative_to(self.media), '预检读取了媒体内容')
return original(path, *args, **kwargs)
with patch.object(Path, 'open', guarded):
result = preview(self.config, self.db, 'all', None)
self.assertTrue(result['inventory']['complete'])
self.assertEqual(result['inventory']['errors'], 0)
self.assertEqual(result['workload']['video_files'], 3)
self.assertEqual(result['workload']['jav_files'], 2)
self.assertEqual(result['workload']['jav_titles'], 1)
self.assertGreater(result['estimated_bytes'], 0)
self.assertEqual(self.db.read_bytes(), before)
def test_release_code_families_count_as_jav_during_inventory(self):
for name in ('LAF-41.mpg', 'MKBD-S94.mkv', 'MKD-S118.iso', '040221_0012.flv',
'IMG_3092.mpg', 'VIDEO-2022.rmvb'):
(self.media / name).touch()
before = self.db.read_bytes()
result = preview(self.config, self.db, 'all', None)
self.assertEqual(result['workload']['video_files'], 6)
self.assertEqual(result['workload']['jav_files'], 4)
self.assertEqual(result['workload']['jav_titles'], 4)
self.assertEqual(self.db.read_bytes(), before)
def test_trailers_are_excluded_from_discovered_and_registered_workloads(self):
for name in ('ABC-123.mp4', 'ABC-123-trailer.mp4', 'ABC-123.TRAILER.MKV', 'short.mp4'):
(self.media / name).touch()
before = self.db.read_bytes()
discovered = preview(self.config, self.db, 'all', None)
self.assertEqual(discovered['workload']['video_files'], 2)
self.assertEqual(discovered['workload']['jav_files'], 1)
self.assertEqual(self.db.read_bytes(), before)
with closing(sqlite3.connect(self.db)) as connection, connection:
for name in ('ABC-123.mp4', 'ABC-123-trailer.mp4'):
connection.execute('INSERT INTO asset(location,path,name,medium) VALUES(?,?,?,?)',
('mounted', 'A:\\Cloud\\' + name, name, 'video'))
before = self.db.read_bytes()
registered = preview(self.config, self.db, 'all', None)
self.assertEqual(registered['workload']['video_files'], 2)
self.assertEqual(self.db.read_bytes(), before)
def test_extra_directories_have_no_discovered_or_registered_workload(self):
relative = ('Film/Film.mp4', 'Film/Trailers/Teaser.mp4',
'Film/Extras/Disc 1/Making of.mp4', 'Film/Actor-interview2.mkv')
for value in relative:
target = self.media / value
target.parent.mkdir(parents=True, exist_ok=True)
target.touch()
before = self.db.read_bytes()
result = preview(self.config, self.db, 'all', None)
self.assertEqual(result['workload']['video_files'], 1)
self.assertEqual(self.db.read_bytes(), before)
with closing(sqlite3.connect(self.db)) as connection, connection:
for value in relative:
connection.execute('INSERT INTO asset(location,path,name,medium) VALUES(?,?,?,?)',
('mounted', 'A:/Cloud/' + value, Path(value).name, 'video'))
before = self.db.read_bytes()
result = preview(self.config, self.db, 'all', None)
self.assertEqual(result['workload']['video_files'], 1)
self.assertEqual(self.db.read_bytes(), before)
def test_each_query_advances_a_batch_and_preserves_deduplication(self):
for number in range(1100):
(self.media / f'ABW-001-CD{number}.mp4').touch()
first = preview(self.config, self.db, 'scan', None)
self.assertFalse(first['inventory']['complete'])
token = first['inventory']['token']
session = _SESSIONS[token]
count = len(session.rows)
self.assertEqual(len(session.rows), count)
result = first
for _ in range(20):
if result['inventory']['complete']:
break
result = preview(self.config, self.db, 'scan', None, token)
self.assertTrue(result['inventory']['complete'])
self.assertEqual(result['workload']['video_files'], 1100)
self.assertEqual(result['workload']['jav_titles'], 1)
self.assertNotIn(token, _SESSIONS)
def test_selected_assets_and_collection_do_not_list_directories(self):
with closing(sqlite3.connect(self.db)) as connection, connection:
connection.execute("INSERT INTO asset(location,path,name,medium,size) VALUES(?,?,?,?,?)",
('mounted', 'A:\\Cloud\\ABW-001.mp4', 'ABW-001.mp4', 'video', 1024))
for stage, asset_ids in (('all', [1]), ('collect', None)):
result = preview(self.config, self.db, stage, asset_ids)
self.assertTrue(result['inventory']['complete'])
self.assertEqual(result['workload']['video_files'], 1)
self.walk.assert_not_called()
def test_full_library_inventory_discovers_new_files_beside_registered_files(self):
with closing(sqlite3.connect(self.db)) as connection, connection:
connection.execute("INSERT INTO asset(location,path,name,medium,size) VALUES(?,?,?,?,?)",
('mounted', 'A:\\Cloud\\ABW-001.mp4', 'ABW-001.mp4', 'video', 1024))
(self.media/'ABW-001.mp4').touch()
(self.media/'ABW-002.mp4').touch()
result = preview(self.config,self.db,'all',None)
self.assertTrue(result['inventory']['complete'])
self.assertEqual(result['workload']['video_files'],2)
def test_unreadable_directory_reports_incomplete_counts(self):
with patch('peach.scan.os.scandir', side_effect=PermissionError('拒绝访问')):
result = preview(self.config, self.db, 'all', None)
self.assertTrue(result['inventory']['complete'])
self.assertEqual(result['inventory']['errors'], 1)
self.assertEqual(result['workload']['video_files'], 0)
def test_invalid_continuation_is_rejected(self):
with self.assertRaisesRegex(ValueError, '预检已结束'):
preview(self.config, self.db, 'scan', None, 'invalid')
def test_incremental_preview_finds_additions_in_a_root_with_completed_assets(self):
for name in ('old.mp4','new.mp4'):
(self.media/name).touch()
with closing(sqlite3.connect(self.db)) as connection, connection:
connection.execute("INSERT INTO asset(location,path,name,medium) VALUES(?,?,?,?)",
('mounted','A:\\Cloud\\old.mp4','old.mp4','video'))
result = preview(self.config,self.db,'all',None,incremental=True)
self.assertEqual(result['workload']['video_files'],1)
self.assertTrue(result['inventory']['complete'])
def test_started_task_uses_recent_inventory_and_registered_rows_take_precedence(self):
(self.media / 'ABW-001.mp4').touch()
before = preview(self.config, self.db, 'all', None)
rows = recent_rows(self.config, self.db, 'all', None)
task = ProcessingTraffic(self.config, self.db, 'all', inventory_rows=rows)
self.assertEqual(task.snapshot()['workload'], before['workload'])
self.assertEqual(task.snapshot()['estimated_bytes'], before['estimated_bytes'])
with closing(sqlite3.connect(self.db)) as connection, connection:
connection.execute("INSERT INTO asset(location,path,name,medium,size,duration) VALUES(?,?,?,?,?,?)",
('mounted', 'A:\\Cloud\\ABW-001.mp4', 'ABW-001.mp4', 'video', 1234, 3600))
task = ProcessingTraffic(self.config, self.db, 'all', inventory_rows=rows)
self.assertEqual(task.snapshot()['workload']['video_files'], 1)
self.assertEqual(task.plan.assets[0]['size'], 1234)
self.assertEqual(recent_rows(self.config, self.db, 'scan', None), [])