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()