| """Tests for the simulation-mode executor.""" |
| from forgeenv.sandbox.simulation_mode import SimulationExecutor |
| from forgeenv.tasks.models import Task |
|
|
| VALID_HF = """ |
| from transformers import Trainer, TrainingArguments |
| from datasets import load_dataset |
| import torch |
| |
| dataset = load_dataset("glue", "sst2") |
| trainer = Trainer(model=None, args=None, train_dataset=dataset) |
| trainer.train() |
| trainer.save_model("/tmp/forge_output/checkpoint") |
| print("TRAINING_COMPLETE") |
| """ |
|
|
| SYNTAX_ERROR = "def foo(\n broken" |
|
|
| OS_IMPORT = "import os\nos.listdir('.')" |
|
|
|
|
| def _task(content: str) -> Task: |
| return Task( |
| task_id="t", |
| description="d", |
| script_content=content, |
| difficulty="easy", |
| ) |
|
|
|
|
| def test_valid_script_can_succeed(): |
| """With seed 0, the valid HF script eventually returns a positive case.""" |
| executor = SimulationExecutor(seed=0) |
| result = executor.execute(VALID_HF, _task(VALID_HF)) |
| |
| |
| assert result.exit_code in (0, 1) |
| if result.exit_code == 0: |
| assert "TRAINING_COMPLETE" in result.stdout |
|
|
|
|
| def test_syntax_error_fails(): |
| executor = SimulationExecutor(seed=0) |
| result = executor.execute(SYNTAX_ERROR, _task(SYNTAX_ERROR)) |
| assert result.exit_code == 1 |
| assert "SyntaxError" in result.stderr |
|
|
|
|
| def test_forbidden_import_fails(): |
| executor = SimulationExecutor(seed=0) |
| result = executor.execute(OS_IMPORT, _task(OS_IMPORT)) |
| assert result.exit_code == 1 |
| assert "Validation failed" in result.stderr |
|
|
|
|
| def test_simulation_is_fast(): |
| """Simulation mode must complete each call in <100ms wall_time. |
| |
| The reported wall_time_ms field includes a synthetic delay so we measure |
| real elapsed time at this layer instead. |
| """ |
| import time |
| executor = SimulationExecutor(seed=0) |
| t0 = time.time() |
| executor.execute(VALID_HF, _task(VALID_HF)) |
| elapsed_ms = (time.time() - t0) * 1000 |
| assert elapsed_ms < 200, f"Simulation took {elapsed_ms:.1f}ms" |
|
|
|
|
| def test_seed_is_deterministic(): |
| e1 = SimulationExecutor(seed=42) |
| e2 = SimulationExecutor(seed=42) |
| r1 = e1.execute(VALID_HF, _task(VALID_HF)) |
| r2 = e2.execute(VALID_HF, _task(VALID_HF)) |
| assert r1.exit_code == r2.exit_code |
| assert r1.stderr == r2.stderr |
|
|
