"""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