265 lines
14 KiB
Python
265 lines
14 KiB
Python
"""Real SQL selection logic, simulated metadata/transaction transport; no LLM."""
|
|
import copy
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
import tomllib
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from viano_bug_analysis.tools import ToolError
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch, ProjectLimits
|
|
from viano_bug_analysis.tools.sql_database import SQLDatabaseProvider
|
|
from viano_bug_analysis.tools.database_relations import DatabaseRelations
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
|
|
|
|
PROFILE = {'inspect': {'description': 'Inspect parent records and equal-key children.', 'nodes': [
|
|
{'name': 'root', 'table': 'templates', 'match_column': 'entry', 'columns': ['entry', 'name']},
|
|
{'name': 'spawns', 'table': 'spawns', 'match_column': 'entry', 'parent': 'root',
|
|
'parent_column': 'entry', 'columns': ['guid', 'entry', 'zone', 'mask'], 'group_by': ['zone', 'mask']},
|
|
{'name': 'links', 'table': 'links', 'match_column': 'guid', 'parent': 'spawns',
|
|
'parent_column': 'guid', 'columns': ['guid', 'event']}]}}
|
|
|
|
|
|
class Database:
|
|
def __init__(self):
|
|
self.db = sqlite3.connect(':memory:')
|
|
self.db.row_factory = sqlite3.Row
|
|
self.db.executescript('CREATE TABLE templates(entry INTEGER PRIMARY KEY, name TEXT);'
|
|
'CREATE TABLE spawns(guid INTEGER PRIMARY KEY, entry INTEGER, zone INTEGER, mask INTEGER);'
|
|
'CREATE TABLE links(guid INTEGER, event INTEGER);'
|
|
"INSERT INTO templates VALUES (42,'subject');"
|
|
'INSERT INTO spawns VALUES (1,42,14,1),(2,42,14,1),(3,42,14,1);'
|
|
'INSERT INTO links VALUES (1,24),(2,24),(3,24);')
|
|
self.calls = []; self.connections = self.closed = self.rollbacks = 0
|
|
self.engine = 'InnoDB'; self.fail = False
|
|
|
|
def connect(self):
|
|
self.connections += 1
|
|
return self
|
|
|
|
def cursor(self): return self
|
|
def __enter__(self): return self
|
|
def __exit__(self, *args): pass
|
|
def close(self): self.closed += 1
|
|
def rollback(self): self.rollbacks += 1
|
|
|
|
def execute(self, sql, args=()):
|
|
self.calls.append((sql, args))
|
|
if self.fail: raise RuntimeError('private server error')
|
|
if sql.startswith(('SET ', 'START ')): self.result = []; return
|
|
if 'information_schema.TABLES' in sql:
|
|
self.result = [{'TABLE_NAME': t, 'ENGINE': self.engine} for t in args[1:]]; return
|
|
if 'information_schema.COLUMNS' in sql:
|
|
self.result = [{'COLUMN_NAME': r['name'], 'COLUMN_TYPE': 'int' if r['type']=='INTEGER' else 'text',
|
|
'IS_NULLABLE': 'YES', 'COLUMN_KEY': 'PRI' if r['pk'] else ''}
|
|
for r in self.db.execute('PRAGMA table_info(`'+args[1]+'`)')]; return
|
|
self.result = [dict(r) for r in self.db.execute(sql.replace('%s','?'), args)]
|
|
|
|
def fetchall(self): return copy.deepcopy(self.result)
|
|
|
|
|
|
def service(limit=50, profile=None):
|
|
db = Database(); limits = ProjectLimits(max_rows=limit)
|
|
provider = SQLDatabaseProvider('localhost', 3306, 'fixture', 'reader', 'snapshot',
|
|
['templates', 'spawns', 'links'], password_env='UNUSED', limits=limits, connection_factory=db.connect)
|
|
return ProjectResearch({}, database=provider, limits=limits, relations=profile or PROFILE), db
|
|
|
|
|
|
def query(value=42): return {'kind': 'database.relations', 'profile': 'inspect', 'value': value}
|
|
|
|
|
|
def test_complete_plan_single_readonly_snapshot_and_frozen_capture():
|
|
s, db = service()
|
|
capture = json.loads(s.collect(query()))
|
|
result = capture['result']; nodes = result['items']
|
|
assert result['coverage']=='complete_configured_relations' and not result['truncated']
|
|
assert [n['returned_rows'] for n in nodes]==[1,3,3]
|
|
assert nodes[1]['distribution']['groups']==[{'values':[14,1], 'count':3}]
|
|
assert nodes[2]['query']['filters'][0]['value']==[1,2,3]
|
|
assert capture['origin']['relation_profile_sha256']==s.relations.binding('inspect')
|
|
assert db.connections==db.closed==db.rollbacks==1
|
|
assert sum(sql.startswith('START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY') for sql,_ in db.calls)==1
|
|
assert all(sql.startswith(('SELECT ', 'SET SESSION ', 'SET TRANSACTION ', 'START TRANSACTION ')) for sql,_ in db.calls)
|
|
before = len(db.calls)
|
|
assert json.loads(s.collect(query()))==capture and len(db.calls)==before
|
|
|
|
|
|
def test_partial_parent_propagates_even_when_child_page_itself_is_complete():
|
|
s, db = service(limit=2)
|
|
# Remove one selected child's link so local child page is not truncated.
|
|
db.db.execute('DELETE FROM links WHERE guid=2')
|
|
r = json.loads(s.collect(query()))['result']; root,spawns,links=r['items']
|
|
assert root['complete'] and spawns['local_truncated']
|
|
assert not links['local_truncated'] and not links['upstream_complete'] and not links['complete']
|
|
assert r['truncated'] and r['coverage']=='partial_configured_relations'
|
|
assert spawns['distribution']['scope']=='returned_rows_only'
|
|
|
|
|
|
def test_empty_root_and_null_key_skip_without_empty_in_or_global_absence():
|
|
s, db = service()
|
|
r=json.loads(s.collect(query(99)))['result']
|
|
assert [n['rows'] for n in r['items']]==[[],[],[]]
|
|
assert r['items'][1]['skip_reason']=='no_nonnull_parent_keys'
|
|
assert r['items'][1]['query'] is None
|
|
assert 'dynamic code' in r['interpretation']
|
|
assert not any('IN ()' in sql for sql,_ in db.calls)
|
|
|
|
|
|
@pytest.mark.parametrize('value',["42 OR 1=1", "x' UNION SELECT * FROM links --"])
|
|
def test_values_are_bound_not_sql(value):
|
|
profile=copy.deepcopy(PROFILE);profile['inspect']['nodes'][0]['match_column']='name'
|
|
s, db=service(profile=profile); assert json.loads(s.collect(query(value)))['result']['items'][0]['rows']==[]
|
|
assert all(value not in sql for sql,_ in db.calls)
|
|
|
|
|
|
def test_numeric_start_keys_do_not_use_mysql_implicit_string_coercion():
|
|
s,_=service()
|
|
with pytest.raises(ToolError,match='VALUE_TYPE'):s.collect(query('42garbage'))
|
|
|
|
|
|
@pytest.mark.parametrize('change',[{'profile':'absent'}, {'sql':'SELECT * FROM accounts'}, {'value':True}, {'value':42.0}, {'table':'accounts'}])
|
|
def test_invalid_requests_do_not_connect(change):
|
|
s,db=service()
|
|
with pytest.raises(ToolError):s.collect({**query(),**change})
|
|
assert db.connections==0
|
|
|
|
|
|
@pytest.mark.parametrize('mutation',['cycle','unknown_parent_column','forbidden_table','bad_identifier','too_many_nodes'])
|
|
def test_invalid_operator_plans_are_rejected(mutation):
|
|
p=copy.deepcopy(PROFILE); n=p['inspect']['nodes']
|
|
if mutation=='cycle':n[1]['parent']='spawns'
|
|
elif mutation=='unknown_parent_column':n[1]['parent_column']='missing'
|
|
elif mutation=='forbidden_table':n[1]['table']='accounts'
|
|
elif mutation=='bad_identifier':n[1]['match_column']='x; DROP TABLE x'
|
|
else:p['inspect']['nodes']=n*5
|
|
with pytest.raises(ToolError):service(profile=p)
|
|
|
|
|
|
def test_unknown_schema_engine_and_sql_errors_are_controlled():
|
|
s,db=service();s.relations.profiles['inspect']['nodes'][0]['columns']=['missing']
|
|
with pytest.raises(ToolError,match='COLUMN_UNKNOWN'):s.collect(query())
|
|
assert db.closed==db.rollbacks==1
|
|
s,db=service();db.engine='MyISAM'
|
|
with pytest.raises(ToolError,match='SNAPSHOT_UNSUPPORTED'):s.collect(query())
|
|
s,db=service();db.fail=True
|
|
with pytest.raises(ToolError,match='QUERY_FAILED') as e:s.collect(query())
|
|
assert 'private' not in str(e.value) and db.closed==1
|
|
|
|
|
|
def test_result_limit_is_enforced_and_no_capture_published():
|
|
s,db=service();s.database.limits=ProjectLimits(max_result_bytes=100)
|
|
with pytest.raises(ToolError,match='RESULT_LIMIT'):s.collect(query())
|
|
assert not s._captures and db.closed==db.rollbacks==1
|
|
|
|
|
|
def test_profile_change_changes_capture_binding_and_missing_profiles_stay_unadvertised():
|
|
s,db=service();a=s.collect(query())
|
|
s.relations.profiles['inspect']['description']='Updated meaning.'
|
|
b=s.collect(query());assert a!=b and db.connections==2
|
|
assert len(ProjectResearch.schema()['oneOf'])+1==len(s.query_schema()['oneOf'])
|
|
with pytest.raises(ToolError):ProjectResearch({},database=s.database).collect(query())
|
|
|
|
|
|
def test_vcore_profile_is_adapter_data_and_generic_plan_has_no_wow_assumptions():
|
|
p=tomllib.loads(Path('config/project-relations.vcore.toml').read_text(encoding='utf-8'))
|
|
assert 'creature_spawn_context' in p['relations']
|
|
s,_=service(); assert 'Wolpertinger' not in json.dumps(s.relations.catalog())
|
|
|
|
|
|
def test_new_context_strict_actions_allow_relations_but_old_context_stays_frozen():
|
|
from test_investigation_lean import context as lean_context
|
|
from viano_bug_analysis.analysis.investigation_prompt import output_schema
|
|
c=lean_context();c.pop('lean_session',None)
|
|
old=output_schema(c,strict=True)['properties']['next']
|
|
action={'kind':'tool','operation':'project.query','arguments':query()}
|
|
assert not Draft202012Validator(old).is_valid(action)
|
|
c['tool_schema']['x-project-bindings']['relation_profiles']=PROFILE
|
|
assert Draft202012Validator(output_schema(c,strict=True)['properties']['next']).is_valid(action)
|
|
|
|
|
|
def test_total_deadline_is_controlled_and_publishes_no_partial_capture(monkeypatch):
|
|
import viano_bug_analysis.tools.database_relations as module
|
|
s,db=service();clock=iter([0,100])
|
|
monkeypatch.setattr(module.time,'monotonic',lambda:next(clock))
|
|
with pytest.raises(ToolError,match='RELATIONS_TIMEOUT'):s.collect(query())
|
|
assert not s._captures and db.closed==db.rollbacks==1
|
|
|
|
|
|
def test_relational_rows_participate_in_coding_value_guard():
|
|
from viano_bug_analysis.analysis.database_value_grounding import unsupported_assignments
|
|
s,_=service();capture=json.loads(s.collect(query()))
|
|
c={'evidence':{'E000001':{'content':{'items':[capture]}}}}
|
|
assert unsupported_assignments({'implementation_guidance':['Set mask=14.']},c)==['mask']
|
|
assert unsupported_assignments({'implementation_guidance':['Set mask=1.']},c)==[]
|
|
|
|
|
|
def test_relations_remain_portable_budgeted_evidence(tool_case,tmp_path):
|
|
from test_investigation_tools import batch
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.evidence import EvidenceStore
|
|
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
|
|
store=tool_case[0];s,db=service()
|
|
with PackageReader(tool_case[1]) as reader:s.project=reader.manifest['project']
|
|
router=InvestigationToolRouter(project_research=s)
|
|
budget=BudgetManager(store);budget.initialize('case-A','stable-package')
|
|
reserved=budget.reserve('case-A','stable-package','H1','A1','project.query')
|
|
asked=batch('project.query',query());reply=router.query(reserved.package,asked)
|
|
saved=EvidenceStore(store,router=router).store_tool_results('case-A','stable-package','relations-1',asked,reply)
|
|
assert PackageValidator().validate(saved.package).valid
|
|
budget.record_evidence('case-A','stable-package','H1','A1',saved.record_refs[0])
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'))
|
|
portable.intake(store.latest('case-A','stable-package'))
|
|
with PackageReader(portable.latest('case-A','stable-package')) as reader:
|
|
record=json.loads(reader.read_bytes(saved.record_refs[0]))
|
|
content=json.loads(reader.read_bytes(record['content_ref']))
|
|
capture=content['items'][0]
|
|
assert capture['result']['items'][2]['returned_rows']==3
|
|
assert capture['origin']['relation_plan']['nodes'][2]['parent_column']=='guid'
|
|
assert db.connections==1
|
|
|
|
|
|
def test_hybrid_loop_uses_relations_with_fixture_llm_and_can_resume(tool_case,tmp_path):
|
|
from test_case_interpreter import gateway
|
|
from test_investigation_hybrid import HybridAdapter
|
|
from test_investigator import run,state
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
|
|
s,db=service()
|
|
with PackageReader(tool_case[1]) as reader:s.project=reader.manifest['project']
|
|
adapter=HybridAdapter()
|
|
adapter.bootstrap.change=lambda r,c:r.update(next={'kind':'tool','operation':'project.query','arguments':query()})
|
|
router=InvestigationToolRouter(project_research=s)
|
|
loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,lean_checkpoint_steps=0),router=router)
|
|
first=run(loop,steps=3)
|
|
assert all(h['iterations']==1 for h in state(first)['hypotheses'].values())
|
|
assert 'relation_profiles' in adapter.calls[0]['input']['tool_schema']['project']
|
|
portable=CheckpointStore(Workspace(tmp_path/'resumed'));portable.intake(first.package)
|
|
final=run(Investigator(portable,gateway(adapter),router=router))
|
|
assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid
|
|
# Identical structured query is frozen once in the shared research service.
|
|
assert db.connections==1
|
|
|
|
|
|
def test_real_local_world_relations():
|
|
import os
|
|
config=os.environ.get('VBA_TEST_PROJECT_CONFIG')
|
|
if not config:pytest.skip('Opt-in local SELECT-only database')
|
|
s=ProjectResearch.load(config)
|
|
capture=json.loads(s.collect({'kind':'database.relations','profile':'creature_spawn_context','value':23487}))
|
|
nodes={n['node']:n for n in capture['result']['items']}
|
|
assert not capture['result']['truncated']
|
|
assert nodes['template']['rows'][0]['entry']==23487
|
|
assert len(nodes['spawns']['rows'])==18
|
|
assert {r['map'] for r in nodes['spawns']['rows']}=={1}
|
|
assert {r['guid'] for r in nodes['event_links']['rows']}=={r['guid'] for r in nodes['spawns']['rows']}
|
|
assert {r['eventEntry'] for r in nodes['event_links']['rows']}=={24}
|
|
assert nodes['events']['rows'][0]['eventEntry']==24
|