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'}) view=case_view(reader) 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']==[]