import json import sys def test_environment() -> None: print("\n-- Test 1: Environment reset and step --") from environment.sound_env import SoundLimiterEnv, SoundObservation, SoundReward env = SoundLimiterEnv(initial_sound=75.0, noise_std=2.0, seed=42) obs = env.reset(seed=42) assert isinstance(obs, SoundObservation), "reset() must return SoundObservation" assert 0 <= obs.sound_level <= 100, "sound_level out of range" assert 0 <= obs.gain <= 1, "gain out of range" print(f" reset() -> sound_level={obs.sound_level}, gain={obs.gain} [PASS]") for action in range(4): obs2, reward, done, _ = env.step(action) assert isinstance(reward, SoundReward), "step() reward must be SoundReward" assert -2.0 <= reward.value <= 1.25, f"Reward out of expected range: {reward.value}" assert isinstance(done, bool) print(f" step(action={action}) -> reward={reward.value:+.4f}, sound={obs2.sound_level:.1f} dB [PASS]") env.sound_level = 71.0 near_safe = env._compute_reward().value env.sound_level = 84.0 far_from_safe = env._compute_reward().value assert near_safe > far_from_safe, "Reward shaping should provide a usable gradient" print(" reward_gradient_check [PASS]") state = env.state() assert "sound_level" in state, "state() must contain sound_level" print(f" state() -> {state} [PASS]") print(" [PASS] Environment\n") def test_tasks() -> None: print("-- Test 2: Tasks and graders --") from environment.tasks import ALL_TASKS, grade_all_tasks import random assert len(ALL_TASKS) >= 3, "Must have at least 3 tasks" difficulties = [t.difficulty for t in ALL_TASKS] assert "easy" in difficulties, "Missing easy task" assert "medium" in difficulties, "Missing medium task" assert "hard" in difficulties, "Missing hard task" print(f" Found {len(ALL_TASKS)} tasks: {[t.id for t in ALL_TASKS]} [PASS]") def random_agent(_obs_dict: dict) -> int: return random.randint(0, 3) results = grade_all_tasks(agent_fn=random_agent, n_episodes=3) for tid, result in results.items(): assert 0.0 <= result.score <= 1.0, f"Score out of range for {tid}: {result.score}" assert 0.0 <= result.reward <= 1.0, f"Reward out of range for {tid}: {result.reward}" print(f" {tid}: score={result.score:.4f}, reward={result.reward:.4f} [PASS]") print(" [PASS] Tasks and graders\n") def test_inference_format() -> None: print("-- Test 3: Inference log format --") import subprocess test_code = ''' import json from environment.sound_env import ACTION_MAP, SoundLimiterEnv from environment.tasks import ALL_TASKS task = ALL_TASKS[0] env = SoundLimiterEnv( initial_sound=task.initial_sound, noise_std=task.noise_std, max_steps=task.max_steps, seed=42, ) obs = env.reset(seed=42) print(json.dumps({ "type": "[START]", "task_id": task.id, "task": task.name, "difficulty": task.difficulty, "state": env.state(), }), flush=True) done = False step_num = 0 total_reward = 0.0 steps_in_safe = 0 while not done: step_num += 1 action = 2 if obs.sound_level > 70 else 0 obs, reward, done, info = env.step(action) total_reward += reward.value if reward.in_safe_zone: steps_in_safe += 1 print(json.dumps({ "type": "[STEP]", "task_id": task.id, "step": step_num, "action": ACTION_MAP[action], "action_id": action, "sound_level": round(info["sound_level"], 2), "gain": round(info["gain"], 3), "reward": round(reward.value, 4), "in_safe_zone": reward.in_safe_zone, "reward_reason": reward.reason, }), flush=True) score = round(steps_in_safe / step_num, 4) if step_num > 0 else 0.0 print(json.dumps({ "type": "[END]", "task_id": task.id, "task": task.name, "difficulty": task.difficulty, "total_steps": step_num, "total_reward": round(total_reward, 4), "score": score, "reward": score, "passed": score >= task.success_threshold, }), flush=True) ''' try: result = subprocess.run( [sys.executable, "-c", test_code], capture_output=True, text=True, timeout=45, check=False, ) except subprocess.TimeoutExpired as exc: partial_stdout = (exc.stdout or "")[-1200:] partial_stderr = (exc.stderr or "")[-1200:] raise AssertionError( "Inference format subprocess timed out after 45s.\n" f"Partial stdout:\n{partial_stdout}\n" f"Partial stderr:\n{partial_stderr}" ) if result.returncode != 0: raise AssertionError( "Inference format subprocess failed.\n" f"Return code: {result.returncode}\n" f"Stdout:\n{result.stdout[-2000:]}\n" f"Stderr:\n{result.stderr[-2000:]}" ) lines = [line for line in result.stdout.strip().split("\n") if line.strip()] assert len(lines) >= 3, f"Expected >= 3 log lines, got {len(lines)}" first = json.loads(lines[0]) assert first["type"] == "[START]", f"First line must be [START], got {first['type']}" assert "task_id" in first last = json.loads(lines[-1]) assert last["type"] == "[END]", f"Last line must be [END], got {last['type']}" assert "score" in last assert "reward" in last assert 0.0 <= last["score"] <= 1.0 step_lines = [json.loads(line) for line in lines[1:-1]] for step in step_lines: assert step["type"] == "[STEP]" assert "action" in step assert "reward" in step print(f" Log lines: {len(lines)} ({len(step_lines)} steps) [PASS]") print(f" [START] -> task_id={first['task_id']} [PASS]") print(f" [END] -> score={last['score']:.4f} [PASS]") print(" [PASS] Inference log format\n") def test_pydantic_models() -> None: print("-- Test 4: Pydantic typed models --") from environment.sound_env import SoundObservation, SoundReward obs = SoundObservation( sound_level=65.0, gain=0.8, step_count=5, above_safe=False, below_safe=False, loud_streak=0, source_levels=[62.0, 64.0, 66.0], ) assert obs.model_dump()["sound_level"] == 65.0 print(" SoundObservation serializes correctly [PASS]") reward = SoundReward(value=1.0, reason="In safe zone", in_safe_zone=True) assert reward.value == 1.0 print(" SoundReward serializes correctly [PASS]") print(" [PASS] Pydantic models\n") if __name__ == "__main__": print("Sound Limiter Env - Pre-submission Tests") print("=" * 52) errors: list[str] = [] for test_fn in [test_pydantic_models, test_environment, test_tasks, test_inference_format]: try: test_fn() except Exception as exc: print(f" [FAIL] {exc}\n") errors.append(str(exc)) print("=" * 52) if errors: print(f"[FAIL] {len(errors)} test(s) failed:") for err in errors: print(f" - {err}") sys.exit(1) print("[PASS] All tests passed. Ready to submit.\n")