207 lines
9.5 KiB
Python
207 lines
9.5 KiB
Python
import contextlib
|
||
import io
|
||
import json
|
||
from pathlib import Path
|
||
import tempfile
|
||
import sqlite3
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
import numpy as np
|
||
from PIL import Image
|
||
import photo_sort as s
|
||
|
||
|
||
class SortTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.base = Path(self.tmp.name).resolve()
|
||
self.source = self.base/'incoming'
|
||
self.dest = self.base/'archive'
|
||
self.source.mkdir(); self.dest.mkdir()
|
||
self.run = self.base/'run'
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def photo(self, path, seed=1, date=True, size=(160,120)):
|
||
image = Image.fromarray(np.random.default_rng(seed).integers(0,256,(size[1],size[0],3),dtype=np.uint8))
|
||
exif = Image.Exif()
|
||
if date: exif[36867]='2025:05:20 10:10:20'
|
||
image.save(path, exif=exif)
|
||
|
||
def analyze(self, *extra):
|
||
with contextlib.redirect_stdout(io.StringIO()):
|
||
s.main(['analyze','--source',str(self.source),'--dest',str(self.dest),'--run',str(self.run),'--database',str(self.base/'catalog.sqlite'),*extra])
|
||
return s.load_review(self.run)
|
||
|
||
def apply(self):
|
||
with contextlib.redirect_stdout(io.StringIO()):
|
||
s.apply(self.run, yes=True)
|
||
|
||
def test_dry_run_collisions_and_repeat(self):
|
||
self.photo(self.source/'one.jpg')
|
||
self.photo(self.source/'two.jpg',seed=2)
|
||
self.photo(self.source/'copy.jpg')
|
||
self.analyze()
|
||
self.assertEqual(len(list(self.source.iterdir())),3)
|
||
self.apply()
|
||
self.assertEqual(len(list(self.source.iterdir())),0)
|
||
names={p.name for p in (self.dest/'2025'/'май').iterdir()}
|
||
self.assertEqual(names,{'20052025101020.jpg','20052025101020_dub.jpg'})
|
||
self.assertEqual(len(list((self.dest/'_DEL').iterdir())),1)
|
||
with self.assertRaises(RuntimeError): self.apply()
|
||
self.run=self.base/'run2'
|
||
plan=self.analyze()
|
||
self.assertFalse(s.targets(plan))
|
||
|
||
def test_stale_source_and_retained_original(self):
|
||
self.photo(self.source/'one.jpg')
|
||
self.photo(self.dest/'existing.jpg',seed=2)
|
||
self.analyze()
|
||
(self.dest/'existing.jpg').write_bytes(b'changed')
|
||
with self.assertRaises(RuntimeError): self.apply()
|
||
self.assertTrue((self.source/'one.jpg').exists())
|
||
self.assertFalse((self.run/'moves.jsonl').exists())
|
||
|
||
def test_manual_date_and_nonphotos(self):
|
||
self.photo(self.source/'undated.jpg',date=False)
|
||
(self.source/'notes.txt').write_text('hello')
|
||
(self.source/'broken.jpg').write_bytes(b'not a valid JPEG')
|
||
plan=self.analyze()
|
||
row=next(r for r in plan['rows'] if r['photo'])
|
||
self.assertEqual(row['action'],'ras')
|
||
row.update(action='sort',date='2026-01-02T03:04:05')
|
||
s.write_review(self.run,plan['rows'])
|
||
self.apply()
|
||
self.assertTrue((self.dest/'2026'/'январь'/'02012026030405.jpg').exists())
|
||
self.assertTrue((self.dest/'_RAS'/'notes.txt').exists())
|
||
self.assertTrue((self.dest/'_DEL'/'broken.jpg').exists())
|
||
|
||
def test_bigger_resolution_replaces_existing(self):
|
||
y,x=np.mgrid[:480,:640]
|
||
pixels=np.stack([128+60*np.sin(x/30),128+60*np.cos(y/33),128+60*np.sin((x+y)/45)],axis=2).astype('uint8')
|
||
image=Image.fromarray(pixels);exif=Image.Exif();exif[36867]='2025:05:20 10:10:20'
|
||
image.save(self.source/'large.jpg',quality=98,exif=exif)
|
||
image.resize((320,240)).save(self.dest/'small.jpg',quality=95,exif=exif)
|
||
plan=self.analyze('--visual')
|
||
small=next(r for r in plan['rows'] if r['existing'])
|
||
self.assertEqual(small['action'],'del')
|
||
self.assertTrue(small['winner'].endswith('large.jpg'))
|
||
self.apply()
|
||
self.assertTrue((self.dest/'_DEL'/'small.jpg').exists())
|
||
|
||
def test_ras_explicit_source_and_selection(self):
|
||
self.source=self.dest/'_RAS';self.source.mkdir()
|
||
self.photo(self.source/'first.jpg',date=False)
|
||
self.photo(self.source/'second.jpg',date=False)
|
||
plan=self.analyze('--select','first.jpg','--date','2025-05-20 10:10:20')
|
||
self.assertEqual(len(plan['rows']),1)
|
||
self.apply()
|
||
self.assertTrue((self.source/'second.jpg').exists())
|
||
self.assertFalse((self.source/'first.jpg').exists())
|
||
|
||
def test_alpha_is_not_lost(self):
|
||
a=Image.new('RGBA',(20,20),(20,30,40,0))
|
||
b=Image.new('RGBA',(20,20),(20,30,40,255))
|
||
a.save(self.source/'a.png');b.save(self.source/'b.png')
|
||
plan=self.analyze('--visual')
|
||
self.assertTrue(all(r['action']=='ras' for r in plan['rows']))
|
||
|
||
def test_existing_ras_matches_new_batch(self):
|
||
ras = self.dest/'_RAS'
|
||
ras.mkdir()
|
||
self.photo(ras/'original.jpg', date=False)
|
||
self.photo(self.source/'duplicate.jpg', date=False)
|
||
plan = self.analyze()
|
||
incoming = next(r for r in plan['rows'] if not r['existing'])
|
||
self.assertEqual(incoming['action'], 'del')
|
||
self.assertEqual(incoming['winner'], str(ras/'original.jpg'))
|
||
self.apply()
|
||
self.assertTrue((ras/'original.jpg').exists())
|
||
self.assertTrue((self.dest/'_DEL'/'duplicate.jpg').exists())
|
||
|
||
def test_new_destination_never_overwrites(self):
|
||
self.photo(self.source/'one.jpg')
|
||
self.analyze()
|
||
d=self.dest/'2025'/'май';d.mkdir(parents=True)
|
||
original=d/'20052025101020.jpg';original.write_bytes(b'keep me')
|
||
self.apply()
|
||
self.assertEqual(original.read_bytes(),b'keep me')
|
||
self.assertTrue((d/'20052025101020_dub.jpg').exists())
|
||
|
||
def test_cache_invalidates_changed_file(self):
|
||
path=self.source/'photo.jpg';self.photo(path)
|
||
with sqlite3.connect(':memory:') as db:
|
||
db.execute('CREATE TABLE files (path TEXT PRIMARY KEY, signature TEXT, data TEXT)')
|
||
with patch.object(s,'inspect',wraps=s.inspect) as spy:
|
||
first=s.cached_inspect(path,{},db)
|
||
second=s.cached_inspect(path,{},db)
|
||
self.assertEqual(first['sha'],second['sha'])
|
||
self.assertEqual(spy.call_count,1)
|
||
self.photo(path,seed=12)
|
||
third=s.cached_inspect(path,{},db)
|
||
self.assertNotEqual(first['sha'],third['sha'])
|
||
self.assertEqual(spy.call_count,2)
|
||
|
||
def test_no_date_inheritance(self):
|
||
self.photo(self.source/'large.jpg', date=False)
|
||
# Те же декодированные пиксели, но меньший файл со своей датой.
|
||
a=s.inspect(self.source/'large.jpg',{})
|
||
b=dict(a,source=str(self.dest/'old.jpg'),size=a['size']-1,date='2012-01-01T12:00:00',sha='different')
|
||
a.update(existing=False,id='1',group_id='new',winner='')
|
||
b.update(existing=True,id='2',group_id='old',winner='')
|
||
s.classify([a,b],True)
|
||
self.assertEqual(a['date'],'')
|
||
self.assertEqual(a['action'],'ras')
|
||
self.assertEqual(b['action'],'del')
|
||
self.assertEqual(a['group_id'],b['group_id'])
|
||
|
||
def test_conflicting_embedded_dates(self):
|
||
row={}
|
||
s.resolve_dates(row,{'EXIF:CreateDate':'2018:10:15 09:30:16','EXIF:DateTimeOriginal':'2012:08:28 18:47:41'})
|
||
self.assertEqual(row['date'],'')
|
||
self.assertEqual(len(row['date_candidates']),2)
|
||
|
||
def test_replacement_updates_catalog_and_reuses_name(self):
|
||
self.photo(self.source/'first.jpg')
|
||
self.analyze(); self.apply()
|
||
original=self.dest/'2025'/'май'/'20052025101020.jpg'
|
||
original_sha=s.digest(original)
|
||
# Дополнительные байты не меняют пиксели JPEG, но увеличивают входящий файл.
|
||
(self.source/'better.jpg').write_bytes(original.read_bytes()+b'extra bytes')
|
||
self.run=self.base/'run2';plan=self.analyze()
|
||
ops=s.targets(plan)
|
||
self.assertEqual(ops[-1]['target'],str(original))
|
||
self.apply()
|
||
self.assertNotEqual(s.digest(original),original_sha)
|
||
self.assertTrue((self.dest/'_DEL'/'20052025101020.jpg').exists())
|
||
with sqlite3.connect(self.base/'catalog.sqlite') as db:
|
||
live=db.execute("SELECT path,group_id,state FROM catalog WHERE state IN ('archive','del')").fetchall()
|
||
self.assertEqual(len({r[1] for r in live}),1)
|
||
self.assertEqual(db.execute('SELECT count(*) FROM history').fetchone()[0],3)
|
||
self.assertEqual(db.execute('SELECT sha FROM catalog WHERE path=?',(str(original),)).fetchone()[0],s.digest(original))
|
||
|
||
def test_manually_moved_original_is_reconciled(self):
|
||
self.photo(self.source/'first.jpg');self.analyze();self.apply()
|
||
original=self.dest/'2025'/'май'/'20052025101020.jpg'
|
||
renamed=original.with_name('manual.jpg');original.rename(renamed)
|
||
(self.source/'copy.jpg').write_bytes(renamed.read_bytes())
|
||
self.run=self.base/'run2';plan=self.analyze()
|
||
incoming=next(r for r in plan['rows'] if not r['existing'])
|
||
self.assertEqual(incoming['action'],'del')
|
||
self.assertEqual(incoming['winner'],str(renamed))
|
||
renamed.unlink()
|
||
self.run=self.base/'run3';plan=self.analyze()
|
||
self.assertEqual(plan['rows'][0]['action'],'sort')
|
||
|
||
def test_catalog_matches_ras_and_retains_manual_date(self):
|
||
self.photo(self.source/'first.jpg',date=False)
|
||
self.analyze('--date','2018-10-15 09:30:16');self.apply()
|
||
self.run=self.base/'run2';plan=self.analyze()
|
||
self.assertEqual(plan['rows'][0]['date'],'2018-10-15T09:30:16')
|
||
self.assertEqual(plan['rows'][0]['action'],'keep')
|
||
|
||
|
||
if __name__=='__main__': unittest.main()
|