Spaces:
Runtime error
Runtime error
File size: 3,257 Bytes
f8cc947 | 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 | import pytest
from unittest.mock import MagicMock
import httpx
import inference
def test_success_true_when_score_above_threshold(monkeypatch, capsys):
# Mock environment server and openai
mock_client = MagicMock()
mock_post = MagicMock()
def fake_post(url, json=None, timeout=None):
resp = MagicMock()
resp.raise_for_status = lambda: None
if "reset" in url:
resp.json.return_value = {"max_steps": 2}
else:
# step
operation = json.get("operation")
if operation == "done":
resp.json.return_value = {
"observation": {}, "reward": 0.99, "done": True,
"info": {"current_score": 0.40}
}
else:
resp.json.return_value = {
"observation": {}, "reward": 0.20, "done": False,
"info": {"current_score": 0.20}
}
return resp
mock_post.side_effect = fake_post
mock_client.post = mock_post
class FakeContext:
def __enter__(self): return mock_client
def __exit__(self, *args): pass
monkeypatch.setattr(httpx, "Client", lambda: FakeContext())
mock_llm = MagicMock()
mock_create = MagicMock()
mock_create.side_effect = [
MagicMock(choices=[MagicMock(message=MagicMock(content='{"operation": "add_comment", "line_number": 1, "severity": "minor", "category": "bug", "message": "issue"}'))]),
MagicMock(choices=[MagicMock(message=MagicMock(content='{"operation": "done"}'))])
]
mock_llm.chat.completions.create = mock_create
monkeypatch.setattr(inference, "OpenAI", lambda **kwargs: mock_llm)
# Run
inference.run_task("easy", env_base_url="fake", api_base_url="fake", model_name="fake", hf_token="fake", timeout_s=10)
# Check
captured = capsys.readouterr()
assert "[END] success=true" in captured.out
def test_success_false_when_invalid_model(monkeypatch, capsys):
class FakeContext:
def __enter__(self):
c = MagicMock()
c.post.return_value.json.return_value = {"max_steps": 2}
return c
def __exit__(self, *args): pass
monkeypatch.setattr(httpx, "Client", lambda: FakeContext())
def mock_raise(**kwargs):
raise Exception("Error code: 400 - Invalid model")
mock_llm = MagicMock()
mock_llm.chat.completions.create = mock_raise
monkeypatch.setattr(inference, "OpenAI", lambda **kwargs: mock_llm)
# Run
inference.run_task("easy", env_base_url="fake", api_base_url="fake", model_name="fake", hf_token="fake", timeout_s=10)
# Check
captured = capsys.readouterr()
assert "[END] success=false" in captured.out
def test_llm_mode_calls_api_not_deterministic_fallback(monkeypatch):
monkeypatch.setenv("REVIEW_STRATEGY", "llm")
action = inference._get_benchmark_action("hard", 1)
assert action is None
def test_hard_task_system_prompt_contains_no_line_numbers():
prompt = inference.load_system_prompt()
lines = ["line 23", "line 28", "line 34", "line 40", "line 18", "line 26"]
for l in lines:
assert l not in prompt.lower()
|