#!/usr/bin/env python3
"""Run sql/cases/*.sqlt against the Python translator.

The counterpart of php/bin/sqlt, and deliberately not all of it. That runner has
five checks; three ask about the HOST and are here, two ask about the MAP and the
SUITE and are not:

  here   run_case          the emitted string, the error code and position, the
                           parameter list -- what this host does with shared data
  here   the mirror re-run every mariadb case re-run as mysql, requiring the same
                           string: does THIS host emit identical SQL for a leaf
                           that overrides nothing
  here   the slot checks   every bound value is emitted, and bindings() agrees in
                           count with the placeholders

  not    caveat_pins       every caveated entry is pinned by some case. A property
                           of the case files against the dialect data, both shared
                           and both already checked by PHP. Running it twice
                           measures the same thing twice.
  not    mirror_parity     mariadb and mysql declare the same map. Likewise a
                           property of sql/dialects/*.json and nothing to do with
                           which host reads it.

That split is why python/sel/sql/map.py has no trace facility: nothing here needs
one, and adding one would only be to re-run a check PHP already owns.

    python/bin/sqlt [name-substring ...]
"""

from __future__ import annotations

import json
import os
import re
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
# case_data.py is generated beside this script and is runner tooling, not part
# of the package, so its directory is always added.
sys.path.insert(0, _HERE)
# The source tree is added only when `sel` is not already importable, so running
# this under the wheel's interpreter grades the WHEEL rather than silently
# shadowing it with python/sel/. Same probe as python/bin/conformance.py.
try:
    import sel as _sel_probe                                        # noqa: F401
except ImportError:
    sys.path.insert(1, os.path.join(_HERE, '..'))


class SuiteError(Exception):
    """A malformed suite. Not a failing case -- a suite that cannot be run."""


# Imported by load_translator() rather than here, so an import error reports as
# a suite error rather than a traceback at start-up. The case data is generated
# code that constructs Bindings, so it needs the package -- there is no longer a
# reading of the roster that is independent of the library, and there no longer
# needs to be: one generator writes both hosts' copies, so they cannot disagree
# about what the suite contains.
sel = None
SelError = Exception
Sql = None
SqlError = None
sqlmap = None


SQL_CASES = None


def load_translator() -> None:
    global sel, SelError, Sql, SqlError, sqlmap, SQL_CASES
    import sel as _sel
    from sel.errors import SelError as _SelError
    from sel.sql import Sql as _Sql, SqlError as _SqlError
    from sel.sql import map as _map
    from case_data import SQL_CASES as _cases
    sel, SelError, Sql, SqlError, sqlmap = _sel, _SelError, _Sql, _SqlError, _map
    SQL_CASES = _cases

MIRRORS = {'mariadb': 'mysql'}

# `--- throws` names PHP's class, because the cases were written for one host.
# The line it draws is the one that matters and it is the same in both: a
# malformed map or registration is a mistake in the application's startup, not a
# rule that cannot be translated, so it must NOT be catchable as SqlError --
# tryTranslate() swallows the second and must not swallow the first. PHP spells
# that LogicException; this host spells it RuntimeError, which is what
# sel/registry.py already raises for a function defined twice.
#
# A name with no entry here is a suite error rather than a pass: an expectation
# nobody has mapped must never be satisfied by whatever happened to be raised.
THROWS = {'LogicException': RuntimeError}


def apply_registrations(ops) -> None:
    """A case's runtime registrations, applied before translating."""
    if ops is None:
        return
    for op in ops:
        if 'define' in op:
            dialect, section, key, entry = op['define']
            sqlmap.define(dialect, section, key, entry)
            continue
        if 'dialect' in op:
            rest = {k: v for k, v in op.items() if k != 'dialect'}
            sqlmap.define_dialect(op['dialect'], rest)
            continue
        raise SuiteError('a register op needs a dialect or a define')


ERROR_RE = re.compile(r'^(\S+)(?:\s+(\d+):(\d+))?$')
TILDE_RE = re.compile(r'~\d+~')


