198 lines
9.3 KiB
Python
198 lines
9.3 KiB
Python
import os
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from queue import Queue
|
|
from unittest.mock import Mock, patch
|
|
|
|
from PIL import Image
|
|
|
|
from copienator import CliError, EvaluationWorkspace, ExitCode
|
|
from copienator.copy_errors import copy_errors, clear_copy_error, mark_copy_error, marked_copy_paths
|
|
from copienator.commands import cutleft, page_splitter
|
|
from copienator_gui.app import CopienatorApp
|
|
|
|
|
|
class MarkedCopiesTests(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.original_copies_dir.mkdir()
|
|
self.files = [self.workspace.copies_dir / name for name in ("Copie01.pdf", "Copie02.pdf")]
|
|
for path in self.files:
|
|
path.write_bytes(b"processed")
|
|
(self.workspace.original_copies_dir / path.name).write_bytes(b"original")
|
|
|
|
def reviewer(self):
|
|
review = cutleft.ImageReviewer.__new__(cutleft.ImageReviewer)
|
|
review.workspace = self.workspace
|
|
review.files = self.files
|
|
review.output_dir = self.workspace.cutleft_dir
|
|
review.index = 0
|
|
review.is_processing = False
|
|
review.had_errors = False
|
|
review.completed = False
|
|
review.current_shift = 50
|
|
review.default_max_per_file = 5
|
|
review.current_max_per_file = 1
|
|
review.root = Mock()
|
|
review.stop_prefetch = threading.Event()
|
|
review.load_current_image = Mock()
|
|
review.update_display = Mock()
|
|
image = Image.new("RGB", (8, 8), "white")
|
|
review.current_result = (image, [image], {"total_pages": 1, "columns_per_file": [1]})
|
|
return review
|
|
|
|
def test_marks_survive_reload_and_resolve_originals_only_for_splitting(self):
|
|
mark_copy_error(self.workspace, self.files[1], "Wrong page order")
|
|
reloaded = EvaluationWorkspace(self.workspace.root)
|
|
self.assertEqual(marked_copy_paths(reloaded), [self.files[1]])
|
|
original = self.workspace.original_copies_dir / self.files[1].name
|
|
self.assertEqual(marked_copy_paths(reloaded, originals=True), [original])
|
|
self.files[1].unlink()
|
|
with self.assertRaises(CliError):
|
|
marked_copy_paths(reloaded)
|
|
self.assertEqual(marked_copy_paths(reloaded, originals=True), [original])
|
|
clear_copy_error(reloaded, original)
|
|
self.assertEqual(copy_errors(self.workspace), {})
|
|
|
|
def test_skip_flags_and_advances_without_replacing_existing_crop(self):
|
|
review = self.reviewer()
|
|
review.output_dir.mkdir()
|
|
previous = review.output_dir / "Copie01_01.jpg"
|
|
previous.write_bytes(b"previous crop")
|
|
review.handle_processing_result(review.current_result, self.files[0])
|
|
review.on_skip()
|
|
self.assertEqual(previous.read_bytes(), b"previous crop")
|
|
self.assertIn("Copie01.pdf", copy_errors(self.workspace))
|
|
self.assertEqual(review.index, 1)
|
|
self.assertEqual(review.current_max_per_file, 5)
|
|
self.assertTrue(review.had_errors)
|
|
review.load_current_image.assert_called_once_with()
|
|
|
|
def test_accept_saves_and_clears_only_current_flag(self):
|
|
for path in self.files:
|
|
mark_copy_error(self.workspace, path, "Review needed")
|
|
review = self.reviewer()
|
|
review.on_next(None)
|
|
self.assertTrue((review.output_dir / "Copie01_01.jpg").is_file())
|
|
self.assertEqual(set(copy_errors(self.workspace)), {"Copie02.pdf"})
|
|
self.assertEqual(review.index, 1)
|
|
|
|
def test_failed_save_and_window_close_preserve_flag(self):
|
|
mark_copy_error(self.workspace, self.files[0], "Review needed")
|
|
review = self.reviewer()
|
|
with patch.object(cutleft, "save_results", side_effect=OSError("disk full")), patch.object(
|
|
cutleft.messagebox, "showerror"
|
|
):
|
|
review.on_next(None)
|
|
self.assertEqual(review.index, 0)
|
|
self.assertIn("Copie01.pdf", copy_errors(self.workspace))
|
|
review.on_close()
|
|
self.assertFalse(review.completed)
|
|
self.assertTrue(review.stop_prefetch.is_set())
|
|
self.assertIn("Copie01.pdf", copy_errors(self.workspace))
|
|
|
|
def test_processing_blocks_skip_and_failed_conversion_is_flagged(self):
|
|
review = self.reviewer()
|
|
review.is_processing = True
|
|
review.on_skip()
|
|
self.assertEqual(review.index, 0)
|
|
self.assertEqual(copy_errors(self.workspace), {})
|
|
review.manual_queue = Queue()
|
|
review.manual_queue.put(None)
|
|
review.load_current_image.side_effect = lambda: setattr(review, "is_processing", True)
|
|
review.check_manual_queue(self.files[0])
|
|
self.assertIn("Copie01.pdf", copy_errors(self.workspace))
|
|
self.assertEqual(review.index, 1)
|
|
self.assertTrue(review.is_processing) # Still loading the next copy.
|
|
|
|
def test_cli_splitting_preserves_flags_and_cropping_reports_partial_or_interrupted(self):
|
|
mark_copy_error(self.workspace, self.files[1], "Wrong order")
|
|
with patch.object(page_splitter.tk, "Tk"), patch.object(page_splitter, "PDFPreviewer") as preview:
|
|
preview.return_value.failed = False
|
|
self.assertEqual(page_splitter.main([str(self.workspace.root), "--marked"]), 0)
|
|
self.assertEqual(preview.call_args.args[2], [self.workspace.original_copies_dir / "Copie02.pdf"])
|
|
self.assertIn("Copie02.pdf", copy_errors(self.workspace))
|
|
with patch.object(cutleft, "ImageReviewer") as reviewer:
|
|
for completed, errors, expected in ((True, True, ExitCode.PARTIAL),
|
|
(False, False, ExitCode.INTERRUPTED),
|
|
(True, False, ExitCode.SUCCESS)):
|
|
reviewer.return_value.completed = completed
|
|
reviewer.return_value.had_errors = errors
|
|
self.assertEqual(cutleft.main([str(self.workspace.root), "--marked"]), expected)
|
|
self.assertEqual(reviewer.call_args.args[0], [self.files[1]])
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("DISPLAY"), "Tk requires a display")
|
|
class MarkedCopiesGuiTests(unittest.TestCase):
|
|
setUp = MarkedCopiesTests.setUp
|
|
|
|
def test_keyboard_skip_then_accept_and_retry_clears_flag(self):
|
|
real_tk = cutleft.tk.Tk
|
|
|
|
def review_with_keys(files, keys):
|
|
sent = []
|
|
|
|
def make_root():
|
|
root = real_tk()
|
|
|
|
def send_when_ready():
|
|
info = root.winfo_children()[-1].cget("text")
|
|
index = len(sent)
|
|
if index < len(keys) and info.startswith(f"[{index + 1}/{len(files)}]"):
|
|
root.focus_force()
|
|
sent.append(keys[index])
|
|
root.event_generate(keys[index])
|
|
if len(sent) < len(keys):
|
|
root.after(10, send_when_ready)
|
|
|
|
root.after(20, send_when_ready)
|
|
root.after(3000, root.destroy) # Bound a failed keyboard test.
|
|
return root
|
|
|
|
with patch.object(cutleft.tk, "Tk", side_effect=make_root), patch.object(
|
|
cutleft, "get_pdf_pages", return_value=[Image.new("RGB", (600, 300), "white")]
|
|
), patch.object(cutleft, "OUTPUT_SIZE", (400, 200)):
|
|
reviewer = cutleft.ImageReviewer(files, self.workspace.cutleft_dir)
|
|
self.assertEqual(sent, keys)
|
|
self.assertTrue(reviewer.completed)
|
|
return reviewer
|
|
|
|
first = review_with_keys(self.files, ["<KeyPress-s>", "<Return>"])
|
|
self.assertTrue(first.had_errors)
|
|
self.assertEqual(set(copy_errors(self.workspace)), {"Copie01.pdf"})
|
|
self.assertFalse((self.workspace.cutleft_dir / "Copie01_01.jpg").exists())
|
|
self.assertTrue((self.workspace.cutleft_dir / "Copie02_01.jpg").exists())
|
|
second = review_with_keys(marked_copy_paths(self.workspace), ["<Return>"])
|
|
self.assertFalse(second.had_errors)
|
|
self.assertEqual(copy_errors(self.workspace), {})
|
|
self.assertTrue((self.workspace.cutleft_dir / "Copie01_01.jpg").exists())
|
|
|
|
def test_both_buttons_run_marked_and_single_copy_target_clears_filter(self):
|
|
mark_copy_error(self.workspace, self.files[1], "Review needed")
|
|
app = CopienatorApp(Path.cwd(), False, self.workspace.root)
|
|
try:
|
|
app.update()
|
|
for ident, command in (("page_splitter", "page-split"), ("cutleft", "crop-labels")):
|
|
app.tree.selection_set(ident)
|
|
app.update()
|
|
self.assertIn("(1)", app.marked_copies_button.cget("text"))
|
|
with patch.object(app, "_run_current_step") as run:
|
|
app.marked_copies_button.invoke()
|
|
run.assert_called_once_with()
|
|
self.assertIn(command, app._make_command())
|
|
self.assertIn("--marked", app._make_command())
|
|
self.assertEqual(app.arg_vars["target"].get(), app._evaluation_arg())
|
|
app.copy_var.set("Copie01.pdf")
|
|
app._target_selected_copy()
|
|
self.assertNotIn("--marked", app._make_command())
|
|
self.assertIn(str(self.files[0]), app._make_command())
|
|
finally:
|
|
for callback in app.tk.splitlist(app.tk.call("after", "info")):
|
|
app.after_cancel(callback)
|
|
app.destroy()
|