122 lines
7.4 KiB
Python
122 lines
7.4 KiB
Python
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
"""Opt-in, read-only checks against the local restored World snapshot; no LLM."""
|
|
import os
|
|
import pytest
|
|
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
from viano_bug_analysis.tools import ToolError
|
|
|
|
|
|
@pytest.fixture
|
|
def service():
|
|
path=os.environ.get("VBA_TEST_PROJECT_CONFIG")
|
|
if not path:
|
|
pytest.skip("Set VBA_TEST_PROJECT_CONFIG to a local SELECT-only development configuration")
|
|
return ProjectResearch.load(path)
|
|
|
|
|
|
def test_real_world_names_and_exact_absence(service):
|
|
p=service.database
|
|
query={"kind":"database.rows","table":"creature_template","columns":["entry","name"],
|
|
"filters":[{"column":"name","operator":"contains","value":"Wolpertinger"}],"offset":0,"limit":10}
|
|
result=p.query(query)
|
|
assert {r["row"]["entry"] for r in result["items"]}=={22943,23487}
|
|
count=p.query({"kind":"database.count","table":"creature","columns":[],
|
|
"filters":[{"column":"id","operator":"eq","value":23487},{"column":"map","operator":"eq","value":0}],"offset":0,"limit":1})
|
|
assert count["coverage"]=="complete_filtered_count" and count["items"]==[{"matched_rows":0}]
|
|
|
|
|
|
def test_real_readonly_privileges_and_isolation(service):
|
|
connection=service.database._connect()
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SHOW GRANTS FOR CURRENT_USER")
|
|
grants=[next(iter(row.values())) for row in cursor.fetchall()]
|
|
assert all(g.startswith("GRANT USAGE ") or g.startswith("GRANT SELECT ") for g in grants)
|
|
assert all("WITH GRANT OPTION" not in g for g in grants)
|
|
cursor.execute("SELECT @@read_only AS readonly")
|
|
assert cursor.fetchone()["readonly"]==1
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def test_real_schema_injection_and_source_schema(service):
|
|
p=service.database
|
|
schema=p.query({"kind":"database.schema","table":"creature"})
|
|
assert "position_x" in [c["COLUMN_NAME"] for c in schema["items"][0]["columns"]]
|
|
query={"kind":"database.rows","table":"creature_template","columns":["entry"],
|
|
"filters":[{"column":"name","operator":"eq","value":"' OR 1=1 --"}],"offset":0,"limit":1}
|
|
assert p.query(query)["items"]==[]
|
|
with pytest.raises(ToolError):p.query({**query,"table":"account"})
|
|
source=service.source.query({"kind":"source.schema","text":"CREATE TABLE `creature`","path_prefix":"sql/base/", "case_sensitive":True,"offset":0,"limit":5})
|
|
assert source["items"] and source["items"][0]["git_blob"]
|
|
|
|
|
|
def test_real_contextual_identity_resolution_with_fixture_llm(service, tmp_path):
|
|
"""Real backup + real canonical case; semantic replies are explicit fixtures, no API."""
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from viano_bug_analysis.analysis import CaseInterpreter
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, Workspace
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from test_case_interpreter import gateway, missing_payload
|
|
|
|
package=os.environ.get('VBA_TEST_CASE_PACKAGE')
|
|
if not package:
|
|
pytest.skip('Set VBA_TEST_CASE_PACKAGE to a built, not-yet-interpreted real case')
|
|
store=CheckpointStore(Workspace(tmp_path/'real-identity'))
|
|
store.intake(Path(package))
|
|
with PackageReader(package) as r:
|
|
identity=(r.manifest['case_id'],r.manifest['package_id'])
|
|
|
|
class Model:
|
|
def __init__(self):self.calls=[];self.mention=None
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.calls.append(request['prompt_version'])
|
|
data=json.loads(request['messages'][1]['content'])
|
|
if request['prompt_version']=='entity_resolution_plan_v4':
|
|
self.mention=next(k for k,v in data['context']['sources'].items() if v['kind']=='fact'
|
|
and v['data'].get('name')=='project_description'
|
|
and 'Wild Wolpertinger' in json.dumps(v['data']['value']))
|
|
doc={'searches':[{'catalog':'creatures','term':'Wild Wolpertinger','role':'affected_subject','source_ids':[self.mention]}]}
|
|
elif request['prompt_version']=='entity_resolution_assessment_v4':
|
|
candidates=data['candidates']
|
|
assert len(candidates)==1 and candidates[0]['id']=='23487'
|
|
assert data['captures']['V0001']['result']['items'][0]['row']['name']=='Wild Wolpertinger'
|
|
doc={'resolutions':[{'search_index':0,'status':'selected','candidate_id':candidates[0]['candidate_id'],
|
|
'reason':'Exact wild creature name matches the capture quest description, not its quest giver.',
|
|
'source_ids':[self.mention,'V0001']}]}
|
|
else:
|
|
context=restore(data['input'])
|
|
entities={k:v['data'] for k,v in context['sources'].items() if v['kind']=='entity'}
|
|
assert any(e['id']=='23486' for e in entities.values())
|
|
resolved=next(k for k,e in entities.items() if e['id']=='23487')
|
|
doc=missing_payload()
|
|
if request["prompt_version"]in {"case_interpretation_resolved_v3","case_interpretation_resolved_v4","case_interpretation_resolved_v5","case_interpretation_resolved_v6","case_interpretation_resolved_v7","case_interpretation_resolved_v8"}:
|
|
doc.update(described_behavior=[{'text':'The quest describes catching wild creatures using a net while intoxicated or wearing goggles.',
|
|
'source_ids':[self.mention],'basis':'local_description','verification':'not_independently_verified'}],
|
|
provisional_behavior_gap={'text':'The tester does not see these creatures despite wearing goggles.',
|
|
'source_ids':['O0002',self.mention],'basis':'local_description','status':'provisional'})
|
|
doc['observed_behavior']=[{'text':'Tester reports that Wolpertingers are not seen at the quest location.',
|
|
'source_ids':['O0001','O0002']}]
|
|
doc['interpretation_summary']={'text':'The described capture flow differs from the observation; formal verification is still missing.',
|
|
'source_ids':['O0002',self.mention,'Q0001']}
|
|
doc['relevant_entities']=[{'entity_id':resolved,'reason':'Wild creature required by the quest, distinct from Goldark.',
|
|
'source_ids':[resolved,self.mention]}]
|
|
return ProviderResponse(doc,'fixture-model',100,20)
|
|
|
|
model=Model()
|
|
result=asyncio.run(CaseInterpreter(store,gateway(model),project_research=service).interpret(*identity))
|
|
assert result.status=='COMPLETED' and PackageValidator().validate(result.package).valid
|
|
with PackageReader(result.package) as reader:
|
|
document=json.loads(reader.read_bytes(result.interpretation_ref))
|
|
assert document['interpretation']['described_behavior'][0]['verification']=='not_independently_verified'
|
|
assert document['interpretation']['provisional_behavior_gap']['status']=='provisional'
|
|
assert document['expected_behavior_completeness']=='missing'
|
|
assert 'Locally described behavior' in reader.read_bytes(result.summary_ref).decode()
|
|
assert len(model.calls)==3
|
|
assert asyncio.run(CaseInterpreter(store,gateway(model)).interpret(*identity)).reused
|
|
assert len(model.calls)==3
|