180 lines
10 KiB
Python
180 lines
10 KiB
Python
import json
|
|
import pytest
|
|
|
|
from viano_bug_analysis.tools.context_expansion import ContextExpansion
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
from viano_bug_analysis.tools.config import ToolError
|
|
from viano_bug_analysis.case import CaseBuilder
|
|
from viano_bug_analysis.package import PackageReader
|
|
from test_case_builder import case_setup
|
|
|
|
|
|
class RelatedData:
|
|
def __init__(self):self.calls=[]
|
|
def binding(self):return {"snapshot_id":"generic-fixture"}
|
|
def validate(self,query):pass
|
|
def query(self,query):
|
|
self.calls.append(query)
|
|
value=query['filters'][0]['value']
|
|
rows={('tasks',10):[{'id':10,'description':'Use the supplied capture device to find the missing objects.','device':22,'required':22,'unused':0}],
|
|
('devices',22):[{'id':22,'name':'Capture device','script':'capture_handler'}]}
|
|
return {'items':[{'table':query['table'],'row':r} for r in rows.get((query['table'],value),[])],
|
|
'coverage':'complete_filtered_rows','truncated':False}
|
|
|
|
|
|
def initial():
|
|
return {'kind':'database.rows','table':'tasks','columns':['id','description','device','required','unused'],
|
|
'filters':[{'column':'id','operator':'eq','value':10}],'offset':0,'limit':1}
|
|
|
|
|
|
def rule(column='device'):
|
|
return {'from_table':'tasks','from_id_column':'id','from_column':column,
|
|
'relation':'provided_device','entity_type':'device','table':'devices','column':'id','columns':['id','name','script']}
|
|
|
|
|
|
def test_relationship_traversal_keeps_text_and_deduplicates_queries():
|
|
db=RelatedData();service=ProjectResearch({},database=db)
|
|
additions,facts,context,entities=ContextExpansion(service,{'relationships':[rule(),rule('required'),rule('unused')]}).collect([initial()])
|
|
assert len(db.calls)==2 and len(additions)==2
|
|
assert 'Use the supplied capture device' in str(facts)
|
|
assert {e['id'] for e in entities}=={'22'}
|
|
assert all(e['roles']==['provided_device'] for e in entities)
|
|
edges=[f for f in facts if f['name']=='project_relationship']
|
|
assert {e['value']['from_column'] for e in edges}=={'device','required'}
|
|
for edge in edges:
|
|
origin=edge['origin'];capture=json.loads(additions[origin['source_ref']])
|
|
assert capture['result']['items'][0]['row'][edge['value']['from_column']]==22
|
|
assert all(e['value']['meaning']=='explicit_data_reference_not_causation' for e in edges)
|
|
|
|
|
|
def test_descriptive_context_is_presented_once_and_keeps_raw_proof():
|
|
service=ProjectResearch({},database=RelatedData())
|
|
additions,facts,_,_=ContextExpansion(service,{'texts':[{'table':'tasks','id_column':'id','context_role':'task_definition','columns':['description']}]}).collect([initial()])
|
|
description=next(f for f in facts if f['name']=='project_description')['value']
|
|
assert description['entity_id']==10 and description['context_role']=='task_definition'
|
|
assert 'not_verified_runtime_behavior' in description['meaning']
|
|
text=description['texts'][0]
|
|
capture=json.loads(additions[text['origin']['source_ref']])
|
|
assert text['text']==capture['result']['items'][0]['row']['description']
|
|
row=facts[0]['value']['result']['items'][0]['row']
|
|
assert row['description']=={'text_ref':text['origin']}
|
|
|
|
|
|
def test_cycle_and_operational_limits_are_explicit():
|
|
db=RelatedData();service=ProjectResearch({},database=db)
|
|
cycle={'from_table':'devices','from_id_column':'id','from_column':'id','relation':'self','table':'devices','column':'id','columns':['id','name','script']}
|
|
result=ContextExpansion(service,{'relationships':[rule(),cycle]}).collect([initial()])
|
|
assert len(db.calls)==2
|
|
db.calls.clear()
|
|
service=ProjectResearch({},database=db)
|
|
_,facts,_,_=ContextExpansion(service,{'relationships':[rule()],'max_queries':1}).collect([initial()])
|
|
assert len(db.calls)==1 and facts[-1]['value']['omitted_queries']==1
|
|
_,facts,_,_=ContextExpansion(service,{'relationships':[rule(),rule('required')],'max_relationships':1}).collect([initial()])
|
|
assert facts[-1]['value']['omitted_relationships']==1
|
|
|
|
|
|
def test_depth_limit_does_not_claim_complete_graph():
|
|
service=ProjectResearch({},database=RelatedData())
|
|
further={'from_table':'devices','from_id_column':'id','from_column':'id','relation':'device_detail','table':'other','column':'id','columns':['id']}
|
|
_,facts,_,_=ContextExpansion(service,{'relationships':[rule(),further],'max_depth':1}).collect([initial()])
|
|
assert facts[-1]['value']['depth_limited_relationships']==1
|
|
assert 'not_exhaustive' in facts[-1]['value']['meaning']
|
|
|
|
|
|
def test_missing_relationship_rows_remain_uninterpreted():
|
|
service=ProjectResearch({},database=RelatedData())
|
|
absent=rule();absent['table']='unknown_devices'
|
|
_,facts,_,entities=ContextExpansion(service,{'relationships':[absent]}).collect([initial()])
|
|
results=[f for f in facts if f['name']=='project_query']
|
|
assert results[-1]['value']['result']['items']==[]
|
|
assert entities[0]['id']=='22' # Referenced identity survives missing metadata.
|
|
assert 'root_cause' not in str(facts)
|
|
|
|
|
|
def test_script_search_uses_bound_symbol_not_numeric_grep():
|
|
db=RelatedData()
|
|
class Source:
|
|
def binding(self):return {'commit':'fixture'}
|
|
def validate(self,q):assert q['kind']=='source.symbol'
|
|
def query(self,q):return {'items':[{'path':'capture.cpp','line':3,'text':'class capture_handler {};'}],'coverage':'complete_selected_text_search','truncated':False}
|
|
service=ProjectResearch({},database=db,source=Source())
|
|
script={'from_table':'devices','from_column':'script','from_id_column':'id','relation':'device_script','kind':'source.symbol','path_prefix':'src/'}
|
|
_,facts,context,_=ContextExpansion(service,{'relationships':[rule(),script]}).collect([initial()])
|
|
assert len(context['source'])==1
|
|
assert facts[2]['value']['query']['text']=='capture_handler'
|
|
|
|
|
|
def test_casebuilder_links_entities_and_preserves_prior_inputs(case_setup):
|
|
store=case_setup[0]
|
|
before=store.latest('case-A','stable-package')
|
|
with PackageReader(before) as reader:project=reader.manifest['project']
|
|
service=ProjectResearch(project,database=RelatedData())
|
|
service.enrichment={'queries':[initial()],'relationships':[rule()]}
|
|
built=CaseBuilder(store,project_research=service).build('case-A','stable-package')
|
|
with PackageReader(before) as old,PackageReader(built.package) as reader:
|
|
case=json.loads(reader.read_bytes(built.case_ref))
|
|
assert any(e['type']=='device' and e['id']=='22' for e in case['entities'])
|
|
assert 'provided_device' in str(case['runtime_facts'])
|
|
for path in old.files:
|
|
if path!='manifest.json':assert old.read_bytes(path)==reader.read_bytes(path)
|
|
assert CaseBuilder(store,project_research=service).build('case-A','stable-package').reused
|
|
from viano_bug_analysis.lifecycle import State
|
|
for i,target in enumerate([State.INTERPRETING_CASE,State.GENERATING_HYPOTHESES,State.INVESTIGATING_HYPOTHESES,
|
|
State.BUILDING_DIAGNOSIS,State.OUTPUT_READY,State.HANDED_OFF,State.REOPENED]):
|
|
store.advance('case-A','stable-package',target,step_id=f'context-reopen-{i}')
|
|
reopened=CaseBuilder(store,project_research=service).build('case-A','stable-package')
|
|
assert reopened.analysis_revision==2
|
|
with PackageReader(reopened.package) as reader:
|
|
case=json.loads(reader.read_bytes(reopened.case_ref))
|
|
paths=[r['path'] for r in case['database_context']['refs']]
|
|
assert len(paths)==len(set(paths))
|
|
|
|
|
|
@pytest.mark.parametrize('field',['max_queries','max_depth','max_relationships'])
|
|
def test_invalid_limits_rejected(field):
|
|
with pytest.raises(ToolError):ContextExpansion(ProjectResearch({}),{field:True})
|
|
|
|
|
|
def test_typed_names_keep_exact_id_and_name_row_provenance():
|
|
from viano_bug_analysis.analysis.quality import bind_entity_names, unnamed_entity_reasons
|
|
service=ProjectResearch({},database=RelatedData())
|
|
settings={'relationships':[rule()], 'entities':[{'entity_type':'device','table':'devices',
|
|
'column':'id','name_column':'name','columns':['id','name','script']}]}
|
|
additions,facts,_,_=ContextExpansion(service,settings).collect([initial()])
|
|
named=next(f for f in facts if f['name']=='project_entity_name')
|
|
value=named['value']
|
|
assert (value['type'],value['id'],value['name'])==('device','22','Capture device')
|
|
capture=json.loads(additions[value['provenance']['source_ref']])
|
|
assert capture['result']['items'][0]['row']['id']==22
|
|
assert value['provenance']['id_pointer']=='/result/items/0/row/id'
|
|
assert value['provenance']['json_pointer']=='/result/items/0/row/name'
|
|
context={'sources':{'F0001':{'kind':'fact','data':named},
|
|
'N0001':{'kind':'entity','data':{'type':'device','id':'22'}},
|
|
'N0002':{'kind':'entity','data':{'type':'task','id':'22'}},
|
|
'N0003':{'kind':'entity','data':{'type':'device','id':'23'}}}}
|
|
bind_entity_names(context)
|
|
assert context['sources']['N0001']['data']['name_bindings'][0]['source_id']=='F0001'
|
|
assert set(unnamed_entity_reasons(context))=={'N0002','N0003'}
|
|
before=json.dumps(context,sort_keys=True)
|
|
bind_entity_names(context)
|
|
assert json.dumps(context,sort_keys=True)==before
|
|
|
|
|
|
def test_invalid_name_rule_is_a_controlled_configuration_error():
|
|
with pytest.raises(ToolError,match='PROJECT_ENRICHMENT_CONFIG'):
|
|
ContextExpansion(ProjectResearch({}),{'entities':[{'entity_type':'device','table':'devices',
|
|
'column':'id','name_column':'name','columns':['id']}]})
|
|
|
|
|
|
def test_example_project_identity_configuration_is_complete():
|
|
import tomllib
|
|
from pathlib import Path
|
|
from viano_bug_analysis.analysis.entity_resolution import policy
|
|
document=tomllib.loads((Path(__file__).parents[1]/'config/project.example.toml').read_text(encoding='utf-8'))
|
|
settings=policy(document['entity_resolution'])
|
|
for catalog in settings['catalogs']:
|
|
assert catalog['name_column'] in catalog['columns']
|
|
assert catalog['id_column'] in catalog['columns']
|
|
expansion=ContextExpansion(ProjectResearch({}),document['enrichment'])
|
|
assert {r['entity_type'] for r in expansion.name_rules}=={'creature','quest'}
|