170 lines
8.3 KiB
Python
170 lines
8.3 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from queue import Queue
|
|
from unittest.mock import Mock, patch
|
|
|
|
from PIL import Image
|
|
|
|
from copienator import EvaluationWorkspace, ExitCode, atomic_write_json, read_json
|
|
from copienator.commands import cutleft, gemini_for_labels, plotting
|
|
from copienator.copy_errors import clear_copy_error, copy_errors, mark_copy_error, marked_copy_paths
|
|
|
|
|
|
class LabelCopyFlagTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.workspace = EvaluationWorkspace(Path(self.temp.name))
|
|
self.workspace.copies_dir.mkdir()
|
|
self.workspace.cutleft_dir.mkdir()
|
|
self.pdf = self.workspace.copies_dir / 'Copie01.pdf'
|
|
self.pdf.touch()
|
|
|
|
def flag(self, source):
|
|
mark_copy_error(self.workspace, self.pdf, source, source=source)
|
|
|
|
def viewer(self, parts=2):
|
|
viewer = plotting.ImageViewer.__new__(plotting.ImageViewer)
|
|
viewer.workspace = self.workspace
|
|
viewer.base_dir = self.workspace.root
|
|
viewer.active_copie_name = 'Copie01'
|
|
viewer.accumulated_results = {'name': 'Student', 'list': []}
|
|
viewer.current_meta = {'copie': 'Copie01', 'part': 1,
|
|
'schema': {'columns_per_file': [1] * parts}}
|
|
viewer.current_json_path = self.pdf.with_name('Copie01_01.json')
|
|
atomic_write_json(viewer.current_json_path, {'name': 'Student', 'list': []})
|
|
viewer.valid_labels = set()
|
|
viewer.history = []
|
|
viewer.forward_stack = []
|
|
viewer.skipped_copies = set()
|
|
viewer.current_pil_image = Image.new('RGB', (20, 20), 'white')
|
|
viewer.is_viewing = True
|
|
viewer.failed = False
|
|
viewer.label = Mock()
|
|
viewer.root = Mock()
|
|
viewer.image_queue = Queue()
|
|
return viewer
|
|
|
|
def test_legacy_flags_migrate_and_steps_clear_independently(self):
|
|
atomic_write_json(self.workspace.metadata_dir / 'copy_errors.json',
|
|
{self.pdf.name: 'Legacy cropping problem'})
|
|
self.flag('plotting')
|
|
clear_copy_error(self.workspace, self.pdf)
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'plotting'})
|
|
self.flag('cutleft')
|
|
clear_copy_error(self.workspace, self.pdf, source='plotting')
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'cutleft'})
|
|
clear_copy_error(self.workspace, self.pdf)
|
|
self.assertEqual(marked_copy_paths(self.workspace), [])
|
|
|
|
def test_successful_crop_preserves_review_flag_and_selection(self):
|
|
self.flag('cutleft')
|
|
self.flag('plotting')
|
|
image = Image.new('RGB', (20, 20), 'white')
|
|
with patch.object(cutleft, 'process_single_pdf', return_value=(image, [image], {})), \
|
|
patch.object(cutleft, 'save_results'):
|
|
self.assertEqual(cutleft.run(self.workspace, self.workspace.root, marked=True,
|
|
fullpage=True, no_visual_check=True), ExitCode.SUCCESS)
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'plotting'})
|
|
self.assertEqual(marked_copy_paths(self.workspace), [self.pdf])
|
|
|
|
def test_full_review_clears_only_review_flag_after_save(self):
|
|
self.flag('cutleft')
|
|
self.flag('plotting')
|
|
viewer = self.viewer()
|
|
viewer.on_enter(None)
|
|
viewer.current_meta = dict(viewer.current_meta, part=2)
|
|
viewer.is_viewing = True
|
|
viewer.on_enter(None)
|
|
with patch.object(plotting, 'atomic_write_json', side_effect=OSError('disk full')):
|
|
with self.assertRaises(OSError):
|
|
viewer.save_current_batch()
|
|
self.assertIn('plotting', copy_errors(self.workspace)[self.pdf.name])
|
|
viewer.save_current_batch()
|
|
self.assertEqual(read_json(self.pdf.with_suffix('.json')), {'name': 'Student', 'list': []})
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'cutleft'})
|
|
|
|
def test_incomplete_review_preserves_previous_aggregate_and_flag(self):
|
|
self.flag('plotting')
|
|
old = {'name': 'Old', 'list': [{'label': 'Ex 1'}]}
|
|
atomic_write_json(self.pdf.with_suffix('.json'), old)
|
|
viewer = self.viewer()
|
|
viewer.on_enter(None)
|
|
viewer.save_current_batch()
|
|
self.assertEqual(read_json(self.pdf.with_suffix('.json')), old)
|
|
self.assertIn(self.pdf.name, copy_errors(self.workspace))
|
|
self.assertTrue(viewer.failed)
|
|
|
|
def test_flag_skips_remaining_copy_parts_without_saving_partial_data(self):
|
|
self.flag('cutleft')
|
|
old = {'name': 'Old', 'list': []}
|
|
atomic_write_json(self.pdf.with_suffix('.json'), old)
|
|
viewer = self.viewer(parts=3)
|
|
viewer.on_enter(None)
|
|
viewer.is_viewing = True
|
|
viewer.current_meta = dict(viewer.current_meta, part=2)
|
|
viewer.on_skip()
|
|
viewer.save_current_batch()
|
|
self.assertEqual(read_json(self.pdf.with_suffix('.json')), old)
|
|
self.assertEqual(set(read_json(self.workspace.metadata_dir / 'copy_errors.json')[self.pdf.name]),
|
|
{'cutleft', 'plotting'})
|
|
viewer.image_queue.put((viewer.current_pil_image, viewer.current_json_path,
|
|
dict(viewer.current_meta, part=3)))
|
|
viewer.display_image = Mock()
|
|
viewer.poll_queue()
|
|
viewer.display_image.assert_not_called()
|
|
self.assertFalse(viewer.is_viewing)
|
|
|
|
def test_detection_regenerates_only_flagged_copies_and_preserves_flags(self):
|
|
self.flag('plotting')
|
|
(self.workspace.copies_dir / 'Copie02.pdf').touch()
|
|
for name in ('Copie01_01', 'Copie02_01'):
|
|
(self.workspace.cutleft_dir / f'{name}.jpg').touch()
|
|
atomic_write_json(self.workspace.copies_dir / f'{name}.json', {'name': 'Old', 'list': []})
|
|
with patch.object(self.workspace.__class__, 'require_files'), \
|
|
patch.object(gemini_for_labels, 'read_all_labels', return_value=[]), \
|
|
patch.object(self.workspace.__class__, 'names_file', return_value=self.pdf), \
|
|
patch.object(gemini_for_labels, 'process_copy_group', return_value=1) as process:
|
|
status = gemini_for_labels.run(self.workspace, [self.workspace.root], marked=True, client=Mock())
|
|
self.assertEqual(status, ExitCode.SUCCESS)
|
|
process.assert_called_once()
|
|
self.assertEqual(process.call_args.args[1], 'Copie01')
|
|
self.assertTrue(process.call_args.kwargs['overwrite'])
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'plotting'})
|
|
|
|
def test_visual_review_selects_only_flagged_copy_images(self):
|
|
self.flag('plotting')
|
|
for name in ('Copie01_01', 'Copie01_02', 'Copie02_01'):
|
|
(self.workspace.cutleft_dir / f'{name}.jpg').touch()
|
|
with patch.object(self.workspace.__class__, 'require_files'), \
|
|
patch.object(plotting, 'read_all_labels', return_value=[]), \
|
|
patch.object(plotting.threading, 'Thread') as worker, \
|
|
patch.object(plotting.tk, 'Tk'), \
|
|
patch.object(plotting, 'ImageViewer') as viewer:
|
|
viewer.return_value.failed = False
|
|
self.assertEqual(plotting.run(self.workspace, self.workspace.root, marked=True),
|
|
ExitCode.SUCCESS)
|
|
selected = worker.call_args.kwargs['args'][1]
|
|
self.assertEqual([image.name for image in selected], ['Copie01_01.jpg', 'Copie01_02.jpg'])
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'plotting'})
|
|
|
|
def test_closing_review_preserves_flags_and_does_not_save_partial_result(self):
|
|
self.flag('plotting')
|
|
viewer = self.viewer()
|
|
viewer.on_enter(None)
|
|
viewer.close()
|
|
self.assertEqual(copy_errors(self.workspace), {self.pdf.name: 'plotting'})
|
|
self.assertFalse(self.pdf.with_suffix('.json').exists())
|
|
|
|
def test_empty_marked_selection_does_not_process_whole_evaluation(self):
|
|
with patch.object(self.workspace.__class__, 'require_files'), \
|
|
patch.object(gemini_for_labels, 'selected_images') as select:
|
|
self.assertEqual(gemini_for_labels.run(self.workspace, [self.workspace.root], marked=True),
|
|
ExitCode.SUCCESS)
|
|
select.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|