def snapshot_ast(node) -> tuple | None:
    """The tree as nested tuples, with the registry's Spec objects left out:
    they are looked up by name and compared by identity, and a snapshot is
    compared by value. Everything else that identifies a node -- kind, position,
    literal, name, operator, grouping, children -- is in here.
    """
    if node is None:
        return None
    return (node.t, node.pos, node.v, node.name, node.op, node.grouped,
            snapshot_ast(node.l), snapshot_ast(node.r), snapshot_ast(node.x),
            snapshot_ast(node.obj), snapshot_ast(node.idx),
            snapshot_ast(node.target), snapshot_ast(node.value),
            tuple(snapshot_ast(i) for i in node.items),
            tuple(snapshot_ast(a) for a in node.args))


def classify(plan) -> str:
    if plan.pure_sql:
        return 'pure_sql'
    if plan.pure_memory:
        return 'pure_memory'
    return 'hybrid'


def run_plan_case(c: dict) -> str | None:
    """None when the planner case passes, else what went wrong.

    A `--- plan` case asks the planner rather than the translator. It asserts
    the classification, the physical sources, the SQL prefix, that the
    continuation exists exactly when the classification says so, and that the
    caller's AST is the same tree afterwards -- after planning, which runs
    stage 1 and the logical optimiser, and after the physical optimiser
    Program.run() uses.
    """
    from sel.optimizer import optimize_ast_in_memory
    dialect = c['dialect']
    options = c['options'] or {}
    mode = c['mode'] or 'inline'
    plan = None
    error = None
    program = None
    before = None
    try:
        apply_registrations(c['register'])
        bindings = c['bindings']()
        program = sel.compile(c['source'])
        before = snapshot_ast(program.ast)
        plan = Sql.plan_hybrid(program, dialect, bindings, options)
    except SqlError as e:
        error = e
    except SelError as e:
        return f'the source did not compile: {e}'
    except SuiteError:
        raise
    except Exception as e:                                    # noqa: BLE001
        raise SuiteError(f"{c['at']}: unexpected {type(e).__name__}: {e}")

    if c['plan'] == 'refused':
        if error is None:
            return f"expected {c['error']}, got a {classify(plan)} plan"
        m = ERROR_RE.match(c['error'])
        if m is None:
            raise SuiteError(f"{c['at']}: malformed error expectation")
        # The code alone: a plan refusal blames the bindings or the dialect,
        # not a node of the rule, so it carries no position, and the generator
        # refuses a case that writes one (SEL-0046).
        if error.code != m.group(1):
            return f'expected {m.group(1)}, got {error.code} ({error.message})'
        return None
    if error is not None:
        return f"expected a {c['plan']} plan, got {error.code} ({error.message})"

    got = classify(plan)
    if got != c['plan']:
        return f"expected a {c['plan']} plan, got {got}"
    if c['tables'] is not None and list(plan.source_tables) != list(c['tables']):
        return (f"source tables got:  {json.dumps(list(plan.source_tables))}\n"
                f"     want: {json.dumps(c['tables'])}")
    if plan.dialect != dialect:
        return f'plan.dialect is {plan.dialect}, not {dialect}'

    if c['plan'] == 'pure_memory':
        if plan.sql_statement is not None:
            return 'a pure-memory plan carries a SQL statement'
        if plan.continuation_program is not program:
            return 'a pure-memory plan must run the original program'
        if plan.continuation_ast is not program.ast:
            return 'a pure-memory plan must expose the original AST as its continuation'
    else:
        if plan.sql_statement is None:
            return f"a {c['plan']} plan has no SQL statement"
        if plan.sql_prefix_ast is None:
            return f"a {c['plan']} plan has no SQL prefix AST"
        sql = plan.sql_statement.as_statement(mode)
        if sql != c['expect']:
            return f"got:  {sql}\n     want: {c['expect']}"
        if c['plan'] == 'pure_sql':
            if plan.continuation_program is not None or plan.continuation_ast is not None:
                return 'a pure-SQL plan carries a continuation'
        elif plan.continuation_program is None or plan.continuation_ast is None:
            return 'a hybrid plan has no continuation'

    # Planning must not have touched the tree, and neither may the physical
    # optimiser that every run() goes through.
    if snapshot_ast(program.ast) != before:
        return 'planning mutated the program AST'
    optimize_ast_in_memory(program.ast)
    if snapshot_ast(program.ast) != before:
        return 'the physical optimiser mutated the program AST'
    return None


