#!/usr/bin/env python3 """Execute the published SQL fixture with SQLite; this is not MySQL evidence.""" import argparse from decimal import Decimal from pathlib import Path import re import sqlite3 import sys # Worked fixture outcomes, deliberately literal rather than recomputed with SQL. EXPECTED = { "window_frames": [(11, 101, 9000, 9000, 18000), (11, 102, 9000, 18000, 18000), (12, 103, 7000, 7000, 7000), (12, 104, 7000, 14000, 14000), (12, 105, 5000, 19000, 19000), (21, 201, 4000, 4000, 6000), (21, 202, 2000, 6000, 6000)], "first_threshold": [(11, 102, 18000), (12, 105, 19000)], "null_logic": [(None, None, 1, 0, 1)], "not_in_null": [], "not_exists_null": [(12,), (21,), (31,)], "consecutive_days": [(11, '2026-10-01', '2026-10-03', 3), (21, '2026-10-02', '2026-10-04', 3)], "retention": [(1, 2, 1, 0.5, 2, 1.0), (2, 2, 2, 1.0, 0, 0.0), (3, 3, 1, 0.3333, 1, 0.3333), (4, 1, 0, 0.0, None, None), (5, 1, 0, 0.0, None, None), (6, 0, 0, None, None, None), (7, 0, 0, None, None, None), (8, 2, 0, 0.0, None, None), (9, 0, 0, None, None, None), (10, 1, None, None, None, None)], "monthly_change": [(1, 100, None, None, None, None, None, None), (2, 0, 100, None, 0.0, -1.0, None, None), (3, 50, 0, None, None, None, None, None), (4, 0, 50, None, 0.0, -1.0, None, None), (5, 0, 0, None, None, None, None, None), (6, 0, 0, None, None, None, None, None), (7, 0, 0, None, None, None, None, None), (8, 0, 0, None, None, None, None, None), (9, 0, 0, None, None, None, None, None), (10, 0, 0, None, None, None, None, None), (11, 0, 0, None, None, None, None, None), (12, 0, 0, None, None, None, None, None), (13, 200, 0, 100, None, None, 2.0, 1.0), (14, 0, 200, 0, 0.0, -1.0, None, None)], "top_three_rows": [(1, 102, 9000, 1), (1, 101, 9000, 2), (1, 104, 7000, 3), (2, 201, 4000, 1), (2, 202, 2000, 2), (3, None, None, None)], "top_three_prices": [(1, 102, 9000, 1), (1, 101, 9000, 1), (1, 104, 7000, 2), (1, 103, 7000, 2), (1, 105, 5000, 3), (2, 201, 4000, 1), (2, 202, 2000, 2), (3, None, None, None)], "above_average": [(1, 101, 9000), (1, 102, 9000), (2, 201, 4000)], "latest_paid": [(11, 102), (12, 105), (21, 202)], "pivot_counts": [(1, 5, 1, 1), (2, 2, 0, 0), (3, 0, 0, 0)], "unpivot_counts": [(1, 'cancel', 1), (1, 'paid', 5), (1, 'refund', 1), (2, 'cancel', 0), (2, 'paid', 2), (2, 'refund', 0), (3, 'cancel', 0), (3, 'paid', 0), (3, 'refund', 0)], "join_counts": [(1, 5, 5, 37000), (2, 2, 2, 6000), (3, 1, 0, None)], } SQL_PATH = Path(__file__).with_name("practice.sql") def load_fixture(sql_path=SQL_PATH): """Return the published setup SQL and ordered named SELECT statements.""" parts = re.split(r"^-- query: ([a-z_]+)\s*$", Path(sql_path).read_text(), flags=re.MULTILINE) queries = dict(zip(parts[1::2], parts[2::2])) if len(queries) != (len(parts) - 1) // 2: raise ValueError("duplicate query names") if set(queries) != set(EXPECTED): raise ValueError(f"query set differs: SQL={sorted(queries)}, expected={sorted(EXPECTED)}") return parts[0], queries def _check(name, rows): # MySQL returns Decimal; SQLite returns float for ROUND(..., 4). # Both are normalized to the exact published four-decimal output, with no # tolerance that could conceal an unrounded or incorrect query result. actual = [tuple(float(value) if isinstance(value, Decimal) else value for value in row) for row in rows] expected = EXPECTED[name] if actual != expected: raise AssertionError(f"{name}: expected {expected!r}; actual {actual!r}") return {"name": name, "rows": len(actual), "status": "PASS"} def verify_sqlite(sql_path=SQL_PATH): """Execute the fixture on a real in-memory SQLite connection.""" setup, queries = load_fixture(sql_path) with sqlite3.connect(":memory:") as connection: connection.executescript(setup) checks = [_check(name, connection.execute(queries[name]).fetchall()) for name in EXPECTED] return {"engine": "SQLite", "version": sqlite3.sqlite_version, "checks": checks, "total": len(checks), "scope": "fixture relational/window results only"} def verify_mysql(connection, sql_path=SQL_PATH): """Execute in a caller-created disposable schema; never create/drop a schema. Requires a DB-API connection with a tuple-returning cursor, such as PyMySQL. The caller owns CREATE DATABASE (without IF NOT EXISTS) and finally cleanup. """ setup, queries = load_fixture(sql_path) with connection.cursor() as cursor: cursor.execute("SELECT VERSION(), DATABASE()") version, schema = cursor.fetchone() if not schema or not re.fullmatch(r"study_t16_\d{8}_[0-9a-f]{8}_practice", schema): raise ValueError("MySQL fixture requires an isolated study_t16_YYYYMMDD_8hex_practice schema") statement = "" for character in setup: statement += character if character == ";" and sqlite3.complete_statement(statement): cursor.execute(statement) statement = "" if statement.strip(): raise ValueError("incomplete fixture setup SQL") checks = [] for name in EXPECTED: cursor.execute(queries[name]) checks.append(_check(name, cursor.fetchall())) return {"engine": "MySQL", "version": str(version), "schema": schema, "checks": checks, "total": len(checks), "scope": "fixture results on the reported MySQL version"} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--sql", type=Path, default=SQL_PATH, help="SQL input (defaults to the published practice.sql)") args = parser.parse_args() try: result = verify_sqlite(args.sql) for check in result["checks"]: print(f"PASS {check['name']} ({check['rows']} rows)") print(f"PASS {result['total']} checks; SQLite {result['version']} semantics only; " "MySQL 8.0.39 execution and optimizer plans NOT verified") except (OSError, ValueError, sqlite3.Error, AssertionError) as error: print(f"FAIL {error}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())