69 lines
4.3 KiB
Python
69 lines
4.3 KiB
Python
import pytest
|
|
from starlette.testclient import TestClient
|
|
from viano_bug_analysis.api import APIConfig,Client,Project
|
|
from viano_bug_analysis.api.app import create_app
|
|
from viano_bug_analysis.api.cases import case_key
|
|
from viano_bug_analysis.package import PackageReader
|
|
from test_async_tests import setup, hypotheses, case_setup, built_case, interpreted_case
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def project_manifest(manifest):
|
|
manifest['project']={'id':'p','type':'wow','profile':'vcore'}
|
|
|
|
|
|
@pytest.mark.parametrize('outcome',['completed','unable_to_test'])
|
|
def test_web_human_feedback_auth_scope_replay_and_resume(hypotheses,monkeypatch,outcome):
|
|
store,budget,coordinator,before,request,_=setup(hypotheses)
|
|
coordinator.submit(request)
|
|
monkeypatch.setenv('TESTER_TOKEN','tester_'+'x'*40)
|
|
monkeypatch.setenv('CAPTURE_TOKEN','capture_'+'x'*40)
|
|
monkeypatch.setenv('OTHER_TOKEN','other_'+'x'*40)
|
|
config=APIConfig(store.workspace.root,(Project('p','wow','vcore'),Project('other','wow','vcore')),
|
|
(Client('tester','TESTER_TOKEN',('p',),tester=True),Client('capture','CAPTURE_TOKEN',('p',)),Client('other','OTHER_TOKEN',('other',),tester=True)))
|
|
app=create_app(config);client=TestClient(app)
|
|
app.state.sessions.users.manage('create_user','alice','test-password')
|
|
url='/v1/test-requests';handle=case_key(*('case-A','stable-package'))
|
|
assert client.get(url).status_code==401
|
|
login=client.post('/v1/auth/login',json={'username':'alice','password':'test-password'})
|
|
assert login.status_code==200
|
|
auth={'X-CSRF-Token':login.json()['csrf']}
|
|
rows=client.get(url,headers=auth).json()
|
|
assert rows['can_answer'] and rows['requests'][0]['project_id']=='p'
|
|
answer={'request_id':request['request_id'],'analysis_revision':1,'outcome':outcome,
|
|
'observations':['Transition absent'],'environment':'build-A','confirmed':True}
|
|
assert client.post(url+'/'+handle,headers={'Authorization':'Bearer capture_'+'x'*40},json=answer).status_code==403
|
|
other={'Authorization':'Bearer other_'+'x'*40}
|
|
assert client.get(url,headers=other).json()['requests']==[]
|
|
assert client.post(url+'/'+handle,headers=other,json=answer).status_code==403
|
|
claim={'request_id':request['request_id'],'analysis_revision':1,'action':'edit'}
|
|
assert client.post(url+'/'+handle,headers=auth,json=answer).status_code==409
|
|
assert client.post(url+'/'+handle+'/claim',headers=auth,json=claim).status_code==200
|
|
app.state.sessions.users.manage('create_user','bob','second-password')
|
|
second=TestClient(app)
|
|
info=second.post('/v1/auth/login',json={'username':'bob','password':'second-password'}).json()
|
|
second_auth={'X-CSRF-Token':info['csrf']}
|
|
assert second.post(url+'/'+handle+'/claim',headers=second_auth,json=claim).status_code==409
|
|
assert second.post(url+'/'+handle,headers=second_auth,json=answer).status_code==409
|
|
assert client.post(url+'/'+handle,headers=auth,json=answer|{'request_id':'wrong'}).status_code==409
|
|
assert client.post(url+'/'+handle,headers=auth,json=answer|{'confirmed':False}).status_code==422
|
|
response=client.post(url+'/'+handle,headers=auth,json=answer)
|
|
assert response.status_code==200 and response.json()['status']=='RESUMED'
|
|
assert client.post(url+'/'+handle,headers=auth,json=answer).json()['reused']
|
|
assert client.post(url+'/'+handle,headers=auth,json=answer|{'observations':['different']}).status_code==409
|
|
assert client.get(url,headers=auth).json()['requests']==[]
|
|
assert budget.snapshot('case-A','stable-package').state==before.state
|
|
with PackageReader(store.latest('case-A','stable-package')) as reader:
|
|
assert reader.manifest['state']=='INVESTIGATING_HYPOTHESES' and reader.manifest['analysis_revision']==1
|
|
|
|
|
|
def test_unavailable_test_executors_are_not_advertised():
|
|
from viano_bug_analysis.analysis.investigation_prompt import output_schema
|
|
def context(costs):
|
|
return {'tool_schema':{},'budget_policy':{'action_costs':costs}}
|
|
schema=output_schema(context({}),strict=True)
|
|
assert not any(b.get('properties',{}).get('kind',{}).get('const')=='test' for b in schema['properties']['next']['oneOf'])
|
|
schema=output_schema(context({'testing.request_human':40}),strict=True)
|
|
branch=next(b for b in schema['properties']['next']['oneOf'] if b['properties']['kind'].get('const')=='test')
|
|
assert branch['properties']['executor']['enum']==['human']
|