def run_case(c: dict) -> str | None:
    """None when the case passes, else what went wrong."""
    dialect = c['dialect']
    if dialect is None or dialect == '':
        raise SuiteError(f"{c['at']}: case {c['name']} has no --- dialect")
    options = c['options'] or {}
    as_ = c['as'] or 'value'
    mode = c['mode'] or 'inline'

    sql = None
    error = None
    thrown = None
    frag = None
    program = None
    bindings = None
    try:
        # Inside the try: a bad registration is one of the outcomes a case may
        # assert, so it has to be catchable rather than fatal.
        apply_registrations(c['register'])
        # Inside the try: a Binding constructor refuses a malformed binding at
        # the earliest possible moment, which is construction rather than
        # translation, and that refusal is one of the outcomes a case asserts.
        bindings = c['bindings']()
        program = sel.compile(c['source'])
        frag = Sql.translate(program, dialect, bindings, options)
        if as_ == 'condition':
            sql = frag.as_condition(mode)
        elif as_ == 'statement':
            sql = frag.as_statement(mode)
        else:
            sql = frag.as_value(mode)
    except SqlError as e:
        error = e
    except SelError as e:
        return f'the source did not compile: {e}'
    except SuiteError:
        raise
    except Exception as e:                                    # noqa: BLE001
        thrown = e

    # A malformed map or binding is a mistake in the application's startup and
    # raises the host's own error type; a rule that cannot be translated raises
    # SqlError. Keeping the two apart is why `throws` exists as its own outcome
    # rather than as another E_SQL_ code.
    #
    # The case files name PHP's class -- they were written for one host -- so the
    # name is mapped rather than compared. A name with no mapping is a suite
    # error and not a pass: an unrecognised expectation must never be satisfied
    # by whatever happened to be raised.
    if c['throws'] is not None:
        want = THROWS.get(c['throws'])
        if want is None:
            raise SuiteError(f"{c['at']}: no Python equivalent is recorded for "
                             f"--- throws {c['throws']}; add one to THROWS")
        if thrown is None:
            return f"expected {c['throws']}, got " + (str(error) if error else repr(sql))
        if isinstance(thrown, want):
            return None
        return (f"expected {c['throws']} ({want.__name__}), got "
                f'{type(thrown).__name__} ({thrown})')
    if thrown is not None:
        raise SuiteError(f"{c['at']}: unexpected {type(thrown).__name__}: {thrown}")

    # Every `--- as statement` case is also run through translate_statement,
    # the public full-delegation entry point, which must say exactly what
    # translate() says -- the same text, or the same refusal at the same
    # column. Two hosts ran the logical optimiser in that lane and three did
    # not, and only a twin check can see it (review 2026-09-15 finding C).
    if c.get('as') == 'statement' and program is not None:
        twin_sql = None
        twin_error = None
        try:
            twin_sql = Sql.translate_statement(program, dialect, bindings, options).as_statement(mode)
        except SqlError as e:
            twin_error = e
        except Exception as e:                                    # noqa: BLE001
            raise SuiteError(f"{c['at']}: translate_statement threw {type(e).__name__}: {e}")

        def got(e):
            return 'SQL' if e is None else f'{e.code} at {e.line}:{e.col}'
        if error is not None or twin_error is not None:
            if (error is None or twin_error is None or error.code != twin_error.code
                    or error.line != twin_error.line or error.col != twin_error.col):
                return f'translate() gave {got(error)} but translate_statement() gave {got(twin_error)}'
        elif twin_sql != sql:
            return f'translate_statement() disagrees with translate():\n     {twin_sql}\n     {sql}'

    if c['error'] is not None:
        if error is None:
            return f"expected {c['error']}, got {sql!r}"
        m = ERROR_RE.match(c['error'])
        if m is None:
            raise SuiteError(f"{c['at']}: malformed error expectation")
        if error.code != m.group(1):
            return f'expected {m.group(1)}, got {error.code} ({error.message})'
        if m.group(2) is not None:
            want_pos = f'{m.group(2)}:{m.group(3)}'
            got_pos = f'{error.line}:{error.col}'
            if want_pos != got_pos:
                return f'expected {m.group(1)} at {want_pos}, got it at {got_pos}'
        return None

    if error is not None:
        return f'expected SQL, got {error.code} ({error.message})'
    if sql != c['expect']:
        return f"got:  {sql}\n     want: {c['expect']}"

    # Checked for every case that produces a fragment, not only those asking
    # about params: every slot in the part list must have a value, and every
    # value must be emitted. A value bound but never emitted means a Fragment
    # was rendered and thrown away -- invisible in `inline` mode, which is what
    # the rest of the suite asserts.
    seen = set()
    for p in frag.parts:
        if isinstance(p, str):
            continue
        if p < 1 or p > len(frag.params):
            return f'parameter slot {p} has no value in params'
        seen.add(p)
    orphans = [i for i in range(1, len(frag.params) + 1) if i not in seen]
    if orphans:
        return (f'parameter slot(s) {json.dumps(orphans)} were bound but never '
                'emitted — a fragment was rendered and discarded')
    # `~1~`, not bare tildes: PostgreSQL's regex operator IS `~`, so counting
    # them divided a regex fragment's odd tilde count by two. PHP's runner
    # shipped that bug; this is the corrected form from the start.
    debug_str = frag.as_statement('debug') if frag.kind == 'STATEMENT' else frag.as_value('debug')
    if len(frag.bindings()) != len(TILDE_RE.findall(debug_str)):
        return 'bindings() and the emitted placeholders disagree in count'

    if c['params'] is not None:
        got = ', '.join(v.dump() for v in frag.bindings())
        if got != c['params']:
            return f"params got:  {got}\n     want: {c['params']}"
    return None


