71 lines
3.4 KiB
Python
71 lines
3.4 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
from copienator import CliError, EvaluationWorkspace, ExitCode, atomic_write_json, read_json
|
|
from copienator.commands.submit_batches import run
|
|
|
|
|
|
class BatchSubmissionResumeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.workspace = EvaluationWorkspace(Path(directory.name))
|
|
for tier in ("flash", "pro"):
|
|
(self.workspace.root / f"batch_requests_{tier}.jsonl").write_text("{}\n")
|
|
self.client = Mock()
|
|
self.client.files.upload.return_value = SimpleNamespace(name="files/input")
|
|
self.client.batches.create.return_value = SimpleNamespace(name="batches/flash")
|
|
|
|
def test_restart_after_second_upload_failure_preserves_first_job(self):
|
|
self.client.files.upload.side_effect = [
|
|
SimpleNamespace(name="files/input"), ConnectionResetError("reset"),
|
|
]
|
|
with self.assertRaises(ConnectionResetError):
|
|
run(self.workspace, client=self.client)
|
|
original = read_json(self.workspace.batch_jobs_file)["jobs"]["flash"]
|
|
self.assertEqual(original["name"], "batches/flash")
|
|
|
|
self.client.reset_mock()
|
|
self.client.files.upload.side_effect = None
|
|
self.client.batches.create.return_value = SimpleNamespace(name="batches/pro")
|
|
self.assertEqual(run(self.workspace, client=self.client), ExitCode.SUCCESS)
|
|
self.client.files.upload.assert_called_once()
|
|
self.assertEqual(self.client.files.upload.call_args.kwargs["file"],
|
|
str(self.workspace.root / "batch_requests_pro.jsonl"))
|
|
self.client.batches.create.assert_called_once()
|
|
jobs = read_json(self.workspace.batch_jobs_file)["jobs"]
|
|
self.assertEqual(jobs["flash"], original)
|
|
self.assertEqual(jobs["pro"]["name"], "batches/pro")
|
|
|
|
self.client.reset_mock()
|
|
self.assertEqual(run(self.workspace, client=self.client), ExitCode.SUCCESS)
|
|
self.client.files.upload.assert_not_called()
|
|
self.client.batches.create.assert_not_called()
|
|
|
|
def test_legacy_manifest_is_skipped_and_explicit_overwrite_resubmits(self):
|
|
atomic_write_json(self.workspace.batch_jobs_file, {"jobs": {
|
|
"flash": {"name": "batches/old-flash"},
|
|
"pro": {"name": "batches/old-pro"},
|
|
}})
|
|
self.assertEqual(run(self.workspace, client=self.client), ExitCode.SUCCESS)
|
|
self.client.files.upload.assert_not_called()
|
|
self.assertEqual(run(self.workspace, client=self.client, overwrite=True), ExitCode.SUCCESS)
|
|
self.assertEqual(self.client.batches.create.call_count, 2)
|
|
|
|
def test_invalid_manifest_does_not_trigger_duplicate_submission(self):
|
|
for manifest in ({"jobs": []}, {"jobs": {"flash": {}}}):
|
|
with self.subTest(manifest=manifest):
|
|
atomic_write_json(self.workspace.batch_jobs_file, manifest)
|
|
with self.assertRaises(CliError):
|
|
run(self.workspace, client=self.client)
|
|
self.client.files.upload.assert_not_called()
|
|
|
|
def test_empty_inputs_report_partial(self):
|
|
for tier in ("flash", "pro"):
|
|
(self.workspace.root / f"batch_requests_{tier}.jsonl").write_text("")
|
|
self.assertEqual(run(self.workspace, client=self.client), ExitCode.PARTIAL)
|
|
self.client.files.upload.assert_not_called()
|