272 lines
15 KiB
Python
272 lines
15 KiB
Python
import asyncio,json,time
|
|
from dataclasses import replace
|
|
from datetime import datetime,timedelta,timezone
|
|
import pytest
|
|
from viano_bug_analysis.telemetry import Telemetry,TelemetryConfig,TelemetryError,read_status,scan_workspace
|
|
from viano_bug_analysis.telemetry.__main__ import main,read_events
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore,State,InboxProcessor,LockBusy
|
|
from viano_bug_analysis.telemetry.decorators import observed
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import FixtureAdapter,gateway
|
|
from viano_bug_analysis.analysis import CaseInterpreter
|
|
|
|
@pytest.fixture
|
|
def session(tmp_path):
|
|
ws=Workspace(tmp_path/'work')
|
|
with Telemetry(ws,TelemetryConfig(heartbeat_seconds=1,heartbeat_stale_seconds=3,inventory_interval_seconds=2,inventory_stale_seconds=5)) as telemetry:
|
|
yield telemetry
|
|
|
|
def test_live_idle_clean_stop_and_readonly_commands(session,capsys):
|
|
ws=session.workspace
|
|
status=read_status(ws.root)
|
|
assert status['observer']=='LIVE' and status['work']=='IDLE'
|
|
assert main(['status','--workspace',str(ws.root)])==0
|
|
assert json.loads(capsys.readouterr().out)['observer']=='LIVE'
|
|
assert main(['metrics','--workspace',str(ws.root)])==0
|
|
assert 'vba_observer_alive 1' in capsys.readouterr().out
|
|
session.stop();status=read_status(ws.root);assert status['observer']=='STOPPED'
|
|
|
|
def test_heartbeat_independent_of_long_operation(session):
|
|
before=read_status(session.workspace.root)['heartbeat_at']
|
|
with session.operation('case_build','private-id','private-package'):
|
|
time.sleep(1.2)
|
|
status=read_status(session.workspace.root)
|
|
assert status['heartbeat_at']!=before and status['work']=='WORKING'
|
|
assert status['active_operations'][0]['case_key']!='private-id'
|
|
|
|
def test_error_chain_suppressed_context_and_no_sensitive_values(session):
|
|
from viano_bug_analysis.case.errors import CaseBuildError
|
|
secret='sk-private-value-123'
|
|
with pytest.raises(CaseBuildError):
|
|
with session.operation('case_build',secret,secret):
|
|
try:raise RuntimeError(secret)
|
|
except RuntimeError:raise CaseBuildError('CASE_NOT_FOUND',secret) from None
|
|
records=read_events(session.workspace.root,errors=True)
|
|
assert records[-1]['code']=='CASE_NOT_FOUND'
|
|
exceptions=records[-1]['detail']['exceptions']
|
|
assert len(exceptions)==2 and exceptions[1]['type']=='RuntimeError'
|
|
assert secret not in (session.directory/'service.jsonl').read_text()
|
|
assert secret not in (session.directory/'status.json').read_text()
|
|
assert secret not in (session.directory/'metrics.prom').read_text()
|
|
assert read_status(session.workspace.root)['health']=='DEGRADED'
|
|
|
|
def test_fail_open_on_sink_failure(session,monkeypatch,capsys):
|
|
original=session._write
|
|
monkeypatch.setattr(session,'_write',lambda *a:(_ for _ in ()).throw(OSError('secret filesystem details')))
|
|
with session.operation('case_build'):pass
|
|
assert session.snapshot()['telemetry_failures']>0
|
|
assert 'secret filesystem details' not in capsys.readouterr().err
|
|
monkeypatch.setattr(session,'_write',original);session.flush()
|
|
assert session.snapshot()['telemetry_healthy']
|
|
|
|
def test_passive_inventory_does_not_resume_or_change_packages(session,make_package):
|
|
store=CheckpointStore(session.workspace,telemetry=session)
|
|
status,path=store.intake(make_package());original=path.read_bytes()
|
|
scan=scan_workspace(session.workspace.root)
|
|
assert scan['counts']['ready']==1 and path.read_bytes()==original
|
|
store.advance('case-A','stable-package',State.BUILDING_CASE,step_id='build')
|
|
blocked=store.advance('case-A','stable-package',State.BLOCKED,step_id='blocked',reason='LLM_UNAVAILABLE',retry_not_before=(datetime.now(timezone.utc)+timedelta(hours=6)).isoformat(),details={'automatic_retry':False})
|
|
before=blocked.read_bytes();session.refresh_inventory(force=True)
|
|
status=read_status(session.workspace.root)
|
|
assert status['work']=='WAITING_EXTERNAL' and status['inventory']['counts']['manual_wait']==1
|
|
assert blocked.read_bytes()==before and store.latest('case-A','stable-package')==blocked
|
|
|
|
def test_schema_heartbeat_stale_and_queue_unknown(session):
|
|
status=read_status(session.workspace.root,now=datetime.now(timezone.utc)+timedelta(seconds=10))
|
|
assert status['observer']=='HEARTBEAT_STALE' and status['work']=='QUEUE_UNKNOWN'
|
|
|
|
def test_rotation_and_read_event_filter(tmp_path):
|
|
ws=Workspace(tmp_path/'work')
|
|
with Telemetry(ws,TelemetryConfig(log_max_bytes=1500,log_backups=2)) as t:
|
|
for _ in range(15):
|
|
with t.operation('case_build','case','pack'):pass
|
|
assert (t.directory/'service.jsonl.1').exists()
|
|
assert len(list(t.directory.glob('service.jsonl*')))<=3
|
|
assert read_events(ws.root,case_id='case',max_bytes=1500,backups=2)
|
|
assert not read_events(ws.root,case_id='other',max_bytes=1500,backups=2)
|
|
|
|
def test_single_writer_lease(session):
|
|
with pytest.raises(LockBusy):Telemetry(session.workspace).start()
|
|
|
|
|
|
def test_checkpoint_lookup_duration_is_visible_without_case_content(session):
|
|
store=CheckpointStore(session.workspace,telemetry=session)
|
|
assert store.latest('private-case','private-package') is None
|
|
assert 'vba_operation_duration_seconds_count{operation="checkpoint_lookup"} 1' in session.prometheus()
|
|
events=read_events(session.workspace.root)
|
|
record=next(r for r in events if r['event']=='operation_finished' and r['operation']=='checkpoint_lookup')
|
|
assert record['detail']['duration_seconds']>=0
|
|
assert 'private-case' not in json.dumps(record) and 'private-package' not in json.dumps(record)
|
|
|
|
@pytest.mark.parametrize('value',[0,-1,True])
|
|
def test_config_rejects_bad_defaults(value):
|
|
with pytest.raises(TelemetryError):TelemetryConfig(heartbeat_seconds=value)
|
|
|
|
def test_instrumented_interpreter_and_llm_usage_and_replay(case_setup):
|
|
store,builder,*_=case_setup
|
|
with Telemetry(store.workspace) as t:
|
|
store.telemetry=t
|
|
builder.build('case-A','stable-package')
|
|
interpreter=CaseInterpreter(store,gateway(FixtureAdapter()))
|
|
result=asyncio.run(interpreter.interpret('case-A','stable-package'))
|
|
with open(result.package,'rb') as f:before=f.read()
|
|
assert asyncio.run(interpreter.interpret('case-A','stable-package')).reused
|
|
assert result.package.read_bytes()==before
|
|
data=t.prometheus()
|
|
assert 'vba_llm_rounds_total{phase="INTERPRETING_CASE",status="SUCCESS"} 1' in data
|
|
assert 'vba_llm_tokens_total{direction="input_tokens"} 100' in data
|
|
assert 'vba_llm_usage_unknown_total{field="estimated_cost"} 1' in data
|
|
events=read_events(store.workspace.root,limit=1000)
|
|
assert any(r['event']=='case_checkpoint' and r['case_key'] for r in events)
|
|
assert not (store.workspace.runtime/'logs'/'llm.jsonl').exists()
|
|
|
|
def test_inbox_error_is_visible_without_free_message(session,monkeypatch,tmp_path):
|
|
queue=InboxProcessor(session.workspace,telemetry=session)
|
|
source=session.workspace.inbox/'one.zip';source.write_bytes(b'x')
|
|
def fail(*a):raise OSError('api_key=SECRET')
|
|
monkeypatch.setattr(queue,'_snapshot',fail)
|
|
result=queue.process(source);assert result.status=='ERROR'
|
|
records=read_events(session.workspace.root,errors=True)
|
|
assert records and 'SECRET' not in json.dumps(records)
|
|
|
|
def test_full_pipeline_observed_without_telemetry_artifacts_or_double_accounting(tool_case):
|
|
from viano_bug_analysis.analysis import Investigator,DiagnosisBuilder,ChangeRequestBuilder,OutputValidator
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from test_investigator import LoopAdapter,CASE
|
|
from test_diagnosis_builder import FinalAdapter
|
|
from test_change_request_builder import RequestAdapter
|
|
store=tool_case[0]
|
|
with Telemetry(store.workspace) as t:
|
|
store.telemetry=t
|
|
async def process():
|
|
await Investigator(store,gateway(LoopAdapter())).run(*CASE,max_steps=20)
|
|
await DiagnosisBuilder(store,gateway(FinalAdapter(True))).build(*CASE)
|
|
g=gateway(RequestAdapter())
|
|
await ChangeRequestBuilder(store,g).build(*CASE)
|
|
return await OutputValidator(store,g).publish(*CASE)
|
|
result=asyncio.run(process())
|
|
assert result.status=='OUTPUT_READY'
|
|
from viano_bug_analysis.package import PackageReader
|
|
with PackageReader(result.package) as r:
|
|
assert not any(name.startswith('telemetry/') or name.endswith('service.jsonl') for name in r.files)
|
|
data=t.prometheus()
|
|
assert 'vba_evidence_records_published_total 3' in data
|
|
assert 'vba_llm_tokens_total{direction="input_tokens"} 800' in data
|
|
snapshot=t.snapshot()
|
|
assert snapshot['budgets'][0]['spent']==30
|
|
assert snapshot['budgets'][0]['remaining']==BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['remaining']+180
|
|
assert not snapshot['active_operations']
|
|
|
|
def test_failover_error_no_legacy_provider_identifier_output(case_setup,caplog):
|
|
from viano_bug_analysis.llm import ProviderUnavailable
|
|
from viano_bug_analysis.llm.runner import CaseLLMRunner
|
|
store,builder,*_=case_setup
|
|
with Telemetry(store.workspace) as t:
|
|
store.telemetry=t;builder.build('case-A','stable-package')
|
|
secret='provider-private-secret-value'
|
|
adapter=FixtureAdapter(ProviderUnavailable('LLM_UNAVAILABLE'))
|
|
g=gateway(adapter)
|
|
g.config=replace(g.config,providers=(replace(g.config.providers[0],id=secret),),max_attempts=1)
|
|
result=asyncio.run(CaseInterpreter(store,g).interpret('case-A','stable-package'))
|
|
assert result.status=='BLOCKED'
|
|
assert secret not in (t.directory/'service.jsonl').read_text()
|
|
assert secret not in caplog.text
|
|
assert 'vba_llm_provider_failures_total{phase="INTERPRETING_CASE"} 1' in t.prometheus()
|
|
|
|
def test_explicit_payload_fields_are_dropped_and_unknown_code_is_masked(session):
|
|
session.emit('phase_diagnostic',level='ERROR',code='CUSTOM_SECRET_CODE',extra={'prompt':'SECRET','password':'SECRET','message':'SECRET'})
|
|
records=read_events(session.workspace.root,errors=True)
|
|
assert records[-1]['code']=='UNCLASSIFIED_ERROR'
|
|
assert 'SECRET' not in (session.directory/'service.jsonl').read_text()
|
|
|
|
def test_active_status_and_metrics_do_not_expose_case_labels(session):
|
|
with session.operation('case_build','CASE_SECRET','PACKAGE_SECRET'):
|
|
data=session.prometheus()
|
|
assert 'CASE_SECRET' not in data and 'PACKAGE_SECRET' not in data
|
|
assert not read_events(session.workspace.root,case_id='other')
|
|
|
|
def test_inventory_limit_and_corrupt_package_are_explicit(session,make_package):
|
|
store=CheckpointStore(session.workspace);_,path=store.intake(make_package())
|
|
path.write_bytes(b'broken')
|
|
scan=scan_workspace(session.workspace.root)
|
|
assert not scan['complete'] and scan['errors']
|
|
|
|
def test_cancelled_async_operation_has_no_stale_active_phase(session):
|
|
class App:
|
|
store=type('Store',(),{'telemetry':session})()
|
|
@observed('case_build')
|
|
async def build(self,case_id,package_id):raise asyncio.CancelledError()
|
|
with pytest.raises(asyncio.CancelledError):asyncio.run(App().build('case','package'))
|
|
assert not session.snapshot()['active_operations']
|
|
assert 'status="CANCELLED"' in session.prometheus()
|
|
|
|
def test_noop_without_running_session(tmp_path):
|
|
workspace=Workspace(tmp_path/'work');t=Telemetry(workspace)
|
|
class App:
|
|
store=type('Store',(),{'telemetry':t})()
|
|
@observed('case_build')
|
|
def build(self,case_id,package_id):return 42
|
|
assert App().build('case','pack')==42 and not t.directory.exists()
|
|
from test_investigation_tools import tool_case
|
|
|
|
def test_unfinished_log_record_survives_restart(tmp_path):
|
|
ws=Workspace(tmp_path/'work')
|
|
with Telemetry(ws):pass
|
|
path=ws.runtime/'telemetry'/'service.jsonl'
|
|
with path.open('ab') as stream:stream.write(b'{"event":')
|
|
with Telemetry(ws):pass
|
|
rows=read_events(ws.root)
|
|
assert any(r['event']=='observer_started' for r in rows)
|
|
assert rows[-1]['code']=='TELEMETRY_LOG_CORRUPT'
|
|
|
|
def test_log_channel_failure_does_not_claim_healthy_snapshots(session,monkeypatch):
|
|
original=session._handler.emit
|
|
def fail(*args):raise OSError('SECRET')
|
|
monkeypatch.setattr(session._handler,'emit',fail)
|
|
session.emit('phase_diagnostic');session.flush()
|
|
assert not read_status(session.workspace.root)['telemetry_healthy']
|
|
monkeypatch.setattr(session._handler,'emit',original)
|
|
session.emit('phase_diagnostic');session.flush()
|
|
assert read_status(session.workspace.root)['telemetry_healthy']
|
|
|
|
def test_per_case_stall_not_hidden_by_other_case_progress(tmp_path):
|
|
ws=Workspace(tmp_path/'work');now=[datetime.now(timezone.utc)]
|
|
with Telemetry(ws,TelemetryConfig(progress_warning_seconds=10),clock=lambda:now[0]) as t:
|
|
with t.operation('case_build','A','PA'):
|
|
now[0]+=timedelta(seconds=20)
|
|
t.case_checkpoint({'case_id':'B','package_id':'PB','state':'VALIDATED','package_generation':1,'analysis_revision':1,'routing':{'reason':None,'retry_not_before':None}})
|
|
status=t.snapshot()
|
|
assert status['work']=='NO_RECENT_PROGRESS' and status['active_operations'][0]['stalled']
|
|
|
|
def test_checkpoint_updates_ready_queue_without_rescan(session,make_package):
|
|
store=CheckpointStore(session.workspace,telemetry=session)
|
|
store.intake(make_package())
|
|
assert session.snapshot()['work']=='READY_NOT_PROCESSING'
|
|
|
|
def test_quarantine_event_links_local_diagnostic_without_source_name(session):
|
|
queue=InboxProcessor(session.workspace,telemetry=session)
|
|
source=session.workspace.inbox/'SECRET-source.bin';source.write_bytes(b'invalid ZIP')
|
|
result=queue.process(source);assert result.status=='QUARANTINED'
|
|
rows=read_events(session.workspace.root,source_name='SECRET-source.bin')
|
|
assert rows[-1]['detail']['quarantine_key']==result.package.parent.name
|
|
assert 'SECRET-source.bin' not in (session.directory/'service.jsonl').read_text()
|
|
|
|
def test_config_file_and_not_started_status(tmp_path):
|
|
config=tmp_path/'telemetry.toml';config.write_text('[telemetry]\nheartbeat_seconds=2\nheartbeat_stale_seconds=10\n')
|
|
assert TelemetryConfig.load(config).heartbeat_seconds==2
|
|
config.write_text('[telemetry]\nunknown=true\n')
|
|
with pytest.raises(TelemetryError):TelemetryConfig.load(config)
|
|
assert read_status(tmp_path)['observer']=='NOT_STARTED'
|
|
|
|
def test_bad_result_introspection_does_not_change_method_result(session):
|
|
class Result:
|
|
@property
|
|
def status(self):raise RuntimeError('SECRET')
|
|
result=Result()
|
|
class App:
|
|
store=type('Store',(),{'telemetry':session})()
|
|
@observed('case_build')
|
|
def build(self,*a):return result
|
|
assert App().build('A','B') is result
|
|
assert session.snapshot()['telemetry_failures']>0
|