Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
2.7 kB · 81 lines
Python
at letta-code
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182from __future__ import annotations
import hashlibimport jsonimport sysimport tempfileimport unittestfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[1]sys.path.insert(0, str(ROOT))
from compare_transfer import comparable_manifest, verified_run # noqa: E402
class CompareTransferTest(unittest.TestCase): def write_run(self, root: Path) -> None: results = root / "results.jsonl" summary = root / "summary.json" row = { "language": "python", "strategy": "event-log-retries", "functional_qualified": True, } results.write_text(json.dumps(row) + "\n") summary.write_text( json.dumps( { "status": "completed", "task_digest": "frozen", "model_revision": "source", } ) + "\n" ) manifest = { "status": "completed", "model": {"provider": "test", "model": "checkpoint", "revision": "source"}, "fanout": 1, "languages": ["python"], "selected_tasks": ["event-log-retries"], "task_digest": "frozen", "expected_task_digest": "frozen", "expected_rows": 1, "sandbox": {"image_id": "sha256:test"}, "tasks": [{"problem_id": 9001, "answer": "42"}], "source_sha256": {"score.py": "abc"}, "result_receipt": { "rows": 1, "sha256": hashlib.sha256(results.read_bytes()).hexdigest(), }, "summary_receipt": { "sha256": hashlib.sha256(summary.read_bytes()).hexdigest() }, } (root / "manifest.json").write_text(json.dumps(manifest) + "\n")
def test_verifies_result_and_summary_receipts(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) self.write_run(root) manifest, summary, rows = verified_run(root) self.assertEqual(summary["task_digest"], "frozen") self.assertEqual(len(rows), 1) self.assertEqual( comparable_manifest(manifest)["model_except_revision"]["provider"], "test", )
def test_rejects_tampered_results(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) self.write_run(root) with (root / "results.jsonl").open("a") as handle: handle.write("{}\n") with self.assertRaises(RuntimeError): verified_run(root)
if __name__ == "__main__": unittest.main()