Guide d'utilisateur GUI ; Recoverable statement/Persp Gemini generation
This commit is contained in:
sebastien committed
1 parent
2f8422f6d5
commit
94ca8b8256
9 files changed
+1333
-170
No files matched your search
@@ -74,7 +74,8 @@ class SelectiveGeminiTests(unittest.TestCase):
|
||||
(self.root / folder).mkdir()
|
||||
for label in self.labels:
|
||||
suffix = ".tex" if folder.endswith("2") else ""
|
||||
(self.root / folder / (label + suffix)).write_text(f"Personal {folder}: {label}")
|
||||
(self.root / folder / (label + suffix)).write_text(
|
||||
label + '\n' if folder == 'Persp' else f"Personal {folder}: {label}")
|
||||
(self.root / "Persp" / "Ex 1").write_text("Old aggregate rubric")
|
||||
self.client = SimpleNamespace(models=SimpleNamespace(generate_content=Mock()))
|
||||
|
||||
@@ -113,8 +114,11 @@ class SelectiveGeminiTests(unittest.TestCase):
|
||||
self.assertEqual(gemini.refine_existing(self.workspace, "persp", api_client=self.client), ExitCode.SUCCESS)
|
||||
after = self.snapshot()
|
||||
self.assertEqual({key: value for key, value in before.items() if not key.startswith("Persp/")},
|
||||
{key: value for key, value in after.items() if not key.startswith("Persp/")})
|
||||
self.assertFalse((self.root / "Persp" / "Ex 1").exists())
|
||||
{key: value for key, value in after.items()
|
||||
if not key.startswith(("Persp/", "Cache/"))})
|
||||
responses = list((self.root / 'Cache' / 'rubric_responses').glob('*.json'))
|
||||
self.assertEqual(len(responses), 2)
|
||||
self.assertEqual((self.root / "Persp" / "Ex 1").read_bytes(), before['Persp/Ex 1'])
|
||||
for label in self.labels:
|
||||
self.assertIn("Barème Gemini", (self.root / "Persp" / label).read_text())
|
||||
for call in self.client.models.generate_content.call_args_list:
|
||||
@@ -126,13 +130,22 @@ class SelectiveGeminiTests(unittest.TestCase):
|
||||
self.assertIn("caractères mathématiques Unicode", gemini.PROMPT_4)
|
||||
self.assertIn(r"$\lfloor \sqrt{k} \rfloor$", gemini.PROMPT_4)
|
||||
|
||||
def test_incomplete_or_failed_rubrics_preserve_entire_persp(self):
|
||||
def test_incomplete_or_failed_rubrics_keep_completed_groups(self):
|
||||
before = self.snapshot()
|
||||
for last in (self.response({"rubrics": []}), RuntimeError("API unavailable")):
|
||||
for label in self.labels[:2]:
|
||||
(self.root / 'Persp' / label).write_bytes(before[f'Persp/{label}'])
|
||||
self.client.models.generate_content.side_effect = [self.rubric_response(self.labels[:2]), last]
|
||||
with self.assertRaises((CliError, RuntimeError)):
|
||||
gemini.refine_existing(self.workspace, "persp", api_client=self.client)
|
||||
self.assertEqual(self.snapshot(), before)
|
||||
after = self.snapshot()
|
||||
for label in self.labels[:2]:
|
||||
self.assertIn(b'Bar\xc3\xa8me Gemini', after[f'Persp/{label}'])
|
||||
self.assertEqual(after['Persp/Ex 2'], before['Persp/Ex 2'])
|
||||
self.assertEqual(after['Persp/Ex 1'], before['Persp/Ex 1'])
|
||||
self.assertEqual({key: value for key, value in before.items() if not key.startswith('Persp/')},
|
||||
{key: value for key, value in after.items()
|
||||
if not key.startswith(('Persp/', 'Cache/'))})
|
||||
|
||||
def test_rubric_latex_and_line_breaks_survive_json_and_file_writes(self):
|
||||
expected = (
|
||||
@@ -161,15 +174,96 @@ class SelectiveGeminiTests(unittest.TestCase):
|
||||
{'Ex 2': '$' + command + '$'})
|
||||
self.assertEqual(self.client.models.generate_content.call_count, 1)
|
||||
|
||||
def test_unrepairable_control_character_preserves_all_existing_rubrics(self):
|
||||
def test_control_character_keeps_progress_and_saves_diagnostics(self):
|
||||
before = self.snapshot()
|
||||
broken = self.response({'rubrics': [{'label': 'Ex 2', 'rubric_content': 'invalid \x01 control'}]})
|
||||
self.client.models.generate_content.side_effect = [
|
||||
self.rubric_response(self.labels[:2]), broken]
|
||||
with self.assertRaisesRegex(ValueError, 'control characters'):
|
||||
output = io.StringIO()
|
||||
with patch('sys.stdout', output), self.assertRaisesRegex(ValueError, 'control characters'):
|
||||
gemini.refine_existing(self.workspace, 'persp', api_client=self.client)
|
||||
after = self.snapshot()
|
||||
self.assertEqual(after['Persp/Ex 2'], before['Persp/Ex 2'])
|
||||
for label in self.labels[:2]:
|
||||
self.assertIn(b'Bar\xc3\xa8me Gemini', after[f'Persp/{label}'])
|
||||
self.assertIn('Question Ex 2', output.getvalue())
|
||||
self.assertIn('U+0001 at offset 8', output.getvalue())
|
||||
self.assertIn(r'invalid \x01 control', output.getvalue())
|
||||
responses = list((self.root / 'Cache' / 'rubric_responses').glob('*.json'))
|
||||
self.assertEqual(len(responses), 2)
|
||||
self.assertIn(broken.text, [path.read_text() for path in responses])
|
||||
log = (self.root / 'Cache' / 'rubric_errors.log').read_text()
|
||||
self.assertIn('Question Ex 2', log)
|
||||
self.assertIn(r'invalid \x01 control', log)
|
||||
|
||||
def test_invalid_json_is_saved_exactly_and_reported(self):
|
||||
raw = '{"rubrics": [this is malformed JSON'
|
||||
self.client.models.generate_content.return_value = SimpleNamespace(text=raw)
|
||||
output = io.StringIO()
|
||||
with patch('sys.stdout', output), self.assertRaises(ValueError):
|
||||
gemini.generate_rubrics(self.client, 'context', workspace=self.workspace,
|
||||
expected_labels=['Ex 2'])
|
||||
responses = list((self.root / 'Cache' / 'rubric_responses').glob('*.json'))
|
||||
self.assertEqual(len(responses), 1)
|
||||
self.assertEqual(responses[0].read_text(), raw)
|
||||
self.assertIn(raw, output.getvalue())
|
||||
self.assertIn(str(responses[0]), output.getvalue())
|
||||
|
||||
def test_missing_question_only_preserves_existing_and_keeps_group_context(self):
|
||||
kept = self.root / 'Persp' / self.labels[0]
|
||||
kept.write_text(self.labels[0] + '\nManually reviewed rubric')
|
||||
before = kept.read_bytes()
|
||||
(self.root / 'Persp' / self.labels[1]).write_text('')
|
||||
(self.root / 'Persp' / self.labels[2]).unlink()
|
||||
self.client.models.generate_content.side_effect = [
|
||||
self.rubric_response([self.labels[1]]), self.rubric_response([self.labels[2]])]
|
||||
gemini.refine_existing(self.workspace, 'persp', api_client=self.client)
|
||||
self.assertEqual(kept.read_bytes(), before)
|
||||
call = self.client.models.generate_content.call_args_list[0]
|
||||
text = call.kwargs['contents'][0].parts[1].text
|
||||
self.assertIn('UNIQUEMENT pour les labels suivants : ' + self.labels[1] + '\n', text)
|
||||
self.assertIn('Question [' + self.labels[0] + ']', text)
|
||||
self.assertEqual(self.client.models.generate_content.call_count, 2)
|
||||
|
||||
def test_retry_requests_only_failed_group(self):
|
||||
self.client.models.generate_content.side_effect = [
|
||||
self.rubric_response(self.labels[:2]), RuntimeError('API unavailable')]
|
||||
with self.assertRaises(RuntimeError):
|
||||
gemini.refine_existing(self.workspace, 'persp', api_client=self.client)
|
||||
before = {label: (self.root / 'Persp' / label).read_bytes() for label in self.labels[:2]}
|
||||
self.client.models.generate_content.reset_mock()
|
||||
self.client.models.generate_content.side_effect = [self.rubric_response(self.labels[2:])]
|
||||
gemini.refine_existing(self.workspace, 'persp', api_client=self.client)
|
||||
self.assertEqual(self.client.models.generate_content.call_count, 1)
|
||||
for label, content in before.items():
|
||||
self.assertEqual((self.root / 'Persp' / label).read_bytes(), content)
|
||||
|
||||
def test_complete_persp_needs_no_api_key_or_requests(self):
|
||||
for label in self.labels:
|
||||
(self.root / 'Persp' / label).write_text(label + '\nReviewed rubric')
|
||||
before = self.snapshot()
|
||||
with patch.object(gemini, 'api_key', ''), patch.object(gemini.genai, 'Client') as client:
|
||||
self.assertEqual(gemini.refine_existing(self.workspace, 'persp'), ExitCode.SUCCESS)
|
||||
client.assert_not_called()
|
||||
self.assertEqual(self.snapshot(), before)
|
||||
|
||||
def test_delete_persp_forces_regeneration_and_removes_old_files(self):
|
||||
for label in self.labels:
|
||||
(self.root / 'Persp' / label).write_text(label + '\nReviewed rubric')
|
||||
self.client.models.generate_content.side_effect = [
|
||||
self.rubric_response(self.labels[:2]), self.rubric_response(self.labels[2:])]
|
||||
gemini.refine_existing(self.workspace, 'persp', api_client=self.client, delete_persp=True)
|
||||
self.assertEqual(self.client.models.generate_content.call_count, 2)
|
||||
self.assertEqual({path.name for path in (self.root / 'Persp').iterdir()}, set(self.labels))
|
||||
|
||||
def test_cli_delete_persp_dispatch(self):
|
||||
with patch.object(gemini, 'refine_existing', return_value=ExitCode.SUCCESS) as refine:
|
||||
self.assertEqual(gemini.main([str(self.root), '--persp-only', '--delete-persp']), ExitCode.SUCCESS)
|
||||
self.assertEqual(refine.call_args.kwargs, {'delete_persp': True})
|
||||
refine.reset_mock()
|
||||
self.assertEqual(gemini.main([str(self.root), '--delete-persp']), ExitCode.INVALID_ARGUMENTS)
|
||||
refine.assert_not_called()
|
||||
|
||||
def test_cli_dispatches_selective_modes_without_full_extraction(self):
|
||||
with patch.object(gemini, "refine_existing", return_value=ExitCode.SUCCESS) as refine, patch.object(
|
||||
gemini, "process_exam"
|
||||
@@ -251,6 +345,28 @@ class PersonalStatementGuiTests(unittest.TestCase):
|
||||
self.assertIn(flag, app.command_var.get())
|
||||
self.assertTrue(app.current_step.optional)
|
||||
self.assertFalse(app.runner.running)
|
||||
(evaluation / 'labels').write_text('Ex 1\nEx 2\n')
|
||||
(evaluation / 'Persp').mkdir()
|
||||
(evaluation / 'Persp' / 'Ex 1').write_text('Ex 1\nReviewed rubric')
|
||||
(evaluation / 'Persp' / 'Ex 2').write_text('Ex 2\n')
|
||||
app._refresh_persp_inventory()
|
||||
self.assertIn('Barèmes présents : 1/2', app.persp_inventory_label.cget('text'))
|
||||
self.assertIn('Manquants : Ex 2', app.persp_inventory_label.cget('text'))
|
||||
self.assertFalse(app.arg_vars['delete_persp'].get())
|
||||
app.arg_vars['delete_persp'].set(True)
|
||||
self.assertIn('--delete-persp', app.command_var.get())
|
||||
app._save_current_form()
|
||||
app._render_step()
|
||||
self.assertFalse(app.arg_vars['delete_persp'].get())
|
||||
self.assertNotIn('--delete-persp', app.command_var.get())
|
||||
app.tree.selection_set('statement')
|
||||
app.update()
|
||||
app._select_variant(1)
|
||||
app.arg_vars['restart'].set(True)
|
||||
app._save_current_form()
|
||||
app._render_step()
|
||||
self.assertFalse(app.arg_vars['restart'].get())
|
||||
self.assertNotIn('--restart', app.command_var.get())
|
||||
finally:
|
||||
for callback in app.tk.splitlist(app.tk.call("after", "info")):
|
||||
app.after_cancel(callback)
|
||||
|
||||
@@ -0,0 +1,259 @@
|
||||
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')
|
||||
Reference in new issue
Block a user