322 lines
14 KiB
Python
322 lines
14 KiB
Python
import asyncio
|
|
import json
|
|
from dataclasses import replace
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.api import APIConfig, Client, Project
|
|
from viano_bug_analysis.lifecycle import State
|
|
from viano_bug_analysis.lifecycle.locking import FileLock
|
|
from viano_bug_analysis.lifecycle.errors import LockBusy
|
|
from viano_bug_analysis.package import PackageReader
|
|
from viano_bug_analysis.worker import AnalysisWorker, WorkerConfig, WorkerError
|
|
from viano_bug_analysis.worker.config import binding
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import gateway, FixtureAdapter
|
|
|
|
CASE = ('case-A', 'stable-package')
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def project_manifest(manifest):
|
|
manifest['project'] = {'id': 'test', 'type': 'wow', 'profile': 'vcore'}
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(case_setup):
|
|
store = case_setup[0]
|
|
project = Project('test', 'wow', 'vcore', planning_mode='separate')
|
|
config = WorkerConfig(APIConfig(store.workspace.root, (project,),
|
|
(Client('capture', 'TEST_TOKEN', ('test',)),)), min_free_bytes=1)
|
|
def loader(p):
|
|
return gateway(FixtureAdapter()), {'planning_mode': 'separate'}, binding(p)
|
|
return config, loader, store
|
|
|
|
|
|
def test_real_case_build_then_interpret_and_restart(setup):
|
|
config, loader, store = setup
|
|
worker = AnalysisWorker(config, loader=loader)
|
|
assert asyncio.run(worker.tick()) == 1
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
assert reader.manifest['state'] == State.BUILDING_CASE
|
|
assert reader.manifest['artifacts']['case']
|
|
# Fresh worker object has no in-memory queue. It recovers from managed packages.
|
|
resumed = AnalysisWorker(config, loader=loader)
|
|
assert asyncio.run(resumed.tick()) == 1
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
assert reader.manifest['state'] == State.INTERPRETING_CASE
|
|
assert reader.manifest['artifacts']['case_interpretation']
|
|
|
|
|
|
def test_scheduler_retry_is_durable_and_not_busy_loop(setup):
|
|
config, loader, store = setup
|
|
now = datetime.now(timezone.utc)
|
|
calls = []
|
|
async def dispatch(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
return {'status': 'RETRY_PENDING', 'code': 'LLM_CONNECTION',
|
|
'retry_not_before': (now + timedelta(minutes=5)).isoformat()}
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch, clock=lambda: now)
|
|
assert asyncio.run(worker.tick()) == 1
|
|
again = AnalysisWorker(config, loader=loader, dispatcher=dispatch, clock=lambda: now)
|
|
assert asyncio.run(again.tick()) == 0
|
|
assert len(calls) == 1 and calls[0]['attempt_id'] == 'default'
|
|
after = AnalysisWorker(config, loader=loader, dispatcher=dispatch,
|
|
clock=lambda: now + timedelta(minutes=6))
|
|
assert asyncio.run(after.tick()) == 1
|
|
|
|
|
|
@pytest.mark.parametrize('reason,automatic', [
|
|
('AWAITING_TEST_FEEDBACK', True), ('INVALID_LLM_RESPONSES', True),
|
|
('LLM_UNAVAILABLE', True), ('RESEARCH_UNAVAILABLE', False)])
|
|
def test_worker_never_silently_bypasses_block(setup, reason, automatic):
|
|
config, loader, store = setup
|
|
store.advance(*CASE, State.BUILDING_CASE, step_id='start')
|
|
blocked = store.advance(*CASE, State.BLOCKED, step_id='block', reason=reason,
|
|
retry_not_before='2020-01-01T00:00:00Z', details={'automatic_retry': automatic})
|
|
async def forbidden(*args, **kwargs):
|
|
raise AssertionError('blocked case was dispatched')
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=forbidden)
|
|
assert asyncio.run(worker.tick()) == 0
|
|
assert store.latest(*CASE) == blocked
|
|
if reason != 'RESEARCH_UNAVAILABLE':
|
|
with pytest.raises(WorkerError, match='WORKER_RETRY_STATE'):
|
|
worker.retry(*CASE)
|
|
else:
|
|
worker.retry(*CASE)
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
assert reader.manifest['state'] == 'BUILDING_CASE'
|
|
|
|
|
|
def test_explicitly_permitted_due_retry_resumes(setup):
|
|
config, loader, store = setup
|
|
store.advance(*CASE, State.BUILDING_CASE, step_id='start')
|
|
store.advance(*CASE, State.BLOCKED, step_id='block', reason='RESEARCH_UNAVAILABLE',
|
|
retry_not_before='2020-01-01T00:00:00Z', details={'automatic_retry': True})
|
|
seen = []
|
|
async def dispatch(store, gateway, *identity, **kwargs):
|
|
with PackageReader(store.latest(*identity)) as reader:
|
|
seen.append(reader.manifest['state'])
|
|
return {'status': 'YIELDED'}
|
|
assert asyncio.run(AnalysisWorker(config, loader=loader, dispatcher=dispatch).tick()) == 1
|
|
assert seen == ['BUILDING_CASE']
|
|
|
|
|
|
def test_one_service_and_trial_lease_exclusion(setup):
|
|
config, loader, store = setup
|
|
worker = AnalysisWorker(config, loader=loader)
|
|
with FileLock(worker.workspace.runtime / 'locks/analysis-service.lock'):
|
|
with pytest.raises(LockBusy):
|
|
asyncio.run(worker.run(once=True))
|
|
from viano_bug_analysis.lifecycle.storage import key
|
|
with FileLock(worker.workspace.runtime / 'locks' / ('trial-' + key(CASE[0]) + '.lock')):
|
|
assert asyncio.run(worker.tick()) == 0
|
|
|
|
|
|
def test_exception_backoff_and_explicit_scheduler_release(setup):
|
|
config, loader, store = setup
|
|
calls = []
|
|
async def broken(*args, **kwargs):
|
|
calls.append(1)
|
|
raise ValueError('secret data must not appear in runtime state')
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=broken)
|
|
asyncio.run(worker.tick())
|
|
asyncio.run(worker.tick())
|
|
assert len(calls) == 1
|
|
record = next(worker.records.glob('*.json')).read_text()
|
|
assert 'secret data' not in record
|
|
worker.retry(*CASE)
|
|
asyncio.run(worker.tick())
|
|
assert len(calls) == 2
|
|
|
|
|
|
def test_stop_requested_during_dispatch_finishes_step(setup):
|
|
config, loader, store = setup
|
|
async def dispatch(*args, **kwargs):
|
|
worker.request_stop()
|
|
return {'status': 'YIELDED'}
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
|
|
asyncio.run(worker.run())
|
|
assert json.loads((worker.root / 'status.json').read_bytes())['status'] == 'STOPPED'
|
|
assert next(worker.records.glob('*.json')).exists()
|
|
|
|
|
|
def test_config_binding_persisted_before_dispatch_and_detects_restart_change(setup):
|
|
config, loader, store = setup
|
|
async def dispatch(*args, **kwargs):
|
|
assert next(worker.records.glob('*.json')).exists()
|
|
return {'status': 'YIELDED'}
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
|
|
asyncio.run(worker.tick())
|
|
changed = replace(config.intake.projects[0], planning_mode='joint')
|
|
updated = replace(config, intake=replace(config.intake, projects=(changed,)))
|
|
other = AnalysisWorker(updated, loader=loader, dispatcher=dispatch)
|
|
assert asyncio.run(other.tick()) == 0
|
|
record = json.loads(next(other.records.glob('*.json')).read_bytes())
|
|
assert record['status'] == 'ATTENTION' and record['code'] == 'WORKER_CONFIG_CHANGED'
|
|
|
|
|
|
def test_unknown_project_held_without_model_call(setup):
|
|
config, loader, store = setup
|
|
changed = replace(config.intake.projects[0], id='other')
|
|
updated = replace(config, intake=APIConfig(config.intake.workspace, (changed,),
|
|
(Client('capture', 'TEST_TOKEN', ('other',)),)))
|
|
def forbidden(project):
|
|
raise AssertionError('unknown project')
|
|
worker = AnalysisWorker(updated, loader=forbidden)
|
|
assert asyncio.run(worker.tick()) == 0
|
|
assert json.loads(next(worker.records.glob('*.json')).read_bytes())['code'] == 'WORKER_PROJECT_UNKNOWN'
|
|
|
|
|
|
def test_terminal_result_not_republished_every_pass(setup):
|
|
config, loader, store = setup
|
|
calls = []
|
|
async def dispatch(*args, **kwargs):
|
|
calls.append(1)
|
|
return {'status': 'OUTPUT_READY'}
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
|
|
asyncio.run(worker.tick())
|
|
asyncio.run(worker.tick())
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_reopened_revision_does_not_skip_new_interpretation(setup):
|
|
config, loader, store = setup
|
|
worker = AnalysisWorker(config, loader=loader)
|
|
asyncio.run(worker.tick())
|
|
asyncio.run(worker.tick())
|
|
for state in ('GENERATING_HYPOTHESES', 'INVESTIGATING_HYPOTHESES', 'BUILDING_DIAGNOSIS', 'OUTPUT_READY', 'HANDED_OFF'):
|
|
store.advance(*CASE, state, step_id='test:' + state)
|
|
store.advance(*CASE, 'REOPENED', step_id='test:reopen')
|
|
asyncio.run(worker.tick())
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
assert reader.manifest['analysis_revision'] == 2
|
|
assert reader.manifest['state'] == 'BUILDING_CASE'
|
|
# Old interpretation is deliberately retained in the append-only package.
|
|
old = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_interpretation']))
|
|
assert old['analysis_revision'] == 1
|
|
asyncio.run(worker.tick())
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
doc = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_interpretation']))
|
|
assert doc['analysis_revision'] == 2
|
|
|
|
|
|
def test_multiple_projects_and_poison_case_do_not_starve_other_cases(setup, case_setup, tmp_path):
|
|
from viano_bug_analysis.package.factory import PackageFactory
|
|
config, loader, store = setup
|
|
with PackageReader(case_setup[2]) as reader:
|
|
manifest = reader.manifest
|
|
contents = {ref: reader.read_bytes(ref) for ref in reader.files if ref != 'manifest.json'}
|
|
manifest.update(case_id='case-B', package_id='other-package', project={'id': 'second', 'type': 'wow', 'profile': 'vcore'})
|
|
PackageFactory().create(store.workspace.inbox / 'other.zip', manifest, contents)
|
|
project2 = Project('second', 'wow', 'vcore', planning_mode='separate')
|
|
intake = replace(config.intake, projects=config.intake.projects + (project2,))
|
|
seen = []
|
|
def load(project):
|
|
return project.id, {'project_id': project.id}, binding(project)
|
|
async def dispatch(store, gateway, case_id, package_id, **kwargs):
|
|
seen.append((case_id, gateway, kwargs['policies']['project_id']))
|
|
if case_id == 'case-A':
|
|
raise ValueError('broken first case')
|
|
return {'status': 'YIELDED'}
|
|
worker = AnalysisWorker(replace(config, intake=intake), loader=load, dispatcher=dispatch)
|
|
asyncio.run(worker.tick())
|
|
assert seen == [('case-A', 'test', 'test'), ('case-B', 'second', 'second')]
|
|
asyncio.run(worker.tick())
|
|
assert [v[0] for v in seen].count('case-A') == 1
|
|
assert [v[0] for v in seen].count('case-B') == 2
|
|
|
|
|
|
def test_cli_check_status_and_empty_once_do_not_need_api_token(tmp_path, monkeypatch, capsys):
|
|
from viano_bug_analysis.worker.__main__ import main
|
|
import viano_bug_analysis.worker.__main__ as cli
|
|
monkeypatch.delenv('ABSENT_API_TOKEN', raising=False)
|
|
intake = tmp_path / 'api.toml'
|
|
intake.write_text('''[api]
|
|
workspace = "service"
|
|
[[projects]]
|
|
id = "test"
|
|
type = "wow"
|
|
profile = "vcore"
|
|
[[clients]]
|
|
id = "capture"
|
|
token_env = "ABSENT_API_TOKEN"
|
|
projects = ["test"]
|
|
''')
|
|
config = tmp_path / 'worker.toml'
|
|
config.write_text('[worker]\nintake_config = "api.toml"\nmin_free_bytes = 1\n')
|
|
loaded = []
|
|
monkeypatch.setattr(cli, 'load_project', lambda p: loaded.append(p.id))
|
|
assert main(['check', '--config', str(config)]) == 0
|
|
assert loaded == ['test']
|
|
assert main(['run', '--once', '--config', str(config)]) == 0
|
|
assert main(['status', '--config', str(config)]) == 0
|
|
assert 'STOPPED' in capsys.readouterr().out
|
|
|
|
|
|
def test_storage_floor_pauses_without_dispatch(setup, monkeypatch):
|
|
import viano_bug_analysis.worker.service as module
|
|
from collections import namedtuple
|
|
config, loader, store = setup
|
|
Usage = namedtuple('Usage', 'total used free')
|
|
worker = AnalysisWorker(config, loader=loader)
|
|
monkeypatch.setattr(module.shutil, 'disk_usage', lambda *_: Usage(100, 100, 0))
|
|
assert asyncio.run(worker.tick()) == 0
|
|
assert json.loads((worker.root / 'status.json').read_bytes())['status'] == 'STORAGE_LOW'
|
|
|
|
|
|
def test_exception_after_checkpoint_holds_actual_latest_generation(setup):
|
|
config, loader, store = setup
|
|
calls = []
|
|
async def dispatch(store, gateway, *identity, **kwargs):
|
|
calls.append(1)
|
|
store.advance(*identity, 'BUILDING_CASE', step_id='commit-before-error')
|
|
raise ValueError('unexpected after checkpoint')
|
|
worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
|
|
asyncio.run(worker.tick())
|
|
asyncio.run(worker.tick())
|
|
assert len(calls) == 1
|
|
record = json.loads(next(worker.records.glob('*.json')).read_bytes())
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
assert record['package_generation'] == reader.manifest['package_generation']
|
|
assert record['status'] == 'ATTENTION'
|
|
|
|
|
|
def test_operator_pause_survives_restart_and_drains_started_cases(setup):
|
|
config,loader,store=setup
|
|
worker=AnalysisWorker(config,loader=loader)
|
|
worker.control.pause('operator','test',True)
|
|
assert asyncio.run(worker.tick())==0
|
|
worker=AnalysisWorker(config,loader=loader)
|
|
assert asyncio.run(worker.tick())==0
|
|
worker.control.pause('operator','test',False)
|
|
assert asyncio.run(worker.tick())==1
|
|
worker.control.pause('operator','test',True)
|
|
assert asyncio.run(worker.tick())==1 # already started case continues
|
|
|
|
|
|
def test_operator_commands_recheck_generation_and_forbid_validation_reset(setup):
|
|
config,loader,store=setup
|
|
worker=AnalysisWorker(config,loader=loader)
|
|
store.advance(*CASE,State.BUILDING_CASE,step_id='operator:start')
|
|
store.advance(*CASE,State.BLOCKED,step_id='operator:block',reason='RESEARCH_UNAVAILABLE',
|
|
retry_not_before='2020-01-01T00:00:00Z',details={'automatic_retry':False})
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
manifest=reader.manifest
|
|
worker.control.enqueue('operator','test',manifest)
|
|
worker._commands()
|
|
assert not worker.control.read()['commands']
|
|
assert worker.control.read()['events'][-1]['status']=='APPLIED'
|
|
worker.control.enqueue('operator','test',manifest) # stale UI cannot release new head
|
|
worker._commands()
|
|
assert worker.control.read()['events'][-1]['status']=='REJECTED'
|
|
store.advance(*CASE,State.BLOCKED,step_id='operator:invalid',reason='INVALID_LLM_RESPONSES',
|
|
retry_not_before='2020-01-01T00:00:00Z')
|
|
path=store.latest(*CASE)
|
|
with PackageReader(path) as reader:
|
|
worker.control.enqueue('operator','test',reader.manifest)
|
|
worker._commands()
|
|
assert store.latest(*CASE)==path
|
|
assert worker.control.read()['events'][-1]['status']=='REJECTED'
|