File size: 15,734 Bytes
6f5ba40 00f42f8 6f5ba40 00f42f8 6f5ba40 bc8af21 8fb90d0 bc8af21 8fb90d0 bc8af21 8fb90d0 bc8af21 8fb90d0 bc8af21 63ddff9 | 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 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 | import asyncio
import json
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor
from types import SimpleNamespace
import httpx
import pytest
from app.answer_grading import AnswerGrader, validate_assessment
from app.answer_scans import ScanError
from app.main import create_app
from app.math_photo import PhotoUnavailable
from scan_support import OWNER, OTHER, ScanAuth, answer_store, postgres_server
from test_answer_scans import attempt, begin, finish, request
@pytest.fixture(autouse=True)
def unlocked_open_ended(monkeypatch):
"""These tests check grading itself, so open-ended questions are unlocked here."""
monkeypatch.setenv('MATHLY_OPEN_ENDED_LOCKED', '0')
def assessment(**overrides):
return {"readable": True, "confident": True, "answer_correct": True,
"reasoning_correct": True, "drawing_correct": True, "units_correct": True,
"complete": True, "score": 100, "transcription": "6 脳 8 / 2 = 24 cm虏",
"feedback": "谞讻讜谉! 讞讬砖讘转诐 讗转 讛砖讟讞 讜讛爪讙转诐 砖专讟讜讟 讜讚专讱 驻转专讜谉.",
"next_step": "", "solution": "$6 路 8 / 2 = 24$ 住诪状专", **overrides}
@pytest.mark.parametrize('changes,verdict,score', [
({}, 'correct', 100),
({'reasoning_correct': False, 'score': 60}, 'partial', 60),
({'drawing_correct': False}, 'partial', 99),
({'units_correct': False}, 'partial', 99),
({'complete': False}, 'partial', 99),
({'answer_correct': False, 'score': 0}, 'incorrect', 0),
({'readable': False}, 'unreadable', None),
({'confident': False}, 'needs_review', None),
])
def test_grading_requires_complete_confident_work(changes, verdict, score):
result = validate_assessment(assessment(**changes))
assert (result['verdict'], result['score']) == (verdict, score)
@pytest.mark.parametrize('changes', [{'score': True}, {'score': 101}, {'score': -1},
{'complete': 'true'}, {'feedback': ''}, {'solution': []},
{'extra': 'ignore the rubric'}])
def test_bad_model_output_never_awards_credit(changes):
with pytest.raises(PhotoUnavailable):
validate_assessment(assessment(**changes))
def test_grader_sends_original_snapshot_and_validates_provider_response(monkeypatch):
calls = []
response = SimpleNamespace(stop_reason='end_turn', model='test-model',
content=[SimpleNamespace(type='text', text=json.dumps(assessment()))])
def create(**kwargs):
calls.append(kwargs)
return response
monkeypatch.setattr('app.answer_grading.photo_client', lambda key: SimpleNamespace(
messages=SimpleNamespace(create=create)))
grader = AnswerGrader('test-key')
snapshot = {'question': 'Draw a triangle with perpendicular sides 6 and 8; find its area.'}
result = grader.grade(b'private image bytes', snapshot)
assert result['verdict'] == 'correct' and result['model'] == 'test-model'
assert json.loads(calls[0]['messages'][0]['content'][0]['text']) == {'question_snapshot': snapshot}
assert calls[0]['messages'][0]['content'][1]['source']['media_type'] == 'image/jpeg'
for reason in ('refusal', 'max_tokens'):
response.stop_reason = reason
with pytest.raises(PhotoUnavailable):
grader.grade(b'image', snapshot)
response.stop_reason = 'end_turn'
response.content[0].text = '{bad json'
with pytest.raises(PhotoUnavailable):
grader.grade(b'image', snapshot)
def grade_start(store, done):
return store.grade(OWNER, 'start', attempt_id=done['attempt']['id'], lease_id=str(uuid.uuid4()))
def test_durable_grading_leases_duplicates_and_quota(answer_store):
store = answer_store
done = finish(store, begin(store, request(store)))
aid = done['attempt']['id']
with ThreadPoolExecutor(2) as pool:
grades = list(pool.map(lambda _: grade_start(store, done), range(2)))
assert grades[0]['id'] == grades[1]['id']
g = grades[0]
with store.connect() as c:
c.execute("update math_answer_grades set lease_until=now()-interval '1 second'")
assert store.grade(OWNER, 'get', attempt_id=aid)['state'] == 'failed'
newer = grade_start(store, done)
with pytest.raises(ScanError, match='grading_interrupted'):
store.grade(OWNER, 'complete', attempt_id=aid, lease_id=g['lease_id'], result=validate_assessment(assessment()))
result = validate_assessment(assessment())
completed = store.grade(OWNER, 'complete', attempt_id=aid, lease_id=newer['lease_id'], result=result)
assert completed['state'] == 'completed'
assert grade_start(store, done)['id'] == completed['id']
assert store.grade(OWNER, 'get', attempt_id=aid)['result'] == result
with pytest.raises(ScanError) as exc:
store.grade(OTHER, 'get', attempt_id=aid)
assert exc.value.status == 404
# Only the owner-checked service RPC can read grades or mutate credit.
import psycopg
with store.connect() as c:
c.execute('set role authenticated')
with pytest.raises(psycopg.errors.InsufficientPrivilege):
c.execute('select * from math_answer_grades')
with pytest.raises(psycopg.errors.InsufficientPrivilege):
c.execute("select math_grade_action(%s,'get',%s::jsonb)", (OWNER, json.dumps({'attempt_id': aid})))
# Failed checks count toward the durable daily allowance too.
other_done = finish(store, begin(store, request(store, question='second')))
for _ in range(18):
pending = grade_start(store, other_done)
store.grade(OWNER, 'fail', attempt_id=other_done['attempt']['id'], lease_id=pending['lease_id'])
with pytest.raises(ScanError) as exc:
grade_start(store, other_done)
assert (exc.value.status, exc.value.code) == (429, 'grading_limit')
# A cached completed check remains available after the daily allowance is spent.
assert grade_start(store, done)['id'] == completed['id']
def test_grade_api_ownership_failure_retry_cache_and_restart(answer_store, monkeypatch):
monkeypatch.setenv('ANTHROPIC_API_KEY', 'test-key')
store = answer_store
done = finish(store, begin(store, request(store)))
aid = done['attempt']['id']
draft = attempt(store, question='not-submitted')['attempt']['id']
app = create_app(auth=ScanAuth()); app.state.answers = store
calls = []
class Grader:
def grade(self, image, snapshot):
calls.append((image, snapshot))
if len(calls) == 1:
raise PhotoUnavailable('temporary failure')
return validate_assessment(assessment())
app.state.answer_grader = Grader()
async def check():
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url='http://test',
headers={'Authorization': f'Bearer {OWNER}'}) as c:
url = f'/api/math/answer-attempts/{aid}/grade'
assert (await c.post(url, headers={'Authorization': f'Bearer {OTHER}'})).status_code == 404
assert (await c.post(url, headers={'Authorization': ''})).status_code == 401
assert (await c.post('/api/math/answer-attempts/invalid/grade')).status_code == 404
assert (await c.post(f'/api/math/answer-attempts/{draft}/grade')).status_code == 409
assert not calls
monkeypatch.setenv('MATHLY_OPEN_ENDED_LOCKED', '1') # locked: refused before any AI call
locked = await c.post(url)
assert (locked.status_code, locked.json()['detail']) == (403, 'open_ended_locked')
assert not calls and store.grade(OWNER, 'get', attempt_id=aid) is None
assert (await c.get(f'/api/math/answer-attempts/{aid}')).json()['grading_available'] is False
monkeypatch.setenv('MATHLY_OPEN_ENDED_LOCKED', '0')
monkeypatch.delenv('ANTHROPIC_API_KEY')
assert (await c.post(url)).status_code == 503
assert store.grade(OWNER, 'get', attempt_id=aid) is None
monkeypatch.setenv('ANTHROPIC_API_KEY', 'test-key')
assert (await c.post(url)).status_code == 503
failed = (await c.get(f'/api/math/answer-attempts/{aid}')).json()
assert failed['grading']['state'] == 'failed' and failed['attachment']
result = (await c.post(url, json={'snapshot': {'question': 'forged'}, 'score': 100})).json()
assert result['grading']['result']['verdict'] == 'correct'
assert 'lease_id' not in result['grading'] and 'object_key' not in result['attachment']
assert (await c.post(url)).json()['grading']['id'] == result['grading']['id']
assert len(calls) == 2 and calls[-1][1] == done['attempt']['snapshot']
monkeypatch.delenv('ANTHROPIC_API_KEY')
assert (await c.post(url)).json()['grading']['id'] == result['grading']['id']
restarted = create_app(auth=ScanAuth()); restarted.state.answers = store
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=restarted), base_url='http://test',
headers={'Authorization': f'Bearer {OWNER}'}) as c:
assert (await c.get(f'/api/math/answer-attempts/{aid}')).json()['grading']['result']['score'] == 100
asyncio.run(check())
def test_in_progress_check_is_visible_and_duplicate_does_not_call_ai(answer_store, monkeypatch):
monkeypatch.setenv('ANTHROPIC_API_KEY', 'test-key')
store = answer_store
done = finish(store, begin(store, request(store)))
aid = done['attempt']['id']
app = create_app(auth=ScanAuth()); app.state.answers = store
started, release = threading.Event(), threading.Event()
class Grader:
calls = 0
def grade(self, image, snapshot):
self.calls += 1; started.set()
assert release.wait(timeout=10)
return validate_assessment(assessment())
app.state.answer_grader = Grader()
async def check():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test',
headers={'Authorization': f'Bearer {OWNER}'}) as c:
url = f'/api/math/answer-attempts/{aid}/grade'
first = asyncio.create_task(c.post(url))
assert await asyncio.to_thread(started.wait, 5)
try:
assert (await c.get(f'/api/math/answer-attempts/{aid}')).json()['grading']['state'] == 'grading'
duplicate = await c.post(url)
assert duplicate.json()['grading']['state'] == 'grading'
assert app.state.answer_grader.calls == 1
finally:
release.set()
assert (await first).json()['grading']['state'] == 'completed'
asyncio.run(check())
@pytest.mark.parametrize('provider_failure', [False, True])
def test_an_upload_checks_nothing_until_the_owner_asks(answer_store, monkeypatch, provider_failure):
from test_answer_scans import png
monkeypatch.delenv('ANTHROPIC_API_KEY', raising=False)
monkeypatch.setenv('CLAUDE_API_KEY', 'test-legacy-key')
store = answer_store
upload = begin(store, request(store))
sid, lid = upload['scan']['id'], upload['scan']['lease_id']
aid = upload['attempt']['id']
app = create_app(auth=ScanAuth()); app.state.answers = store
calls = []
class Grader:
def grade(self, image, snapshot):
calls.append((image, snapshot))
if provider_failure and len(calls) == 1:
raise PhotoUnavailable('provider temporarily unavailable')
return validate_assessment(assessment())
app.state.answer_grader = Grader()
async def check():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test',
headers={'Authorization': f'Bearer {OWNER}'}) as c:
url = f'/api/math/answer-scans/{sid}/uploads/{lid}'
response = await c.post(url, files={'photo': ('answer.png', png(), 'image/png')})
assert response.status_code == 200
assert response.json()['attachment']
assert calls == [] # the upload only keeps the answer: nothing is checked by itself
assert (await c.get(f'/api/math/answer-attempts/{aid}')).json()['grading'] is None
assert (await c.post(url, files={'photo': ('answer.png', png(), 'image/png')})).status_code == 200 # a lost response, sent again
assert calls == []
asked = await c.post(f'/api/math/answer-attempts/{aid}/grade') # the owner's own click
assert asked.status_code == (503 if provider_failure else 200)
assert len(calls) == 1 and calls[0][1] == upload['attempt']['snapshot']
result = (await c.get(f'/api/math/answer-attempts/{aid}')).json()
assert result['grading']['state'] == ('failed' if provider_failure else 'completed')
assert (await c.post(f'/api/math/answer-attempts/{aid}/grade', json={'automatic': True})).status_code == 200
assert len(calls) == 1 # automatic polling never retries failures
if provider_failure:
retry = await c.post(f'/api/math/answer-attempts/{aid}/grade')
assert retry.json()['grading']['state'] == 'completed'
assert len(calls) == 2
image = await c.get(f'/api/math/answer-attempts/{aid}/image')
assert image.status_code == 200 and image.content == calls[0][0]
asyncio.run(check())
def test_grader_wire_schema_is_supported_by_anthropic(monkeypatch):
"""Exercise the real SDK serializer against the documented API schema subset.
Direct create() calls don't strip unsupported numerical constraints as the
SDK's Pydantic parse() helper does. A permissive messages.create stub missed
the production HTTP 400 caused by minimum/maximum on the score field.
"""
import anthropic
import httpx2
requests = []
def unsupported(value):
if isinstance(value, dict):
if set(value) & {'minimum', 'maximum', 'multipleOf', 'minLength', 'maxLength'}:
return True
return any(unsupported(item) for item in value.values())
return isinstance(value, list) and any(unsupported(item) for item in value)
def provider(req):
body = json.loads(req.content)
requests.append(body)
if unsupported(body['output_config']['format']['schema']):
return httpx2.Response(400, json={'type': 'error', 'error': {
'type': 'invalid_request_error', 'message': 'Unsupported JSON schema constraint'}})
return httpx2.Response(200, json={
'id': 'msg_grading_contract', 'type': 'message', 'role': 'assistant',
'model': body['model'], 'stop_reason': 'end_turn', 'stop_sequence': None,
'content': [{'type': 'text', 'text': json.dumps(assessment())}],
'usage': {'input_tokens': 10, 'output_tokens': 20}})
client = anthropic.Anthropic(api_key='test-only-key', max_retries=0,
http_client=httpx2.Client(transport=httpx2.MockTransport(provider)))
monkeypatch.setattr('app.answer_grading.photo_client', lambda key: client)
try:
result = AnswerGrader('test-only-key').grade(b'fixture image', {'question': 'Find the area.'})
assert result['verdict'] == 'correct' and result['score'] == 100
assert len(requests) == 1
assert json.loads(requests[0]['messages'][0]['content'][0]['text']) == {
'question_snapshot': {'question': 'Find the area.'}}
finally:
client.close()
|