241 lines
14 KiB
Python
241 lines
14 KiB
Python
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
pytest.importorskip('starlette')
|
|
pytest.importorskip('httpx')
|
|
pytest.importorskip('multipart')
|
|
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.dashboard import aggregate, case_view, phase
|
|
from viano_bug_analysis.llm.accounting import classification
|
|
from viano_bug_analysis.lifecycle.storage import atomic_json
|
|
from test_case_builder import case_setup
|
|
|
|
|
|
@pytest.mark.parametrize('operation,rounds,prompt,expected', [
|
|
('case-interpretation:v1:default',1,'case_plan_v2','regular'),
|
|
('case-interpretation:v1:correction-interpretation-1',1,'case_plan_v2','correction'),
|
|
('hypotheses:v1:default',2,'hypotheses_v4','correction'),
|
|
('investigation:v1:H1:1:0:token',1,'investigation_lean_v21','regular'),
|
|
('investigation:v1:H1:1:1:token',1,'investigation_lean_v21','correction'),
|
|
('investigation-startup-review:v1:0:token',1,'startup','regular'),
|
|
('investigation-startup-review:v1:1:token',1,'startup','correction'),
|
|
('entity-resolution:v1:plan:1',1,'entity_resolution_plan_v4','regular'),
|
|
('entity-resolution:v1:assessment:2',1,'entity_resolution_assessment_v4','correction'),
|
|
('diagnosis:v1:early:2:attempt:1',1,'diagnosis_v8','regular'),
|
|
('change-request:v1:attempt:2',1,'change_request_v9','correction'),
|
|
('other',1,'unknown','unknown'),
|
|
])
|
|
def test_call_classification(operation, rounds, prompt, expected):
|
|
assert classification(operation,rounds,prompt) == expected
|
|
|
|
|
|
class Reader:
|
|
def __init__(self, docs, manifest):
|
|
self.docs, self.manifest = docs, manifest
|
|
self.files = docs
|
|
self._archive = SimpleNamespace(getinfo=lambda ref: SimpleNamespace(file_size=len(json.dumps(docs[ref]))))
|
|
def read_bytes(self, ref):
|
|
return json.dumps(self.docs[ref]).encode()
|
|
|
|
|
|
def test_correction_slice_disjoint_derived_excluded_and_thinking_not_doubled(monkeypatch):
|
|
import viano_bug_analysis.api.dashboard as module
|
|
docs = {'case': {'case': {'analysis_revision': 1}, 'bug': {'reports':[{'title':'Testfall'}]}},
|
|
'request': {'prompt_version':'investigation_lean_v21'}}
|
|
chain = []
|
|
for i, category in enumerate(('regular', 'correction', 'regular')):
|
|
ref = f'call{i}'
|
|
docs[ref] = {'rounds':1, 'phase':'INVESTIGATING_HYPOTHESES', 'request_ref':'request',
|
|
'derived': i==2, 'outcome': {'usage_classification':category,
|
|
'metadata':{'input_tokens':100,'output_tokens':40,'reasoning_tokens':30,'cached_input_tokens':20}}}
|
|
chain.append(('cp'+str(i), {'details':{'operation':'llm-round','result_ref':ref}, 'created_at':'2026-10-21T10:00:00Z'}))
|
|
# Duplicate checkpoint provenance does not charge a provider call twice.
|
|
chain.append(chain[0])
|
|
monkeypatch.setattr(module,'checkpoint_chain',lambda reader:chain)
|
|
reader=Reader(docs,{'case_id':'BUG-1','package_id':'one','analysis_revision':1,'package_generation':10,
|
|
'state':'INVESTIGATING_HYPOTHESES','artifacts':{'case':'case'},'project':{'id':'p'},'updated_at':'now',
|
|
'routing':{'destination':'inbox','reason':'AWAITING_TEST_FEEDBACK'}})
|
|
view=case_view(reader)
|
|
assert view['destination']=='inbox' and view['reason']=='AWAITING_TEST_FEEDBACK'
|
|
assert len(view['calls'])==2
|
|
usage=aggregate([view]);buckets={b['id']:b for b in usage['buckets']}
|
|
assert buckets['investigation']['input_tokens']==100
|
|
assert buckets['correction']['input_tokens']==100
|
|
assert sum(b['input_tokens']+b['output_tokens'] for b in buckets.values())==280
|
|
assert sum(b['reasoning_tokens'] for b in buckets.values())==60
|
|
|
|
|
|
def test_unknown_usage_is_not_reported_as_complete():
|
|
call={'bucket':'unknown','category':'unknown','phase':'diagnosis','day':'2026-10-21',
|
|
'input_tokens':None,'output_tokens':5,'reasoning_tokens':None,'cached_input_tokens':None}
|
|
result=aggregate([{'calls':[call]}])
|
|
assert result['unknown_usage_calls']==1
|
|
assert next(b for b in result['buckets'] if b['id']=='unknown')['output_tokens']==5
|
|
|
|
|
|
def test_pie_accounts_for_whole_pipeline_and_separates_corrections():
|
|
stages = [
|
|
('BUILDING_CASE', 'entity_resolution_plan_v4', 'building'),
|
|
('INTERPRETING_CASE', 'case_plan_v2', 'planning'),
|
|
('GENERATING_HYPOTHESES', 'hypotheses_v4', 'hypotheses'),
|
|
('INVESTIGATING_HYPOTHESES', 'investigation_lean_v21', 'investigation'),
|
|
('BUILDING_DIAGNOSIS', 'diagnosis_v8', 'diagnosis'),
|
|
('BUILDING_DIAGNOSIS', 'change_request_v9', 'coding'),
|
|
]
|
|
calls = []
|
|
for state, prompt, expected in stages:
|
|
assert phase({'phase': state}, prompt) == expected
|
|
calls.append({'bucket': expected, 'category':'regular', 'phase':expected,
|
|
'day':'2026-10-21', 'input_tokens':10, 'output_tokens':2,
|
|
'reasoning_tokens':1, 'cached_input_tokens':0})
|
|
calls.append(calls[3] | {'bucket':'correction','category':'correction'})
|
|
buckets = {b['id']:b for b in aggregate([{'calls':calls}])['buckets']}
|
|
assert all(buckets[expected]['calls']==1 for _, _, expected in stages)
|
|
assert buckets['correction']['calls']==1
|
|
assert sum(b['input_tokens']+b['output_tokens'] for b in buckets.values())==84
|
|
empty = {b['id']:b for b in aggregate([])['buckets']}
|
|
assert all(expected in empty for _, _, expected in stages)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def project_manifest(manifest):
|
|
manifest['project']={'id':'p','type':'wow','profile':'vcore'}
|
|
|
|
|
|
def test_overview_scope_assets_progress_and_completed_case_hidden(case_setup, monkeypatch):
|
|
import viano_bug_analysis.api.dashboard as module
|
|
store,builder,*_=case_setup
|
|
built=builder.build('case-A','stable-package')
|
|
token='dashboard_test_'+'x'*40;monkeypatch.setenv('DASHBOARD_TOKEN',token)
|
|
config=APIConfig(store.workspace.root,(Project('p','wow','vcore'),Project('other','wow','vcore')),
|
|
(Client('allowed','DASHBOARD_TOKEN',('p',)),))
|
|
app=create_app(config);client=TestClient(app);auth={'Authorization':'Bearer '+token}
|
|
assert client.get('/v1/overview').status_code==401
|
|
html=client.get('/')
|
|
assert html.status_code==200 and 'Content-Security-Policy' in html.headers
|
|
assert 'renderPie' in client.get('/assets/dashboard.js').text
|
|
assert client.get('/assets/missing.js').status_code==404
|
|
monkeypatch.setattr(module,'read_status',lambda _: {'observer':'LIVE','service_id':'test-service'})
|
|
directory=store.workspace.runtime/'worker';directory.mkdir(exist_ok=True)
|
|
atomic_json(directory/'status.json',{'status':'ANALYZING','service_id':'test-service','case':{'case_id':'case-A','package_id':'stable-package'}})
|
|
data=client.get('/v1/overview',headers=auth).json()
|
|
assert data['workers'][0]['case']['title']=='Generic problem'
|
|
assert data['workers'][0]['case']['progress']==29 # intake + completed casebuild / seven stages
|
|
assert data['cases'][0]['project_id']=='p'
|
|
assert token not in json.dumps(data) and str(store.workspace.root) not in json.dumps(data)
|
|
monkeypatch.setattr(module,'read_status',lambda _: {'observer':'LIVE','service_id':'unrelated-trial'})
|
|
assert client.get('/v1/overview',headers=auth).json()['workers']==[]
|
|
monkeypatch.setattr(module,'read_status',lambda _: {'observer':'LIVE','service_id':'test-service'})
|
|
stranger=Client('other','DASHBOARD_TOKEN',('other',))
|
|
hidden=module.Dashboard(app.state.intake).overview(stranger)
|
|
assert hidden['cases']==[] and hidden['workers']==[]
|
|
assert sum(b['calls'] for b in hidden['usage']['buckets'])==0
|
|
for state in ('INTERPRETING_CASE','GENERATING_HYPOTHESES','INVESTIGATING_HYPOTHESES','BUILDING_DIAGNOSIS','OUTPUT_READY'):
|
|
store.advance('case-A','stable-package',state,step_id='test:'+state)
|
|
completed=client.get('/v1/overview',headers=auth).json()
|
|
assert completed['workers']==[]
|
|
monkeypatch.setattr(module,'read_status',lambda _: {'observer':'HEARTBEAT_STALE'})
|
|
assert client.get('/v1/overview',headers=auth).json()['workers']==[]
|
|
def corrupted(_):
|
|
from viano_bug_analysis.telemetry.config import TelemetryError
|
|
raise TelemetryError('TELEMETRY_READ')
|
|
monkeypatch.setattr(module,'read_status',corrupted)
|
|
response=client.get('/v1/overview',headers=auth)
|
|
assert response.status_code==200 and response.json()['workers']==[]
|
|
|
|
|
|
def test_case_details_scope_download_and_revision(case_setup, monkeypatch):
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes
|
|
from io import BytesIO
|
|
from zipfile import ZipFile
|
|
from viano_bug_analysis.api.cases import case_key
|
|
store,builder,*_=case_setup
|
|
builder.build('case-A','stable-package')
|
|
monkeypatch.setenv('DETAIL_TOKEN','details_'+'x'*40)
|
|
config=APIConfig(store.workspace.root,(Project('p','wow','vcore'),Project('other','wow','vcore')),
|
|
(Client('allowed','DETAIL_TOKEN',('p',)),))
|
|
app=create_app(config);client=TestClient(app)
|
|
url='/v1/cases/'+case_key('case-A','stable-package')
|
|
auth={'Authorization':'Bearer details_'+'x'*40}
|
|
assert client.get(url).status_code==401
|
|
response=client.get(url,headers=auth)
|
|
assert response.status_code==200
|
|
data=response.json()
|
|
assert data['reports'][0]['title']=='Generic problem'
|
|
assert data['observations'][0]['text']=='The switch remains off'
|
|
assert data['diagnosis'] is None and data['coding']=={}
|
|
assert str(store.workspace.root) not in response.text
|
|
assert client.get(url+'?revision=999',headers=auth).status_code==404
|
|
assert client.get(url+'?revision=../1',headers=auth).status_code==422
|
|
download=client.get(url+'/package',headers=auth)
|
|
assert download.status_code==200 and download.headers['cache-control']=='no-store'
|
|
with ZipFile(BytesIO(download.content)) as archive:
|
|
assert archive.testzip() is None
|
|
from viano_bug_analysis.api.cases import Cases
|
|
with pytest.raises(Exception) as error:
|
|
Cases(app.state.intake).detail(case_key('case-A','stable-package'),Client('stranger','DETAIL_TOKEN',('other',)))
|
|
assert error.value.status==404
|
|
store.advance('case-A','stable-package','INTERPRETING_CASE',step_id='detail:interpret',
|
|
artifacts={'case_interpretation':'detail-interpretation.json'},
|
|
additions={'detail-interpretation.json':json_bytes({'analysis_revision':1,'interpretation':{'observed_behavior':'Old accepted interpretation'}})},
|
|
details={'operation':'case-interpreted','interpretation_ref':'detail-interpretation.json'})
|
|
for state in ('GENERATING_HYPOTHESES','INVESTIGATING_HYPOTHESES','BUILDING_DIAGNOSIS','OUTPUT_READY','HANDED_OFF','REOPENED'):
|
|
store.advance('case-A','stable-package',state,step_id='detail:'+state)
|
|
current=client.get(url,headers=auth).json()
|
|
historical=client.get(url+'?revision=1',headers=auth).json()
|
|
assert current['revision']==2 and current['interpretation']=={}
|
|
assert historical['revision']==1 and historical['state']=='HANDED_OFF'
|
|
assert historical['interpretation']['observed_behavior']=='Old accepted interpretation'
|
|
|
|
|
|
def test_historical_artifacts_are_not_replaced_by_latest_revision():
|
|
from viano_bug_analysis.api.cases import references
|
|
docs={'old':{'analysis_revision':1},'new':{'analysis_revision':2}}
|
|
reader=Reader(docs,{'case_id':'c','package_id':'p','artifacts':{'diagnosis':'new'}})
|
|
chain=[('newcp',{'analysis_revision':2,'details':{'operation':'diagnosis-complete','diagnosis_ref':'new'}}),
|
|
('oldcp',{'analysis_revision':1,'details':{'operation':'diagnosis-complete','diagnosis_ref':'old'}})]
|
|
assert references(reader,chain,1)['diagnosis']=='old'
|
|
assert references(reader,chain,2)['diagnosis']=='new'
|
|
|
|
|
|
def test_operator_permissions_scope_and_audit(case_setup,monkeypatch):
|
|
store,builder,*_=case_setup
|
|
builder.build('case-A','stable-package')
|
|
monkeypatch.setenv('CAPTURE_TOKEN','capture_'+'a'*40)
|
|
monkeypatch.setenv('OPERATOR_TOKEN','operator_'+'b'*40)
|
|
config=APIConfig(store.workspace.root,(Project('p','wow','vcore'),Project('other','wow','vcore')),
|
|
(Client('capture','CAPTURE_TOKEN',('p',)),Client('operator','OPERATOR_TOKEN',('p',),operator=True)))
|
|
app=create_app(config);client=TestClient(app)
|
|
capture={'Authorization':'Bearer capture_'+'a'*40}
|
|
operator={'Authorization':'Bearer operator_'+'b'*40}
|
|
assert client.get('/v1/operations').status_code==401
|
|
assert client.post('/v1/operations',headers=capture,json={'action':'pause','project_id':'p'}).status_code==403
|
|
assert client.post('/v1/operations',headers=operator,json={'action':'pause','project_id':'other'}).status_code==403
|
|
assert client.post('/v1/operations',headers=operator,json={'action':'pause','project_id':'p'}).status_code==200
|
|
view=client.get('/v1/operations',headers=capture).json()
|
|
assert view['projects']==[{'project_id':'p','paused':True}] and view['events']==[]
|
|
view=client.get('/v1/operations',headers=operator).json()
|
|
assert view['events'][-1]['actor']=='operator' and view['worker'] is None
|
|
assert client.post('/v1/operations',headers=operator,json={'action':'resume','project_id':'p','reset_budget':True}).status_code==422
|
|
assert client.post('/v1/operations',headers=operator,content='x'*4097).status_code==415
|
|
assert client.post('/v1/operations',headers=operator|{'Content-Type':'application/json'},content='x'*4097).status_code==413
|
|
from viano_bug_analysis.api.cases import case_key
|
|
from viano_bug_analysis.package import PackageReader
|
|
with PackageReader(store.latest('case-A','stable-package')) as reader:
|
|
generation=reader.manifest['package_generation']
|
|
command={'action':'retry','key':case_key('case-A','stable-package'),'package_generation':generation}
|
|
first=client.post('/v1/operations',headers=operator,json=command)
|
|
assert first.status_code==200 and first.json()['status']=='QUEUED'
|
|
assert client.post('/v1/operations',headers=operator,json=command).json()==first.json()
|
|
assert client.get('/v1/operations',headers=operator).json()['pending']==1
|
|
assert client.post('/v1/operations',headers=operator,json=command|{'package_generation':0}).status_code==409
|
|
store.advance('case-A','stable-package','BLOCKED',step_id='control:invalid',reason='INVALID_LLM_RESPONSES',
|
|
retry_not_before='2020-01-01T00:00:00Z')
|
|
with PackageReader(store.latest('case-A','stable-package')) as reader:
|
|
generation=reader.manifest['package_generation']
|
|
assert client.post('/v1/operations',headers=operator,json=command|{'package_generation':generation}).status_code==409
|