260 lines
14 KiB
Python
260 lines
14 KiB
Python
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
from copienator import CliError, EvaluationWorkspace, ExitCode
|
|
from copienator.commands import gemini_for_enonce as gemini
|
|
from copienator.statement_checkpoints import extract_stage
|
|
|
|
|
|
def response(**data):
|
|
return SimpleNamespace(text=json.dumps(data))
|
|
|
|
|
|
class ExtractionCheckpointTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
|
|
def solutions(self, request):
|
|
return extract_stage(self.root, 'solutions', gemini.SolutionOnlyItem,
|
|
'solution_content', request, labels=['1)', '2)'])
|
|
|
|
def test_valid_sibling_survives_and_retry_only_requests_missing(self):
|
|
request = Mock(return_value=response(solutions=[
|
|
{'label': '1)', 'solution_content': 'good'},
|
|
{'label': '2)', 'solution_content': 'bad\x01'}]))
|
|
output = io.StringIO()
|
|
with patch('sys.stdout', output), self.assertRaises(CliError):
|
|
self.solutions(request)
|
|
saved = json.loads((self.root / 'Cache/statement_solutions.json').read_text())
|
|
self.assertEqual(saved['solutions'], [{'label': '1)', 'solution_content': 'good'}])
|
|
self.assertIn(r'bad\x01', output.getvalue())
|
|
self.assertIn('U+0001', (self.root / 'Cache/statement_errors.log').read_text())
|
|
request = Mock(return_value=response(solutions=[{'label': '2)', 'solution_content': 'fixed'}]))
|
|
items = self.solutions(request)
|
|
request.assert_called_once_with(['2)'])
|
|
self.assertEqual([item.solution_content for item in items], ['good', 'fixed'])
|
|
request.reset_mock()
|
|
self.solutions(request)
|
|
request.assert_not_called()
|
|
|
|
def test_malformed_json_and_api_failure_preserve_accepted_items(self):
|
|
request = Mock(return_value=response(solutions=[{'label': '1)', 'solution_content': 'good'}]))
|
|
with self.assertRaises(CliError):
|
|
self.solutions(request)
|
|
before = (self.root / 'Cache/statement_solutions.json').read_bytes()
|
|
raw = '{"solutions": [broken'
|
|
request.return_value = SimpleNamespace(text=raw)
|
|
with self.assertRaises(CliError):
|
|
self.solutions(request)
|
|
self.assertEqual((self.root / 'Cache/statement_solutions.json').read_bytes(), before)
|
|
self.assertIn(raw, [p.read_text() for p in (self.root / 'Cache/statement_responses').iterdir()])
|
|
request.side_effect = RuntimeError('offline')
|
|
with self.assertRaises(CliError):
|
|
self.solutions(request)
|
|
self.assertEqual((self.root / 'Cache/statement_solutions.json').read_bytes(), before)
|
|
|
|
def test_bad_legacy_cache_is_archived_then_not_replayed(self):
|
|
(self.root / 'Cache').mkdir()
|
|
legacy = self.root / 'Cache/gemini_solutions.json'
|
|
legacy.write_text('broken JSON')
|
|
request = Mock(return_value=response(solutions=[
|
|
{'label': label, 'solution_content': 'fixed'} for label in ['1)', '2)']]))
|
|
with self.assertRaises(CliError):
|
|
self.solutions(request)
|
|
request.assert_not_called()
|
|
self.solutions(request)
|
|
request.assert_called_once_with(['1)', '2)'])
|
|
self.assertEqual(legacy.read_text(), 'broken JSON')
|
|
|
|
def test_questions_retry_full_inventory_keep_valid_text_and_order(self):
|
|
request = Mock(return_value=response(questions=[
|
|
{'label': '1)', 'question_content': 'good'},
|
|
{'label': '2)', 'question_content': 'bad\x01'}]))
|
|
def run():
|
|
return extract_stage(self.root, 'questions', gemini.QuestionOnlyItem, 'question_content', request)
|
|
with self.assertRaises(CliError):
|
|
run()
|
|
request.return_value = response(questions=[{'label': '2)', 'question_content': 'fixed'}])
|
|
with self.assertRaises(CliError):
|
|
run() # Cannot silently lose the known first question.
|
|
request.return_value = response(questions=[{'label': '1)', 'question_content': 'good'}])
|
|
with self.assertRaises(CliError):
|
|
run() # Cannot silently omit the previously rejected question either.
|
|
request.return_value = response(questions=[
|
|
{'label': '1)', 'question_content': 'rewritten'},
|
|
{'label': '2)', 'question_content': 'fixed'}])
|
|
items = run()
|
|
self.assertEqual([item.question_content for item in items], ['good', 'fixed'])
|
|
self.assertEqual(request.call_args.args, (None,))
|
|
|
|
def test_contexts_retry_bad_target_and_allow_no_context(self):
|
|
request = Mock(return_value=response(contexts=[
|
|
{'target_question_label': '1)', 'last_question_label': '2)', 'context_content': 'good'},
|
|
{'target_question_label': '2)', 'last_question_label': '2)', 'context_content': 'bad\x01'}]))
|
|
def run():
|
|
return extract_stage(self.root, 'contexts', gemini.ExtractedContext, 'context_content',
|
|
request, labels=['1)', '2)'])
|
|
with self.assertRaises(CliError):
|
|
run()
|
|
request.return_value = response(contexts=[])
|
|
with self.assertRaises(CliError):
|
|
run()
|
|
request.return_value = response(contexts=[
|
|
{'target_question_label': '2)', 'last_question_label': '2)', 'context_content': 'fixed'}])
|
|
self.assertEqual(len(run()), 2)
|
|
self.assertEqual(request.call_args.args, (['2)'],))
|
|
with tempfile.TemporaryDirectory() as root:
|
|
self.assertEqual(extract_stage(Path(root), 'contexts', gemini.ExtractedContext,
|
|
'context_content', Mock(return_value=response(contexts=[])),
|
|
labels=['1)']), [])
|
|
|
|
def test_duplicate_and_unknown_solutions_are_not_silently_accepted(self):
|
|
request = Mock(return_value=response(solutions=[
|
|
{'label': '1)', 'solution_content': 'good'},
|
|
{'label': '1)', 'solution_content': 'duplicate'},
|
|
{'label': 'unknown', 'solution_content': 'bad'}]))
|
|
with self.assertRaises(CliError):
|
|
self.solutions(request)
|
|
saved = json.loads((self.root / 'Cache/statement_solutions.json').read_text())
|
|
self.assertFalse(saved['complete'])
|
|
self.assertEqual(saved['solutions'], [{'label': '1)', 'solution_content': 'good'}])
|
|
|
|
|
|
class FullStatementResumeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.workspace = EvaluationWorkspace(self.root)
|
|
(self.root / 'enonce.pdf').write_bytes(b'pdf')
|
|
(self.root / 'enonce.tex').write_text('source')
|
|
(self.root / 'correction.tex').write_text('correction')
|
|
self.client = SimpleNamespace(models=SimpleNamespace(generate_content=Mock()))
|
|
self.editor = Mock()
|
|
self.compiler = Mock(side_effect=lambda content, path, **kwargs: Path(path).write_bytes(b'PDF'))
|
|
|
|
def run_exam(self, **kwargs):
|
|
with patch.object(gemini.utils, 'edit_file_and_enter', self.editor), patch.object(
|
|
gemini, 'compile_to_pdf', self.compiler):
|
|
return gemini.process_exam(self.workspace, api_client=self.client, **kwargs)
|
|
|
|
def responses(self):
|
|
return [response(questions=[{'label': 'Ex 1', 'question_content': 'question'}]),
|
|
response(solutions=[{'label': 'Ex 1', 'solution_content': 'answer'}]),
|
|
response(contexts=[]),
|
|
response(rubrics=[{'label': 'Ex 1', 'rubric_content': '4 points : correct'}])]
|
|
|
|
def test_rubric_failure_retry_keeps_grouping_edits_and_sources(self):
|
|
self.client.models.generate_content.side_effect = [*self.responses()[:3], RuntimeError('offline')]
|
|
self.assertEqual(self.run_exam(), ExitCode.PARTIAL)
|
|
self.assertTrue((self.root / 'Cache/statement_grouping.json').is_file())
|
|
self.assertEqual((self.root / 'labels').read_text(), 'Ex 1\n')
|
|
tex = self.root / 'Text2/Ex 1.tex'
|
|
tex.write_text('manually edited source')
|
|
edited_items = (self.root / 'Tmp/exam_items.txt').read_bytes()
|
|
self.client.models.generate_content.reset_mock()
|
|
self.client.models.generate_content.side_effect = [self.responses()[-1]]
|
|
self.editor.reset_mock()
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
self.editor.assert_not_called()
|
|
self.assertEqual(self.client.models.generate_content.call_count, 1)
|
|
self.assertEqual(tex.read_text(), 'manually edited source')
|
|
self.assertEqual((self.root / 'Tmp/exam_items.txt').read_bytes(), edited_items)
|
|
|
|
def test_failed_pdf_only_is_compiled_on_retry(self):
|
|
self.client.models.generate_content.side_effect = self.responses()
|
|
def compile_pdf(content, path, **kwargs):
|
|
if Path(path).parent.name == 'Sol2':
|
|
Path(path).write_bytes(b'partial PDF from failed compilation')
|
|
raise RuntimeError('compiler failure')
|
|
Path(path).write_bytes(b'PDF')
|
|
self.compiler.side_effect = compile_pdf
|
|
self.assertEqual(self.run_exam(), ExitCode.PARTIAL)
|
|
good_pdf = (self.root / 'Text2/Ex 1.pdf').read_bytes()
|
|
self.compiler.reset_mock()
|
|
self.compiler.side_effect = lambda content, path, **kwargs: Path(path).write_bytes(b'PDF')
|
|
self.client.models.generate_content.reset_mock()
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
self.client.models.generate_content.assert_not_called()
|
|
self.assertEqual(self.compiler.call_count, 1)
|
|
self.assertEqual(self.compiler.call_args.args[1], self.root / 'Sol2/Ex 1.pdf')
|
|
self.assertEqual((self.root / 'Text2/Ex 1.pdf').read_bytes(), good_pdf)
|
|
|
|
def test_interrupted_editor_keeps_edits_until_approval(self):
|
|
self.client.models.generate_content.side_effect = self.responses()[:3]
|
|
def interrupt(path):
|
|
Path(path).write_text('Custom label ### manually edited question\n')
|
|
raise KeyboardInterrupt()
|
|
self.editor.side_effect = interrupt
|
|
with self.assertRaises(KeyboardInterrupt):
|
|
self.run_exam()
|
|
self.editor.side_effect = None
|
|
self.client.models.generate_content.side_effect = [response(rubrics=[
|
|
{'label': 'Custom label', 'rubric_content': '4 points : correct'}])]
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
self.assertIn('manually edited question', (self.root / 'Text2/Custom label.tex').read_text())
|
|
|
|
def test_restart_archives_prior_work_and_reextracts(self):
|
|
self.client.models.generate_content.side_effect = self.responses()
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
old = (self.root / 'Persp/Ex 1').read_bytes()
|
|
self.client.models.generate_content.side_effect = self.responses()
|
|
self.editor.reset_mock()
|
|
self.assertEqual(self.run_exam(restart=True), ExitCode.SUCCESS)
|
|
self.editor.assert_called_once()
|
|
archived = list((self.root / 'Cache/statement_history').glob('*/Persp/Ex 1'))
|
|
self.assertEqual(len(archived), 1)
|
|
self.assertEqual(archived[0].read_bytes(), old)
|
|
|
|
def test_failed_restart_then_resume_still_replaces_old_outputs(self):
|
|
self.client.models.generate_content.side_effect = self.responses()
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
self.client.models.generate_content.side_effect = [SimpleNamespace(text='broken JSON')]
|
|
with self.assertRaises(CliError):
|
|
self.run_exam(restart=True)
|
|
self.assertTrue((self.root / 'Persp/Ex 1').exists())
|
|
self.client.models.generate_content.side_effect = self.responses()
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
self.assertTrue(list((self.root / 'Cache/statement_history').glob('*/Persp/Ex 1')))
|
|
|
|
def test_rubric_valid_sibling_saved_even_on_invalid_content(self):
|
|
self.client.models.generate_content.return_value = response(rubrics=[
|
|
{'label': 'Ex 1', 'rubric_content': 'good'},
|
|
{'label': 'Ex 2', 'rubric_content': 'bad\x01'}])
|
|
with self.assertRaises(ValueError):
|
|
gemini.generate_rubrics(self.client, 'context', workspace=self.workspace,
|
|
expected_labels=['Ex 1', 'Ex 2'])
|
|
self.assertEqual((self.root / 'Persp/Ex 1').read_text(), 'Ex 1\ngood')
|
|
self.assertFalse((self.root / 'Persp/Ex 2').exists())
|
|
|
|
def test_rubric_schema_error_also_preserves_valid_sibling(self):
|
|
self.client.models.generate_content.return_value = response(rubrics=[
|
|
{'label': 'Ex 1', 'rubric_content': 'good'}, {'label': 'Ex 2'}])
|
|
with self.assertRaises(ValueError):
|
|
gemini.generate_rubrics(self.client, 'context', workspace=self.workspace,
|
|
expected_labels=['Ex 1', 'Ex 2'])
|
|
self.assertEqual((self.root / 'Persp/Ex 1').read_text(), 'Ex 1\ngood')
|
|
|
|
def test_extraction_failure_leaves_existing_outputs_untouched(self):
|
|
(self.root / 'Text2').mkdir()
|
|
existing = self.root / 'Text2/reviewed.tex'
|
|
existing.write_text('manually reviewed question')
|
|
self.client.models.generate_content.side_effect = [self.responses()[0], response(solutions=[
|
|
{'label': 'Ex 1', 'solution_content': 'bad\x01'}])]
|
|
with self.assertRaises(CliError):
|
|
self.run_exam()
|
|
self.assertEqual(existing.read_text(), 'manually reviewed question')
|
|
self.assertTrue(json.loads((self.root / 'Cache/statement_questions.json').read_text())['complete'])
|
|
self.client.models.generate_content.reset_mock()
|
|
self.client.models.generate_content.side_effect = self.responses()[1:]
|
|
self.assertEqual(self.run_exam(), ExitCode.SUCCESS)
|
|
self.assertEqual(self.client.models.generate_content.call_count, 3)
|
|
self.assertEqual(existing.read_text(), 'manually reviewed question')
|