from __future__ import annotations from io import BytesIO, StringIO import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch from fastapi.testclient import TestClient from PIL import Image from copienator import CliError, EvaluationWorkspace, atomic_write_json, read_json from copienator.annotation_modifications import APPLIED, DRAFT, stage_baseline from copienator.commands import modify_annotations, giving_names from copienator.evaluation_lock import EvaluationLock from copienator.filesystem import staged_files from copienator.finalisation import Finalisation from copienator.finalisation_server import create_app from copienator.return_identity import rename_return_copy from copienator.score_editing import read_scores, set_score def seed(root, mode='Bnot'): (root / 'Copies' / 'Copie01').mkdir(parents=True, exist_ok=True) (root / 'labels').write_text('Ex 1\nEx 2\nAbsent\n') atomic_write_json(root / 'Copies' / 'Copie01.json', {'name': 'Unknown'}) scores = {'Ex 1': '3', 'Ex 2': '4', 'Absent': ''} info = {label: {'present': label != 'Absent', 'not_empty': label != 'Absent', 'touched': False, 'score': score} for label, score in scores.items()} images = {'Ex 1': Image.new('RGB', (100, 120), 'blue'), 'Ex 2': Image.new('RGB', (100, 80), 'green')} output = root / mode / 'Copie01' with staged_files(output) as staging: atomic_write_json(staging / 'score.json', scores) atomic_write_json(staging / 'info.json', info) for label, image in images.items(): image.save(staging / f'{label}.jpg') images['Ex 1'].save(staging / 'Concat.jpg') stage_baseline(root, mode, output, staging, images, {'Ex 1': 20, 'Ex 2': 10}, info, [], scores) class WebTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) seed(self.root) self.service = Finalisation(self.root, 'Bnot', 'worker-token') self.client = TestClient(create_app(self.service, 'browser-token')) self.addCleanup(self.client.close) self.client.get('/?token=browser-token') def save(self, kind, value, copy='01', label='Ex 1'): response = self.client.post('/api/change', json={'kind': kind, 'value': value, 'copy': copy, 'label': label, 'revision': self.service.revision()}) self.assertEqual(response.status_code, 200, response.text) return response.json() def test_session_origin_and_paths(self): outsider = TestClient(create_app(self.service, 'browser-token')) self.assertEqual(outsider.get('/api/state').status_code, 403) self.assertEqual(self.client.get('/api/state', headers={'Host': 'evil.example'}).status_code, 403) self.assertEqual(self.client.post('/api/stop', headers={'Origin':'https://evil.example'}).status_code, 403) self.assertEqual(self.client.get('/api/image', params={'copy':'../', 'label':'Ex 1'}).status_code, 409) self.assertEqual(self.client.get('/api/image', params={'copy':'01', 'label':'../labels'}).status_code, 409) self.assertEqual(self.client.get('/api/document', params={'copy':'01', 'kind':'question','label':'Absent'}).status_code, 404) self.assertEqual(self.client.get('/').status_code, 200) def test_browser_and_existing_editor_share_scores_and_remarks(self): for mode in ('Bnot', 'BGnot'): with self.subTest(mode=mode): seed(self.root, mode) self.service = Finalisation(self.root, mode, 'worker-token') self.client = TestClient(create_app(self.service, 'browser-token')) self.client.get('/?token=browser-token') self.save('score', '2,5') score = self.root / mode / 'Copie01/score.json' self.assertEqual(read_scores(score)['Ex 1'], '2.5') set_score(score, 'Ex 1', '1') self.assertEqual(self.service.state()['copies'][0]['answers'][0]['score'], '1') self.save('remark', 'Une remarque $x^2$.') self.save('general', ['Première', 'Deuxième']) self.save('general', ['Deuxième', 'Première']) self.assertEqual(self.service.state()['copies'][0]['general'], ['Deuxième','Première']) self.save('remark', '') self.assertEqual(self.service.state()['copies'][0]['answers'][0]['remark'], '') def test_stale_tab_and_busy_writes_rejected(self): old = self.service.revision() self.save('score', '2') response = self.client.post('/api/change', json={'kind':'score','value':'1','copy':'01','label':'Ex 1','revision':old}) self.assertEqual(response.status_code, 409) self.service.job['running'] = True self.assertEqual(self.client.post('/api/stop').status_code, 409) response = self.client.post('/api/change', json={'kind':'score','value':'1','copy':'01','label':'Ex 1','revision':self.service.revision()}) self.assertEqual(response.status_code, 409) self.service.job['running'] = False def test_draft_preview_apply_reapply_and_undo(self): text = '### Source: Bnot\nCopie01 Ex 1 c{40}1> Ex 2|\n' before = (self.root / 'Bnot/Copie01/Ex 1.jpg').read_bytes() self.save('draft', text) self.assertEqual((self.root / 'Bnot/Copie01/Ex 1.jpg').read_bytes(), before) self.assertFalse((self.root / 'Bnot' / APPLIED).exists()) result = self.client.post('/api/preview', json={'copy':'01','source':'Ex 1','target':'Ex 2','text':text}) self.assertEqual(result.status_code, 200, result.text) import base64 rendered = Image.open(BytesIO(base64.b64decode(result.json()['Ex 1'].split(',')[1]))) self.assertEqual(rendered.height, 40) # cut uses 100px body, not 120px image modify_annotations.run(EvaluationWorkspace(self.root), annotation_dir='Bnot') applied = (self.root / 'Bnot/Copie01/Ex 2.jpg').read_bytes() modify_annotations.run(EvaluationWorkspace(self.root), annotation_dir='Bnot') self.assertEqual((self.root / 'Bnot/Copie01/Ex 2.jpg').read_bytes(), applied) self.save('draft', '### Source: Bnot\n') self.assertEqual((self.root / 'Bnot' / APPLIED).read_text(), text) modify_annotations.run(EvaluationWorkspace(self.root), annotation_dir='Bnot') self.assertEqual(Image.open(self.root / 'Bnot/Copie01/Ex 1.jpg').height, 120) def test_invalid_draft_preserves_previous_file(self): self.save('draft', '### Source: Bnot\n') for text in ('### Source: BGnot\n', '### Source: Bnot\nCopie01 Ex 1 -> Ex 2|\nCopie01 Ex 2 -> Absent|\n'): response = self.client.post('/api/change', json={'kind':'draft','value':text,'revision':self.service.revision()}) self.assertEqual(response.status_code, 409) self.assertEqual((self.root / DRAFT).read_text(), '### Source: Bnot\n') @patch('copienator.commands.giving_names.configuration.RETURN_ANSWERS_ENABLED', False) def test_names_and_fallback_scores_survive_both_workflows(self): workspace = EvaluationWorkspace(self.root) self.save('name', 'Alice') self.assertEqual(giving_names.prepare_named_returns(workspace, 'Bnot'), 0) folder = rename_return_copy(self.root / 'A Rendre/Alice (01)', 'Alice corrigée') self.assertEqual(self.service.state()['copies'][0]['name'], 'Alice corrigée') self.assertEqual(giving_names.prepare_named_returns(workspace, 'Bnot'), 0) self.assertEqual(list((self.root / 'A Rendre').iterdir()), [folder]) score = folder / 'score.json' score.unlink() atomic_write_json(score, {'Ex 1':'3','Ex 2':'4','Absent':''}) set_score(score, 'Ex 1', '1') self.assertEqual(read_scores(self.root / 'Bnot/Copie01/score.json')['Ex 1'], '1') self.save('score', '2') self.assertEqual(read_scores(score)['Ex 1'], '2') self.save('name', 'Alice finale') self.assertTrue((self.root / 'A Rendre/Alice finale (01)/Alice finale.jpg').is_file()) def test_worker_order_flags_and_failure(self): calls = [] class Process: stdout = StringIO('done\n') def __init__(self, command, **kwargs): calls.append(command) def __enter__(self): return self def __exit__(self, *args): pass def wait(self): return 0 with patch('copienator.finalisation.subprocess.Popen', Process): self.service._regenerate(True) self.assertEqual([c[3] for c in calls], ['read-annotations','modify-annotations','giving-names']) self.assertTrue(any('return-answers-question' in v for v in calls[0])) self.assertEqual(self.service.job['status'], 'success') calls.clear() with patch.object(Process, 'wait', return_value=4), patch('copienator.finalisation.subprocess.Popen', Process): self.service._regenerate(True) self.assertEqual(len(calls), 1) self.assertEqual(self.service.job['status'], 'failed') def test_os_lock_blocks_cli_and_releases(self): with EvaluationLock(self.root): with self.assertRaises(CliError): with EvaluationLock(self.root): pass result = subprocess.run([sys.executable,'-m','copienator','giving-names',str(self.root),'Bnot'], capture_output=True, text=True) self.assertNotEqual(result.returncode, 0) self.assertIn('utilisée', result.stderr) with EvaluationLock(self.root): pass class SnapshotTests(unittest.TestCase): def test_grouped_layout_without_handwriting_replaces_previous_ink(self): from copienator.review_snapshot import stage, load from copienator.annotation_layout import legacy_layout with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) seed(root, 'BGnot') output = root / 'BGnot/Copie01' data = {label: {'pdf_path': root / f'{label}.pdf', 'result': {'score': 3}} for label in ('Ex 1', 'Ex 2')} ink = Image.new('RGBA', (100, 120), 'red') notes = {label: {'img': ink, 'old_header_h': 20, 'layout': legacy_layout()} for label in data} with staged_files(output) as staging: stage(output, staging, data, notes, engine='grouped') notes['Ex 1']['img'] = None for selected in ({'Ex 1'}, None): with staged_files(output) as staging: stage(output, staging, data, notes, engine='grouped', selected=selected) saved, _, restored = load(root, output) self.assertIsNone(saved['notes']['Ex 1']['file']) self.assertEqual(restored['Ex 1'], notes['Ex 1']) self.assertEqual(restored['Ex 2']['img'].getpixel((0, 0)), (255, 0, 0, 255)) def test_selective_redo_merges_raw_inputs_and_preserves_handwriting(self): from copienator.review_snapshot import stage, load with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) seed(root) output = root / 'Bnot/Copie01' data = {l: {'pdf_path': root / 'Copies/Copie01' / f'{l}.pdf', 'result': {'score': 3, 'feedback': []}, 'coordinates': [0, 0]} for l in ('Ex 1', 'Ex 2')} ink = Image.new('RGBA', (100, 120), (255, 0, 0, 255)) bnote = {'images': [{'label':'Ex 1', 'header_height':20}, {'label':'Ex 2','header_height':10}]} with staged_files(output) as staging: stage(output, staging, data, {'Ex 1': {'img': ink}}, engine='standard', bnote=bnote) data['Ex 2']['result']['score'] = 1 with staged_files(output) as staging: stage(output, staging, data, {}, engine='grouped', selected={'Ex 2'}, force_filtered={'Ex 1'}) saved, restored, notes = load(root, output) self.assertEqual(restored['Ex 1']['result']['score'], 3) self.assertEqual(restored['Ex 2']['result']['score'], 1) self.assertEqual(notes['Ex 1']['img'].getpixel((0,0)), (255,0,0,255)) self.assertEqual(notes['Ex 1']['old_header_h'], 20) self.assertIn('Ex 1', saved['force_filtered']) self.assertEqual(restored['Ex 1']['pdf_path'], root / 'Copies/Copie01/Ex 1.pdf') def test_cached_regeneration_does_not_accumulate_custom_remarks(self): from copienator.review_snapshot import stage, load from copienator.commands import regenerate_reviewed from copienator.commands import annotating from copienator.custom_remarks import set_remark for mode in ('Bnot', 'BGnot'): with self.subTest(mode=mode), tempfile.TemporaryDirectory() as temporary: root = Path(temporary) seed(root, mode) output = root / mode / 'Copie01' pdf = root / 'Copies/Copie01/Ex 1.pdf' pdf.write_bytes(b'placeholder') data = {'Ex 1': {'pdf_path': pdf, 'result': {'score': 3, 'feedback': []}, 'coordinates': [0,0]}} notes = {'Ex 1': {'img': None, 'old_header_h': 20}} if mode == 'BGnot' else {} with staged_files(output) as staging: stage(output, staging, data, notes, engine='standard' if mode == 'Bnot' else 'grouped', bnote={'images':[{'label':'Ex 1','header_height':20}]}) set_remark(root, '01', 'Ex 1', 'Une seule remarque') with patch.object(annotating, 'make_base_image', return_value=(Image.new('RGB',(100,100),'white'),0,0)), \ patch.object(annotating, 'compose_label_image', return_value=(Image.new('RGB',(100,120),'white'),20)) as compose, \ patch('copienator.commands.reading_grouped_annotations.get_extra_pdfs_as_images', return_value=[]): for _ in range(2): self.assertEqual(regenerate_reviewed.run(EvaluationWorkspace(root), mode), 0) feedback = compose.call_args.args[2]['feedback'] self.assertEqual(len(feedback), 1) self.assertEqual(feedback[0]['text'], 'Une seule remarque') _, raw, _ = load(root, output) self.assertEqual(raw['Ex 1']['result']['feedback'], []) @unittest.skipUnless(os.environ.get('DISPLAY'), 'Tk requires a display') class WebGuiTests(unittest.TestCase): def test_modal_lock_and_return_to_existing_workflow(self): from copienator_gui.app import CopienatorApp from copienator.finalisation_server import WebSession import tkinter as tk with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) seed(root) app = CopienatorApp(Path.cwd(), False, root) self.addCleanup(app.destroy) service = Finalisation(root, 'Bnot', 'token') class FakeSession: url = 'http://127.0.0.1:12345/' alive = True def request_stop(self): self.alive = False session = FakeSession() session.service = service def descendants(widget): for child in widget.winfo_children(): yield child yield from descendants(child) def button(text): return next(w for w in descendants(app) if w.winfo_class() == 'TButton' and w.cget('text') == text) with patch('copienator.finalisation_server.WebSession', return_value=session), patch('webbrowser.open'): app._open_finalisation_web() app.update() self.assertIsInstance(app.grab_current(), tk.Toplevel) button('Ouvrir').invoke() self.assertIs(app.web_session, session) self.assertEqual(str(button('Ouvrir').cget('state')), 'disabled') service.save({'kind':'score','copy':'01','label':'Ex 1','value':'1','revision':service.revision()}) with patch.object(app.runner, 'start') as start: app._run_current_step() start.assert_not_called() button('Arrêter le serveur et reprendre').invoke() app.after(450, app.quit) app.mainloop() self.assertIsNone(app.web_session) self.assertIsNone(app.grab_current()) self.assertEqual(app.state_store.step('read_annotations')['status'], 'stale')