116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
"""Sandbox + agent driver tests.
|
|
|
|
Run: .venv/bin/python tests/test_sandbox_agent.py
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from evalharness import get_dataset # noqa: E402
|
|
from evalharness.agent import BFCLEnvironment, drive, trajectory_to_prediction # noqa: E402
|
|
from evalharness.data.sample import Sample # noqa: E402
|
|
from evalharness.model import resolve_adapter, run_eval # noqa: E402
|
|
from evalharness.sandbox import get_sandbox # noqa: E402
|
|
|
|
|
|
def test_local_sandbox():
|
|
sbx = get_sandbox('local')
|
|
r = sbx.exec({'main.py': 'print(6*7)'})
|
|
assert r.ok and '42' in r.stdout
|
|
r2 = sbx.exec({'main.py': 'raise RuntimeError("boom")'})
|
|
assert not r2.ok and 'boom' in r2.stderr
|
|
r3 = sbx.exec({'sub/mod.py': 'x=1', 'main.py': 'print("m")'}) # nested files
|
|
assert r3.ok
|
|
|
|
|
|
def test_docker_sandbox_if_available():
|
|
from evalharness.sandbox import docker_available
|
|
|
|
if not docker_available():
|
|
print(' (docker unavailable, skipped)')
|
|
return
|
|
sbx = get_sandbox('docker')
|
|
r = sbx.exec({'main.py': 'print("docker-ok")'}, timeout_s=300)
|
|
assert r.ok and 'docker-ok' in r.stdout, r.stderr[:200]
|
|
r2 = sbx.exec({'main.py': 'import socket\n'
|
|
's=socket.create_connection(("1.1.1.1", 80), timeout=5)\nprint("net-ok")'},
|
|
timeout_s=120)
|
|
assert not r2.ok # --network none: outbound connect must fail
|
|
|
|
|
|
def test_env_handle_refcount():
|
|
from evalharness.sandbox.base import EnvHandle, _ACTIVE, acquire
|
|
|
|
stops = []
|
|
|
|
def start(name):
|
|
return {'api_base': f'http://x/{name}', 'model': name, 'container': f'c-{name}'}
|
|
|
|
h = acquire('t-engine', 'm1', start, lambda h: stops.append(h.container))
|
|
h2 = acquire('t-engine', 'm1', start, lambda h: stops.append(h.container))
|
|
assert h2 is h and h.refs == 2
|
|
h.release() # shared: must NOT stop
|
|
assert stops == []
|
|
h.release() # last ref: stop + unregister
|
|
assert stops == ['c-m1'] and 't-engine/m1' not in _ACTIVE
|
|
h.release() # idempotent
|
|
assert stops == ['c-m1']
|
|
|
|
|
|
def test_agent_drive_single_turn():
|
|
adapter = resolve_adapter('mock')
|
|
|
|
async def go():
|
|
return await drive(adapter, Sample(input='hi', target='hi'))
|
|
|
|
traj = asyncio.run(go())
|
|
assert traj.turns == 1 and traj.messages[-1]['role'] == 'assistant'
|
|
|
|
|
|
def test_bfcl_env_records_calls():
|
|
adapter = resolve_adapter('mock')
|
|
adapter.extra['mode'] = 'fc'
|
|
gt = '{"tool_calls": [{"name": "f", "arguments": {"a": 1}}]}'
|
|
sample = Sample(input='call f', target=gt, task_type='fc')
|
|
|
|
async def go():
|
|
return await drive(adapter, sample, env=BFCLEnvironment(), max_turns=4)
|
|
|
|
traj = asyncio.run(go())
|
|
assert traj.env_state['calls'] == [{'name': 'f', 'arguments': {'a': 1}}]
|
|
pred = trajectory_to_prediction(traj)
|
|
assert pred['env_state']['ground_truth']['tool_calls'][0]['name'] == 'f'
|
|
|
|
|
|
def test_bfcl_pipeline_oracle():
|
|
ds = get_dataset('bfcl_v3')
|
|
rep = asyncio.run(run_eval(ds, 'mock:fc', env='bfcl_mock', limit=30,
|
|
concurrency=8, progress=False))
|
|
assert rep.metrics['acc'] == 1.0
|
|
assert rep.metric_groups['acc']['irrelevance'] == 1.0
|
|
|
|
|
|
def test_humaneval_pipeline_oracle():
|
|
ds = get_dataset('humaneval')
|
|
rep = asyncio.run(run_eval(ds, 'mock:oracle', limit=10, concurrency=4, progress=False))
|
|
assert rep.metrics['pass'] == 1.0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
fails = 0
|
|
for name, fn in sorted({k: v for k, v in globals().items()
|
|
if k.startswith('test_') and callable(v)}.items()):
|
|
try:
|
|
fn()
|
|
print(f'PASS {name}')
|
|
except AssertionError as e:
|
|
fails += 1
|
|
print(f'FAIL {name}: {e}')
|
|
except Exception as e:
|
|
fails += 1
|
|
print(f'ERROR {name}: {type(e).__name__}: {e}')
|
|
sys.exit(1 if fails else 0)
|