def main(argv: list[str]) -> int:
    filters = argv[1:]
    load_translator()
    cases = SQL_CASES

    # `--names` prints what this host loaded and stops. Kept as a cheap way to
    # see the roster; the two hosts cannot disagree about it any more, because
    # tools/gen-sql-cases.mjs is the only thing that reads sql/cases/*.sqlt and
    # both load what it wrote.
    if filters == ['--names']:
        for c in cases:
            print(f"{c['at']}\t{c['name']}")
        return 0

    passed = 0
    mirrored = 0
    failures: list[tuple[dict, str]] = []
    suite_errors = 0

    for c in cases:
        if filters and not any(f in c['name'] for f in filters):
            continue
        sqlmap.reset()        # no case may leak a registration into another
        try:
            problem = run_plan_case(c) if c['plan'] else run_case(c)
        except SuiteError as e:
            print(f'SUITE ERROR {e}')
            suite_errors += 1
            continue
        if problem is not None:
            failures.append((c, problem))
            continue
        passed += 1

        # The same case under the mirrored dialect. A registration case is
        # exempt: it names its dialect in the register data, so re-running it
        # under another name would be testing something it does not claim.
        mirror = MIRRORS.get(c['dialect'])
        if mirror is None or c['register'] is not None:
            continue
        sqlmap.reset()
        # Python's {**a, 'k': v} is RIGHT-wins, where PHP's `+` is left-wins;
        # the assertion below is the same one php/bin/sqlt carries, because
        # getting this backwards re-runs every mariadb case as mariadb and the
        # count line still prints a plausible number.
        mirror_case = {**c, 'dialect': mirror}
        # Same semantics, different supported NO PAD collation names. Pin the
        # spelling rather than consulting the map, so map mutations still fail.
        if mirror_case.get('expect'):
            mirror_case['expect'] = mirror_case['expect'].replace(
                ' COLLATE utf8mb4_nopad_bin', ' COLLATE utf8mb4_0900_bin')
        if mirror_case['dialect'] != mirror:
            print(f"SUITE ERROR the mirrored case for {c['name']} is still "
                  f"{mirror_case['dialect']}, so nothing is being mirrored")
            suite_errors += 1
            continue
        try:
            problem = run_plan_case(mirror_case) if mirror_case['plan'] else run_case(mirror_case)
        except SuiteError as e:
            print(f'SUITE ERROR (mirrored to {mirror}) {e}')
            suite_errors += 1
            continue
        if problem is None:
            mirrored += 1
        else:
            failures.append((c, f'mirrored to {mirror}, which must agree '
                                f"with {c['dialect']}: {problem}"))

    for c, problem in failures:
        print(f"FAIL {c['name']}  ({c['at']})")
        print(f'     {problem}')

    print(f'\n{passed} passed ({mirrored} also checked against a mirrored '
          f'dialect), {len(failures)} failed, {suite_errors} suite errors')
    return 0 if not failures and suite_errors == 0 else 1


if __name__ == '__main__':
    sys.exit(main(sys.argv))
