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()