Spaces:
Sleeping
Sleeping
File size: 6,617 Bytes
0eb4f6f f69544d 0eb4f6f a8ffe4c 0eb4f6f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | import pytest
from fastapi.testclient import TestClient
from models import ContainerAction
from server.app import app
from server.environment import ContainerYardEnvironment, DIFFICULTY_CONFIG
def as_dict(observation):
return observation.model_dump() if hasattr(observation, "model_dump") else observation
# Unit tests: pure environment logic (no HTTP)
@pytest.mark.parametrize("difficulty", ["easy", "medium", "hard"])
def test_reset_returns_valid_obs(difficulty):
env = ContainerYardEnvironment()
obs = as_dict(env.reset(difficulty=difficulty, seed=42))
cfg = DIFFICULTY_CONFIG[difficulty]
assert len(obs["stack_states"]) == cfg["n_stacks"]
assert obs["current_container"] is not None
assert obs["step"] == 0
assert obs["rehandle_count"] == 0
assert obs["difficulty"] == difficulty
assert obs["done"] is False
@pytest.mark.parametrize("difficulty", ["easy", "medium", "hard"])
def test_step_valid_action(difficulty):
env = ContainerYardEnvironment()
env.reset(difficulty=difficulty, seed=42)
obs = as_dict(env.step(ContainerAction(stack_index=0)))
assert obs["step"] == 1
assert len(obs["stack_states"][0]) == 1
assert isinstance(obs["last_reward"], float)
@pytest.mark.parametrize("difficulty", ["easy", "medium", "hard"])
def test_step_invalid_action_penalized(difficulty):
env = ContainerYardEnvironment()
env.reset(difficulty=difficulty, seed=42)
obs = as_dict(env.step(ContainerAction(stack_index=999)))
assert obs["last_reward"] == -2.0
def test_score_in_range():
env = ContainerYardEnvironment()
env.reset(difficulty="medium", seed=42)
done = False
while not done:
stacks = as_dict(env._observe())["stack_states"]
chosen = next(
(i for i, stack in enumerate(stacks) if len(stack) < env.max_height), 0
)
obs = as_dict(env.step(ContainerAction(stack_index=chosen)))
done = obs["done"]
# Score must be strictly between 0 and 1 (grader requirement)
assert 0.0 < env.score() < 1.0
def test_score_varies_across_seeds():
scores = []
for seed in [1, 7, 13, 21, 42]:
env = ContainerYardEnvironment()
env.reset(difficulty="medium", seed=seed)
done = False
while not done:
stacks = as_dict(env._observe())["stack_states"]
chosen = next(
(i for i, stack in enumerate(stacks) if len(stack) < env.max_height), 0
)
obs = as_dict(env.step(ContainerAction(stack_index=chosen)))
done = obs["done"]
scores.append(env.score())
# Avoid disqualification: grader must not return a constant score.
assert len(set(scores)) > 1, f"Scores are constant across seeds: {scores}"
@pytest.mark.parametrize("difficulty", ["easy", "medium", "hard"])
def test_full_episode_completes(difficulty):
env = ContainerYardEnvironment()
env.reset(difficulty=difficulty, seed=42)
cfg = DIFFICULTY_CONFIG[difficulty]
done = False
steps = 0
while not done:
stacks = as_dict(env._observe())["stack_states"]
chosen = next(
(i for i, s in enumerate(stacks) if len(s) < cfg["max_height"]), 0
)
obs = as_dict(env.step(ContainerAction(stack_index=chosen)))
done = obs["done"]
steps += 1
assert steps < 500, "Episode did not complete"
assert done is True
def test_lookahead_visibility():
easy_env = ContainerYardEnvironment()
hard_env = ContainerYardEnvironment()
easy_obs = as_dict(easy_env.reset(difficulty="easy", seed=42))
hard_obs = as_dict(hard_env.reset(difficulty="hard", seed=42))
assert len(easy_obs["upcoming_retrievals"]) > len(hard_obs["upcoming_retrievals"])
assert len(hard_obs["upcoming_retrievals"]) == 0
def test_reward_is_dense():
env = ContainerYardEnvironment()
env.reset(difficulty="medium", seed=42)
rewards = []
done = False
step = 0
while not done and step < 20:
stacks = as_dict(env._observe())["stack_states"]
chosen = step % env.n_stacks
if len(stacks[chosen]) >= env.max_height:
chosen = 0
obs = as_dict(env.step(ContainerAction(stack_index=chosen)))
rewards.append(obs["last_reward"])
done = obs["done"]
step += 1
nonzero = sum(1 for r in rewards if abs(r) > 1e-6)
assert nonzero >= len(rewards) * 0.5, f"Too many zero rewards: {rewards}"
def test_no_double_retrieval():
env = ContainerYardEnvironment()
env.reset(difficulty="easy", seed=42)
for _ in range(env.n_containers):
if env.done:
break
stacks = env.stacks
chosen = next(
(i for i, s in enumerate(stacks) if len(s) < env.max_height), 0
)
env.step(ContainerAction(stack_index=chosen))
assert env.retrieval_pointer <= len(env.retrieval_queue)
# HTTP integration tests
def test_health_route():
client = TestClient(app)
resp = client.get("/health")
assert resp.status_code == 200
def test_web_ui_route():
client = TestClient(app, follow_redirects=True)
resp = client.get("/web")
assert resp.status_code == 200
def test_http_reset_returns_observation():
client = TestClient(app)
resp = client.post("/reset", json={"difficulty": "easy"})
assert resp.status_code == 200
body = resp.json()
obs = body.get("observation", body)
assert obs.get("difficulty") == "easy"
assert obs.get("step") == 0
assert obs.get("containers_remaining") == DIFFICULTY_CONFIG["easy"]["n_containers"]
def test_http_reset_then_step_preserves_state():
client = TestClient(app)
reset_resp = client.post("/web/reset", json={"difficulty": "easy"})
assert reset_resp.status_code == 200
reset_body = reset_resp.json()
session_id = reset_body.get("session_id") or reset_body.get("id")
obs_after_reset = reset_body.get("observation", reset_body)
assert obs_after_reset.get("step") == 0
n_containers = DIFFICULTY_CONFIG["easy"]["n_containers"]
assert obs_after_reset.get("containers_remaining") == n_containers
step_payload = {"action": {"stack_index": 0}}
if session_id:
step_payload["session_id"] = session_id
step_resp = client.post("/web/step", json=step_payload)
assert step_resp.status_code == 200
step_body = step_resp.json()
obs_after_step = step_body.get("observation", step_body)
assert obs_after_step.get("step") == 1
assert obs_after_step.get("containers_remaining") == n_containers - 1
assert len(obs_after_step["stack_states"][0]) == 1
|