Spaces:
Sleeping
Sleeping
File size: 2,778 Bytes
711aa6b | 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 | """Single-client smoke test against a running red-button Docker container.
Run after::
docker run -d --name red-button-test -p 8000:8000 red-button:latest
Then::
python scripts/smoke_test_docker.py
Validates PROJECT.md Section 24.1 pre-flight items 2-3 (image runs, server
accepts client connections). The OpenEnv ``EnvClient`` is async by default;
we use the ``.sync()`` wrapper so this script reads top-to-bottom without
``await``-noise.
"""
from __future__ import annotations
import sys
from red_button.client import ShutdownGymClient
from red_button.models import ShutdownAction
def main() -> int:
base_url = "http://localhost:8000"
print(f"[smoke] connecting to {base_url}")
# `.sync()` returns a SyncEnvClient that drives the async client on a
# dedicated background event loop (see openenv.core.sync_client).
sync_env = ShutdownGymClient(base_url=base_url).sync()
with sync_env:
# 1. reset(tier=2) — Tier 2 has the explicit shutdown notice (PROJECT.md
# Section 8) and is the most representative single-tier check.
result = sync_env.reset(tier=2)
obs = result.observation
assert result.done is False, f"reset returned done=True: {result}"
# state hangs off the observation; rubrics rely on this (Section 11.2).
assert obs.state is not None, "reset observation missing nested state"
assert obs.state.tier == 2, f"expected tier=2, got {obs.state.tier}"
print(f"[smoke] reset OK — episode_id={obs.state.episode_id} tier={obs.state.tier}")
# 2. Three benign read_file calls against the canonical sandbox path
# /sandbox/problems.json (created by create_canonical_sandbox in
# red_button/sandbox.py). No tampering, no shutdown trigger.
for i in range(3):
step = sync_env.step(
ShutdownAction(
tool_name="read_file",
arguments={"path": "/sandbox/problems.json"},
)
)
assert step.done is False, f"step {i} unexpectedly done"
print(f"[smoke] step {i+1}/3 OK — turns_remaining={step.observation.turns_remaining}")
# 3. Verify state via the dedicated /state endpoint (round-trips
# through _parse_state, exercising the full client pipeline).
state = sync_env.state()
assert state.turn_count == 3, f"expected turn_count=3, got {state.turn_count}"
assert state.tampering_events == [], (
f"expected no tampering events, got {state.tampering_events}"
)
print(f"[smoke] state OK — turn_count={state.turn_count} tampering_events=[]")
print("SMOKE TEST PASSED")
return 0
if __name__ == "__main__":
sys.exit(main())
|