191 lines
10 KiB
Python
191 lines
10 KiB
Python
import copy
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.tools.config import ToolError
|
|
from viano_bug_analysis.tools.sql_database import SQLDatabaseProvider
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
from viano_bug_analysis.tools.context_expansion import ContextExpansion
|
|
from viano_bug_analysis.analysis.entity_resolution import policy
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case,batch
|
|
|
|
|
|
class Database:
|
|
"""Real attached-schema SQL execution; only MariaDB metadata/transaction syntax simulated."""
|
|
def __init__(self):
|
|
self.db=sqlite3.connect(':memory:');self.db.row_factory=sqlite3.Row
|
|
self.names={'cata_world':['switches'],'cata_hotfixes':['switches'],'cata_locale':['texts'],'cata_auth':['accounts']}
|
|
for name,tables in self.names.items():
|
|
self.db.execute(f"ATTACH DATABASE ':memory:' AS {name}")
|
|
for table in tables:
|
|
self.db.execute(f'CREATE TABLE {name}.{table}(id INTEGER PRIMARY KEY,name TEXT)')
|
|
self.db.execute(f'INSERT INTO {name}.{table} VALUES (1,?)',(name,))
|
|
self.calls=[];self.connections=0;self.closed=0;self.rollbacks=0;self.engine='InnoDB'
|
|
def connect(self):self.connections+=1;return self
|
|
def close(self):self.closed+=1
|
|
def rollback(self):self.rollbacks+=1
|
|
def cursor(self):return self
|
|
def __enter__(self):return self
|
|
def __exit__(self,*args):pass
|
|
def fetchall(self):return copy.deepcopy(self.result)
|
|
def execute(self,sql,args=()):
|
|
self.calls.append((sql,args))
|
|
if sql.startswith(('SET ','START ')):self.result=[];return
|
|
if 'information_schema.COLUMNS' in sql:
|
|
self.result=[{'COLUMN_NAME':row['name'],'COLUMN_TYPE':'int' if row['type']=='INTEGER' else 'text',
|
|
'IS_NULLABLE':'YES','COLUMN_KEY':'PRI' if row['pk'] else ''}
|
|
for row in self.db.execute(f'PRAGMA {args[0]}.table_info(`{args[1]}`)')];return
|
|
if 'information_schema.TABLES' in sql:
|
|
names=self.names.get(args[0],[])
|
|
if 'ENGINE' in sql:self.result=[{'TABLE_NAME':name,'ENGINE':self.engine} for name in names if name in args[1:]]
|
|
else:self.result=[{'TABLE_NAME':name} for name in sorted(names)[:args[1]]]
|
|
return
|
|
self.result=[dict(row) for row in self.db.execute(sql.replace('%s','?'),args)]
|
|
|
|
|
|
def setup(additional=None):
|
|
db=Database()
|
|
provider=SQLDatabaseProvider('127.0.0.1',3306,'cata_world','buganalysis','snapshot',['switches'],
|
|
password_env='SECRET_NOT_READ',connection_factory=db.connect,
|
|
additional_databases=additional if additional is not None else {'cata_hotfixes':['switches'],'cata_locale':['*']})
|
|
return ProjectResearch({},database=provider),db
|
|
|
|
|
|
def rows(table):return {'kind':'database.rows','table':table,'columns':['id','name'],
|
|
'filters':[{'column':'id','operator':'eq','value':1}],'offset':0,'limit':10}
|
|
|
|
|
|
def test_same_named_tables_have_independent_schemas_and_provenance():
|
|
service,db=setup()
|
|
world=json.loads(service.collect(rows('switches')))
|
|
hotfix=json.loads(service.collect(rows('cata_hotfixes.switches')))
|
|
locale=json.loads(service.collect(rows('cata_locale.texts')))
|
|
assert world['result']['items'][0]['row']['name']=='cata_world'
|
|
assert hotfix['result']['items'][0]['row']['name']=='cata_hotfixes'
|
|
assert locale['result']['items'][0]['row']['name']=='cata_locale'
|
|
assert world['origin']['selected_database']=='cata_world'
|
|
assert hotfix['origin']['selected_database']=='cata_hotfixes'
|
|
assert 'buganalysis' not in str(hotfix['origin']) and '127.0.0.1' not in str(hotfix['origin'])
|
|
assert any('FROM `cata_hotfixes`.`switches`' in sql for sql,_ in db.calls)
|
|
assert {args[0] for sql,args in db.calls if 'information_schema.COLUMNS' in sql}=={'cata_world','cata_hotfixes','cata_locale'}
|
|
|
|
|
|
@pytest.mark.parametrize('table',['cata_auth.accounts','cata_characters.characters','cata_locale_reference.texts',
|
|
'cata_hotfixes.forbidden','cata_locale.texts.extra','cata_locale.texts;DROP TABLE x'])
|
|
def test_unconfigured_databases_tables_and_injection_blocked_before_access(table):
|
|
service,db=setup()
|
|
with pytest.raises(ToolError):service.collect(rows(table))
|
|
assert db.connections==0
|
|
|
|
|
|
def test_catalog_and_readiness_expand_only_explicitly_allowed_schemas():
|
|
service,db=setup()
|
|
result=json.loads(service.collect({'kind':'database.tables','text':'','offset':0,'limit':10}))['result']
|
|
assert [r['table'] for r in result['items']]==['cata_hotfixes.switches','cata_locale.texts','cata_world.switches']
|
|
assert service.database.schema_targets==tuple(r['table'] for r in result['items'])
|
|
assert {args[0] for sql,args in db.calls if 'information_schema.TABLES' in sql}=={'cata_locale'}
|
|
page=json.loads(service.collect({'kind':'database.tables','text':'','offset':1,'limit':1}))['result']
|
|
assert page['items'][0]['table']=='cata_locale.texts' and page['truncated'] and page['next_offset']==2
|
|
|
|
|
|
def test_cross_database_relations_use_one_read_only_repeatable_snapshot():
|
|
service,db=setup()
|
|
from viano_bug_analysis.tools.database_relations import DatabaseRelations
|
|
profile={'inspect':{'description':'Linked schema records','nodes':[
|
|
{'name':'root','table':'switches','match_column':'id','columns':['id','name']},
|
|
{'name':'child','table':'cata_hotfixes.switches','match_column':'id','columns':['id','name'],
|
|
'parent':'root','parent_column':'id'},
|
|
{'name':'locale','table':'cata_locale.texts','match_column':'id','columns':['id','name'],
|
|
'parent':'child','parent_column':'id'}]}}
|
|
service.relations=DatabaseRelations(service.database,profile)
|
|
capture=json.loads(service.collect({'kind':'database.relations','profile':'inspect','value':1}))
|
|
assert [n['database'] for n in capture['result']['items']]==['cata_world','cata_hotfixes','cata_locale']
|
|
assert capture['result']['coverage']=='complete_configured_relations'
|
|
assert db.connections==db.closed==db.rollbacks==1
|
|
assert sum(sql.startswith('START TRANSACTION') for sql,_ in db.calls)==1
|
|
assert capture['origin']['consistency']=='single_repeatable_read_read_only_transaction'
|
|
|
|
|
|
def test_entity_resolution_and_context_rules_allow_qualified_tables():
|
|
catalog={'id':'names','entity_type':'creature','table':'cata_locale.texts','id_column':'id','name_column':'name','columns':['id','name']}
|
|
assert policy({'enabled':True,'catalogs':[catalog]})['catalogs'][0]['table']=='cata_locale.texts'
|
|
service,_=setup()
|
|
expansion=ContextExpansion(service,{'entities':[{'entity_type':'creature','table':'texts',
|
|
'column':'id','name_column':'name','columns':['id','name']}]})
|
|
assert expansion.same_table('switches','cata_world.switches')
|
|
assert not expansion.same_table('switches','cata_hotfixes.switches')
|
|
expansion=ContextExpansion(service,{'entities':[{'entity_type':'creature','table':'cata_locale.texts',
|
|
'column':'id','name_column':'name','columns':['id','name']}]})
|
|
_,facts,_,_=expansion.collect([rows('cata_locale.texts')])
|
|
assert any(f['name']=='project_entity_name' and f['value']['name']=='cata_locale' for f in facts)
|
|
|
|
|
|
def test_config_loading_and_legacy_binding(tmp_path):
|
|
path=tmp_path/'project.toml'
|
|
path.write_text('''[project]
|
|
type="wow"
|
|
profile="vcore"
|
|
[database]
|
|
host="127.0.0.1"
|
|
port=3306
|
|
database="cata_world"
|
|
user="buganalysis"
|
|
password_env="VIANO_DB_PASSWORD"
|
|
snapshot_id="test"
|
|
tables=["switches"]
|
|
[database.additional_databases]
|
|
cata_hotfixes=["switches"]
|
|
cata_locale=["*"]
|
|
''',encoding='utf-8')
|
|
service=ProjectResearch.load(path)
|
|
assert service.database.resolve_table('cata_locale.texts')==('cata_locale','texts')
|
|
with pytest.raises(ToolError):service.database.resolve_table('cata_auth.accounts')
|
|
legacy,_=setup({})
|
|
assert legacy.database.binding()=={'kind':'mariadb','database':'cata_world','snapshot_id':'snapshot',
|
|
'tables':['switches'],'time_scope':'configured_snapshot_not_current_production'}
|
|
assert legacy.database.binding_for(rows('switches'))==legacy.database.binding()
|
|
|
|
|
|
def test_multi_database_evidence_survives_package_validation(tool_case):
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.evidence import EvidenceStore
|
|
store=tool_case[0];service,_=setup()
|
|
with PackageReader(tool_case[1]) as reader:service.project=reader.manifest['project']
|
|
router=InvestigationToolRouter(project_research=service)
|
|
budget=BudgetManager(store);budget.initialize('case-A','stable-package')
|
|
reserved=budget.reserve('case-A','stable-package','H1','A1','project.query')
|
|
request=batch('project.query',rows('cata_hotfixes.switches'))
|
|
reply=router.query(reserved.package,request)
|
|
saved=EvidenceStore(store,router=router).store_tool_results('case-A','stable-package','multi-db',request,reply)
|
|
assert PackageValidator().validate(saved.package).valid
|
|
assert reply['results'][0]['data']['origin']['selected_database']=='cata_hotfixes'
|
|
|
|
|
|
def test_worker_readiness_checks_all_databases_and_rejects_missing_one(tmp_path,monkeypatch):
|
|
from types import SimpleNamespace
|
|
from viano_bug_analysis import deployment
|
|
from viano_bug_analysis.api import APIConfig,Project
|
|
service,db=setup();service.source=SimpleNamespace(inventory=lambda:{})
|
|
config=APIConfig(tmp_path,(Project('p','wow','vcore'),),())
|
|
monkeypatch.setattr(deployment.WorkerConfig,'load',lambda _:SimpleNamespace(intake=config,min_free_bytes=1))
|
|
loader=lambda _:(object(),{'project_research':service},'hash')
|
|
assert deployment.check('worker','unused',loader=loader)['status']=='READY'
|
|
assert {args[0] for sql,args in db.calls if 'information_schema.COLUMNS' in sql}==set(service.database.table_allowlists)
|
|
db.names['cata_locale']=[]
|
|
result=deployment.check('worker','unused',loader=loader)
|
|
assert result['status']=='NOT_READY' and result['code']=='PROJECT_DATABASE_TABLE_UNKNOWN'
|
|
|
|
|
|
def test_full_vcore_multidatabase_template_loads_without_network():
|
|
from pathlib import Path
|
|
service=ProjectResearch.load(Path(__file__).resolve().parents[1]/'config/project.vcore-multidb.example.toml')
|
|
assert set(service.database.table_allowlists)=={'cata_world','cata_hotfixes','cata_locale'}
|
|
assert service.database.user=='buganalysis'
|
|
assert service.relations and service.enrichment and service.entity_resolution
|
|
with pytest.raises(ToolError):service.database.resolve_table('cata_auth.accounts')
|