Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
3.3 kB · 96 lines
Python
at letta-code
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697from __future__ import annotations
import subprocessimport sysimport tempfileimport unittestfrom pathlib import Pathfrom unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]sys.path.insert(0, str(ROOT / "src"))
import pi_docker_sandbox # noqa: E402from pi_docker_sandbox import ( # noqa: E402 DEFAULT_IMAGE, DockerSandboxConfig, docker_sandbox_config, start_docker_sandbox,)from transfer_tasks import ( # noqa: E402 LANGUAGES, build_transfer_tasks, materialize_task_resources, transfer_task_digest,)
EXPECTED_DIGEST = "5857a0999a1d4b14636b3e7a9ad2afb587174eefef370c193dd0960d46e12de1"
class TransferTasksTest(unittest.TestCase): def test_frozen_task_digest_and_answers(self) -> None: tasks = build_transfer_tasks() self.assertEqual(transfer_task_digest(tasks), EXPECTED_DIGEST) self.assertEqual( {task.task_id: task.answer for task in tasks}, { "event-log-retries": "101", "inventory-value": "18990", "repair-weighted-checksum": "107", "sharded-ranges": "65", }, )
def test_materializes_language_specific_repair_files(self) -> None: task = next( task for task in build_transfer_tasks() if task.task_id == "repair-weighted-checksum" ) with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) refs = { language: materialize_task_resources(task, language, root) for language in LANGUAGES } self.assertIn("broken.py", refs["python"]) self.assertIn("broken.js", refs["javascript"]) self.assertIn("broken.jl", refs["julia"]) self.assertTrue((root / "numbers.txt").is_file())
def test_docker_sandbox_defaults_to_pinned_image(self) -> None: config = docker_sandbox_config({}) self.assertIsNotNone(config) assert config is not None self.assertEqual(config.image, DEFAULT_IMAGE) self.assertIsNone(docker_sandbox_config({"enabled": False}))
def test_start_timeout_attempts_container_cleanup(self) -> None: with tempfile.TemporaryDirectory() as temporary: timeout = subprocess.TimeoutExpired(["docker", "run"], 60) removed = subprocess.CompletedProcess( ["docker", "rm", "--force", "machine-euler-timeout"], 0, stdout="machine-euler-timeout\n", stderr="", ) with ( patch.object( pi_docker_sandbox, "_run", side_effect=[timeout, removed] ) as run, self.assertRaises(subprocess.TimeoutExpired), ): start_docker_sandbox( Path(temporary), "machine-euler-timeout", DockerSandboxConfig(), ) self.assertEqual(run.call_count, 2) self.assertEqual( run.call_args_list[1].args[0], ["docker", "rm", "--force", "machine-euler-timeout"], )
if __name__ == "__main__": unittest.main()