Spaces:
Sleeping
Sleeping
File size: 3,316 Bytes
90fc756 bf2775e b83c8ad bf2775e | 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 | from fastapi.testclient import TestClient
from server.app import create_app
from server.environment import SQLReviewEnvironment
def build_client() -> TestClient:
return TestClient(create_app(SQLReviewEnvironment()))
def test_reset_returns_initial_observation() -> None:
client = build_client()
response = client.post("/reset", json={"task_id": "easy_001"})
assert response.status_code == 200
payload = response.json()
assert payload["observation"]["difficulty"] == "easy"
assert payload["reward"] == 0.0
assert payload["done"] is False
def test_identify_issue_returns_positive_reward_for_match() -> None:
client = build_client()
client.post("/reset", json={"task_id": "easy_002"})
response = client.post(
"/step",
json={
"action_type": "identify_issue",
"issue_category": "syntax",
"issue_description": "The query is missing the FROM clause before users.",
"confidence": 0.95,
},
)
assert response.status_code == 200
payload = response.json()
assert payload["reward"] > 0
assert payload["info"]["issue_id"] == "easy_002_missing_from"
def test_suggest_fix_without_identifying_issue_is_penalized() -> None:
client = build_client()
client.post("/reset", json={"task_id": "easy_002"})
response = client.post(
"/step",
json={
"action_type": "suggest_fix",
"suggested_fix": "SELECT id, email FROM users WHERE active = 1;",
"confidence": 0.8,
},
)
assert response.status_code == 200
assert response.json()["reward"] < 0
def test_approve_with_missed_issues_ends_episode_with_penalty() -> None:
client = build_client()
client.post("/reset", json={"task_id": "easy_001"})
response = client.post("/step", json={"action_type": "approve", "confidence": 0.8})
assert response.status_code == 200
payload = response.json()
assert payload["done"] is True
assert payload["reward"] < 0
assert payload["info"]["final_score"] is not None
def test_identify_then_approve_can_finish_successfully() -> None:
client = build_client()
client.post("/reset", json={"task_id": "easy_002"})
client.post(
"/step",
json={
"action_type": "identify_issue",
"issue_category": "syntax",
"issue_description": "The query is missing the FROM clause before users.",
"confidence": 0.95,
},
)
response = client.post("/step", json={"action_type": "approve", "confidence": 0.9})
assert response.status_code == 200
payload = response.json()
assert payload["done"] is True
assert payload["reward"] > 0
assert payload["info"]["final_score"] is not None
def test_request_more_context_returns_context_shared_flag() -> None:
client = build_client()
client.post("/reset", json={"task_id": "easy_001"})
response = client.post(
"/step",
json={"action_type": "request_more_context", "confidence": 0.7},
)
assert response.status_code == 200
payload = response.json()
assert payload["reward"] == -0.03
assert "context_shared" in payload["info"]
assert payload["info"]["context_shared"] is True
assert payload["done"] is False
|