115 lines
5.1 KiB
Python
115 lines
5.1 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from atelier.provider import Provider
|
|
|
|
ORIGINAL = 'Create four games on port 8099. There is no time limit. You have about 5 hours. If needed, use web/image search.'
|
|
BAD = 'Create four games on port 8099. Time limit: 5 hours. You must use web/image search.'
|
|
ISSUES = [{'description': 'Zeitlimit erfunden', 'original_quote': 'There is no time limit.', 'draft_quote': 'Time limit: 5 hours.'}, {'description': 'Optionale Recherche verpflichtend gemacht', 'original_quote': 'If needed, use web/image search.', 'draft_quote': 'You must use web/image search.'}]
|
|
|
|
|
|
def run(responses):
|
|
calls = []
|
|
def transport(request):
|
|
calls.append(json.loads(request.content))
|
|
response = responses[len(calls)-1]
|
|
if isinstance(response, int):
|
|
return httpx.Response(response)
|
|
return httpx.Response(200, json={'choices': [{'message': {'content': response}}]})
|
|
provider = Provider({'base_url': 'http://model/v1', 'chat_model': 'test'}, httpx.MockTransport(transport))
|
|
result = asyncio.run(provider.improve_checked(ORIGINAL, 'Formuliere klarer.'))
|
|
return result, calls
|
|
|
|
|
|
def test_clean_two_calls():
|
|
r, calls = run([ORIGINAL, '{"issues": []}'])
|
|
assert len(calls) == 2
|
|
assert r['review']['status'] == 'passed'
|
|
payload = json.loads(calls[1]['messages'][1]['content'])
|
|
assert payload['original_template'] == ORIGINAL
|
|
assert payload['revision_request'] == 'Formuliere klarer.'
|
|
|
|
|
|
def test_arcade_correction_and_recheck():
|
|
r, calls = run([BAD, json.dumps({'issues': ISSUES}), ORIGINAL, '{"issues": []}'])
|
|
assert len(calls) == 4
|
|
assert r['body'] == ORIGINAL
|
|
assert r['review']['status'] == 'corrected'
|
|
assert len(r['review']['initial_issues']) == 2
|
|
correction = json.loads(calls[2]['messages'][1]['content'])
|
|
assert correction['original_template'] == ORIGINAL and correction['issues'] == ISSUES
|
|
|
|
|
|
def test_remaining_issues_stop_after_four_calls():
|
|
r, calls = run([BAD, json.dumps({'issues': ISSUES}), BAD, json.dumps({'issues': ISSUES})])
|
|
assert len(calls) == 4
|
|
assert r['review']['status'] == 'issues'
|
|
assert len(r['review']['issues']) == 2
|
|
|
|
|
|
@pytest.mark.parametrize('failure', [503, 'not json', '{"issues": "none"}', '{"issues": [{"description":"Wrong", "original_quote":"invented", "draft_quote":""}]}'])
|
|
def test_failed_review_preserves_draft(failure):
|
|
r, calls = run([BAD, failure])
|
|
assert len(calls) == 2
|
|
assert r['body'] == BAD
|
|
assert r['review']['status'] == 'unchecked'
|
|
assert r['review']['warning']
|
|
|
|
|
|
def test_failed_correction_preserves_initial_draft():
|
|
r, calls = run([BAD, json.dumps({'issues': ISSUES}), 500])
|
|
assert len(calls) == 3
|
|
assert r['body'] == BAD and r['review']['status'] == 'unchecked'
|
|
|
|
|
|
def test_failed_final_review_does_not_claim_success():
|
|
r, calls = run([BAD, json.dumps({'issues': ISSUES}), ORIGINAL, 500])
|
|
assert len(calls) == 4
|
|
assert r['body'] == ORIGINAL and r['review']['status'] == 'unchecked'
|
|
|
|
|
|
def test_recheck_is_one_call_and_keeps_text():
|
|
calls=[]
|
|
def handler(request):
|
|
calls.append(json.loads(request.content))
|
|
return httpx.Response(200,json={'choices':[{'message':{'content':'{"issues": []}'}}]})
|
|
p=Provider({'base_url':'http://model/v1','chat_model':'test'},httpx.MockTransport(handler))
|
|
r=asyncio.run(p.recheck(ORIGINAL,'Clarify',BAD))
|
|
assert len(calls)==1 and r['body']==BAD
|
|
assert r['review']['draft_kind']=='current'
|
|
assert r['review']['status']=='passed'
|
|
|
|
|
|
def test_literal_checks():
|
|
from atelier.provider import literal_checks
|
|
assert literal_checks('Use {{name}}, port 8099 and /app/data.', 'Use /app/data. Listen on port 8099 with {{name}}.')==[]
|
|
issues=literal_checks('Use {{name}} on port 8099 at /app/data', 'Use {{user}} on port 8080 at /app/new')
|
|
assert len(issues)==6
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['timeout','connect','json','http'])
|
|
def test_diagnostic_logs_redact_sensitive_data(caplog,kind):
|
|
def handler(request):
|
|
if kind=='timeout':raise httpx.ReadTimeout('SECRET_RESPONSE',request=request)
|
|
if kind=='connect':raise httpx.ConnectError('SECRET_RESPONSE',request=request)
|
|
if kind=='json':return httpx.Response(200,text='SECRET_RESPONSE')
|
|
return httpx.Response(500,text='SECRET_RESPONSE')
|
|
p=Provider({'base_url':'http://secret-host/v1','api_key':'SECRET_KEY','chat_model':'test'},httpx.MockTransport(handler))
|
|
r=asyncio.run(p.recheck('SECRET_PROMPT','SECRET_INSTRUCTION','SECRET_DRAFT'))
|
|
assert r['review']['status']=='unchecked'
|
|
assert r['review']['failed_stage']=='recheck'
|
|
assert 'error_type=' in caplog.text and 'duration_ms=' in caplog.text
|
|
assert not any(v in caplog.text for v in ['SECRET','secret-host'])
|
|
assert 'SECRET' not in r['review']['warning']
|
|
if kind=='json': assert 'HTTP 200' in r['review']['warning']
|
|
if kind=='timeout': assert 'Zeitüberschreitung' in r['review']['warning']
|
|
|
|
|
|
def test_failed_final_review_labels_corrected_draft():
|
|
r,_=run([BAD,json.dumps({'issues':ISSUES}),ORIGINAL,503])
|
|
assert r['review']['draft_kind']=='corrected'
|
|
assert r['review']['failed_stage']=='final_review'
|