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')