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