scripts/sql_fixture_checker.py

#!/usr/bin/env python3
'''Synthetic SQLite learning fixture: no live DB or network access.

This is not a parser, authorization engine or production SQL validator.
'''
import json
import re
import sqlite3
from typing import Any


def fixture_connection() -> sqlite3.Connection:
    c = sqlite3.connect(':memory:')
    c.executescript('''
        CREATE TABLE orders(order_id INTEGER PRIMARY KEY, customer_id TEXT NOT NULL,
                            order_total REAL NOT NULL, order_ts TEXT NOT NULL);
        CREATE TABLE order_items(order_id INTEGER, item_id INTEGER, line_total REAL NOT NULL);
        CREATE TABLE status_history(order_id INTEGER, status TEXT, changed_ts TEXT, event_id INTEGER);
        CREATE TABLE blacklisted(account TEXT);
        CREATE TABLE accounts(account TEXT);
        CREATE TABLE cdc(business_key TEXT, source_version INTEGER,
                         event_ts TEXT, event_id INTEGER, value TEXT);
        INSERT INTO orders VALUES
          (1, 'C001', 100.0, '2026-01-01T00:00:00'),
          (2, 'C002', 75.0, '2026-01-31T23:59:59'),
          (3, 'C003', 100.0, '2026-02-01T00:00:00');
        INSERT INTO order_items VALUES
          (1, 1, 60.0),(1, 2, 40.0),(2, 1, 75.0),
          (3, 1, 20.0),(3, 2, 30.0),(3, 3, 50.0);
        INSERT INTO status_history VALUES
          (1,'created','2026-01-01T00:00:00',101),
          (1,'paid','2026-01-01T01:00:00',102),
          (1,'shipped','2026-01-01T01:00:00',103),
          (2,'paid','2026-01-31T23:59:59',104);
        INSERT INTO blacklisted VALUES ('A'),(NULL);
        INSERT INTO accounts VALUES ('A'),('B'),('C');
        INSERT INTO cdc VALUES
          ('K1',1,'2026-01-01T12:00:00',10,'old'),
          ('K1',2,'2026-01-01T13:00:00',11,'new'),
          ('K1',2,'2026-01-01T13:00:00',12,'winner'),
          ('K2',1,'2026-01-01T15:00:00',13,'only');
    ''')
    return c


def join_fanout(c: sqlite3.Connection) -> dict[str, Any]:
    source = c.execute('SELECT COUNT(*), SUM(order_total) FROM orders').fetchone()
    joined = c.execute('''
      SELECT COUNT(*), SUM(o.order_total)
      FROM orders o JOIN order_items i USING(order_id)
    ''').fetchone()
    corrected = c.execute('SELECT SUM(line_total) FROM order_items').fetchone()[0]
    return {'source_rows': source[0], 'joined_rows': joined[0],
            'source_total': source[1], 'incorrect_joined_total': joined[1],
            'correct_item_total': corrected}


def latest_status(c: sqlite3.Connection) -> list[tuple]:
    return c.execute('''
      WITH ranked AS (
        SELECT order_id, status,
               ROW_NUMBER() OVER (
                 PARTITION BY order_id
                 ORDER BY changed_ts DESC, event_id DESC) AS rn
        FROM status_history
      )
      SELECT order_id, status FROM ranked WHERE rn=1 ORDER BY order_id
    ''').fetchall()


def null_filter(c: sqlite3.Connection) -> dict[str, list[str]]:
    broken = c.execute('''SELECT account FROM accounts
        WHERE account NOT IN (SELECT account FROM blacklisted)''').fetchall()
    fixed = c.execute('''SELECT a.account FROM accounts a
        WHERE NOT EXISTS (SELECT 1 FROM blacklisted b WHERE b.account=a.account)
        ORDER BY a.account''').fetchall()
    return {'not_in': [r[0] for r in broken], 'not_exists': [r[0] for r in fixed]}


def half_open_window(c: sqlite3.Connection) -> list[int]:
    return [r[0] for r in c.execute('''
      SELECT order_id FROM orders
      WHERE order_ts >= '2026-01-01T00:00:00'
        AND order_ts < '2026-02-01T00:00:00'
      ORDER BY order_id
    ''').fetchall()]


def cdc_winners(c: sqlite3.Connection) -> list[tuple]:
    return c.execute('''
       WITH ranked AS (
         SELECT business_key, value,
                ROW_NUMBER() OVER (
                  PARTITION BY business_key
                  ORDER BY source_version DESC, event_ts DESC, event_id DESC) rn
         FROM cdc
       )
       SELECT business_key, value FROM ranked WHERE rn=1 ORDER BY business_key
    ''').fetchall()


def advisory_sql_flags(sql: str) -> list[str]:
    '''Very limited advisory signal only; never use to authorize database execution.'''
    flags=[]
    if re.search(r'\b(?:DELETE|UPDATE|MERGE|TRUNCATE|DROP|ALTER|GRANT|REVOKE)\b', sql, re.I):
        flags.append('potentially_mutating_or_privilege_related')
    if re.search(r'\bDELETE\s+FROM\s+\w+\s*;?\s*$', sql.strip(), re.I):
        flags.append('possible_unqualified_delete')
    if re.search(r'\bSELECT\s+\*', sql, re.I):
        flags.append('select_all_projection_review')
    return flags


def report() -> dict[str, Any]:
    with fixture_connection() as conn:
        return {'fixture_type':'sqlite_in_memory_only',
            'fanout':join_fanout(conn), 'latest_status':latest_status(conn),
            'null_filter':null_filter(conn), 'half_open_window':half_open_window(conn),
            'cdc_winners':cdc_winners(conn),
            'limitations':'Does not establish correctness on non-SQLite target engines or user data.'}

if __name__ == '__main__':
    print(json.dumps(report(),indent=2))