| """Smoke tests for the training pipeline (rollout, dry-run trainers)."""
|
| import json
|
| import tempfile
|
| from pathlib import Path
|
|
|
| from forgeenv.env.forge_environment import ForgeEnvironment
|
| from forgeenv.training.grpo_repair import run_grpo
|
| from forgeenv.training.grpo_drift import run_drift_grpo_dry_run
|
| from forgeenv.training.rollout import (
|
| baseline_oracle_repair_generate,
|
| rollout_one_episode,
|
| )
|
|
|
|
|
| def test_rollout_one_episode_baseline_no_op_repair():
|
| env = ForgeEnvironment(seed=1)
|
| result = rollout_one_episode(env)
|
| assert result.task_id
|
| assert result.primitive_type
|
| assert isinstance(result.visible_reward, float)
|
| assert "executed_cleanly" in result.held_out_breakdown
|
|
|
|
|
| def test_rollout_one_episode_with_oracle_repair_succeeds():
|
| env = ForgeEnvironment(seed=2)
|
| repair_gen = baseline_oracle_repair_generate(env)
|
| result = rollout_one_episode(env, repair_generate=repair_gen)
|
|
|
| assert result.held_out_breakdown.get("intent_preserved", 0.0) > 0.7
|
|
|
|
|
| def test_grpo_repair_dry_run_smoke():
|
| with tempfile.TemporaryDirectory() as tmp:
|
| run_grpo(
|
| base_model="(unused-in-dry-run)",
|
| adapter_path=None,
|
| output_dir=tmp,
|
| total_episodes=5,
|
| group_size=2,
|
| seed=0,
|
| use_unsloth=False,
|
| )
|
| rewards_path = Path(tmp) / "dry_run_rewards.json"
|
| assert rewards_path.exists()
|
| rewards = json.loads(rewards_path.read_text())
|
| assert len(rewards) == 5
|
| assert all(isinstance(r, (int, float)) for r in rewards)
|
|
|
|
|
| def test_grpo_drift_dry_run_smoke():
|
| with tempfile.TemporaryDirectory() as tmp:
|
| run_drift_grpo_dry_run(
|
| output_dir=tmp, total_episodes=3, group_size=2, seed=0
|
| )
|
| log_path = Path(tmp) / "drift_dry_run.json"
|
| assert log_path.exists()
|
| log = json.loads(log_path.read_text())
|
| assert len(log) == 3
|
| for entry in log:
|
| assert "rewards" in entry and "candidates" in entry
|
| assert all(0.0 <= r <= 2.0 for r in entry["rewards"])
|
|
|
