test.py

"""
SQL Script Verification Tool

This script parses all 15 SQL chapters and executes every `SELECT` query block
against the local database to ensure syntax correctness.
It connects to PostgreSQL by default, falling back to SQLite if PG isn't available.
"""

import os
import glob
import re

try:
    import psycopg2
    HAS_PSYCOPG2 = True
except ImportError:
    HAS_PSYCOPG2 = False
    print("⚠️  psycopg2 not installed. Falling back to sqlite3 for testing ANSI SQL chapters.")

import sqlite3


PG_CONN_STR = "host=localhost dbname=ecommerce user=postgres password=masterclass port=5439"
SQLITE_DB = "ecommerce.db"


def extract_queries(filepath):
    """
    Extracts individual SQL queries from a file, ignoring comments.
    Splits by semicolon.
    """
    with open(filepath, 'r', encoding='utf-8') as f:
        content = f.read()

    # Remove multi-line comments
    content = re.sub(r'/\*.*?\*/', '', content, flags=re.DOTALL)
    # Remove single-line comments
    content = re.sub(r'--.*', '', content)

    # Instead of a simple split, iterate over characters to handle $$ dollar quotes
    queries = []
    current_query = []
    in_dollar_quote = False
    i = 0
    while i < len(content):
        char = content[i]
        
        # Check for start/end of dollar quote block
        if char == '$' and i + 1 < len(content) and content[i+1] == '$':
            in_dollar_quote = not in_dollar_quote
            current_query.append('$$')
            i += 2
            continue
            
        if char == ';' and not in_dollar_quote:
            if "".join(current_query).strip():
                queries.append("".join(current_query).strip())
            current_query = []
        else:
            current_query.append(char)
        i += 1
        
    if "".join(current_query).strip():
         queries.append("".join(current_query).strip())
         
    return queries


def verify_file(filepath, pg_conn, sqlite_conn):
    """Run all queries in a file and report failures."""
    filename = os.path.basename(filepath)
    queries = extract_queries(filepath)
    
    if not queries:
        return True, 0

    success_count = 0
    fail_count = 0
    
    # Chapters 13, 14, 15 require PostgreSQL specifically
    requires_pg = filename.startswith(('13', '14', '15'))
    
    conn_to_use = None
    if requires_pg:
        if not pg_conn:
            print(f"⏭️  Skipping {filename}: Requires PostgreSQL")
            return True, 0
        conn_to_use = pg_conn
        cursor = pg_conn.cursor()
    else:
        # Prefer PG, fallback to SQLite
        if pg_conn:
            conn_to_use = pg_conn
            cursor = pg_conn.cursor()
        elif sqlite_conn:
            conn_to_use = sqlite_conn
            cursor = sqlite_conn.cursor()
        else:
            print("❌ No database connection available.")
            return False, 0
            
    print(f"Testing {filename} ({len(queries)} queries)...")
    
    for i, query in enumerate(queries):
        # Skip pure DDL in verification unless it's the view creation chapter
        if query.upper().startswith(('CREATE SCHEMA', 'SET SEARCH_PATH')):
            continue

        try:
            cursor.execute(query)
            
            # Fetch and display some proof of execution
            if query.strip().upper().startswith(('SELECT', 'WITH', 'EXPLAIN')):
                try:
                    rows = cursor.fetchall()
                    row_count = len(rows)
                    
                    if i == 0 and row_count > 0 and cursor.description:
                        col_names = [desc[0] for desc in cursor.description]
                        sample = " | ".join([f"{c}: {str(v)[:20]}" for c, v in zip(col_names, rows[0])])
                        print(f"  ✓ Q{i+1} passed ({row_count} rows). Sample: {sample[:70]}...")
                    else:
                        print(f"  ✓ Q{i+1} passed ({row_count} rows)")
                except Exception as fetch_err:
                     print(f"  ✓ Q{i+1} passed (no rows to fetch)")
            else:
                print(f"  ✓ Q{i+1} passed (executed successfully)")
                
            success_count += 1
        except Exception as e:
            # We don't fail immediately, just log
            print(f"  ❌ Error in Query {i+1}:")
            print(f"     {str(e).strip()}")
            fail_count += 1
            if conn_to_use == pg_conn:
                pg_conn.rollback() # reset transaction block
            
    if conn_to_use == pg_conn:
        cursor.close()
        
    if fail_count > 0:
        print(f"⚠️  {filename}: {success_count} passed, {fail_count} failed.")
        return False, fail_count
    else:
        print(f"✅ {filename}: All {success_count} queries passed.")
        return True, 0


def main():
    pg_conn = None
    sqlite_conn = None
    
    if HAS_PSYCOPG2:
        try:
            pg_conn = psycopg2.connect(PG_CONN_STR)
            pg_conn.autocommit = True
            print("🔗 Connected to PostgreSQL for verification.")
        except Exception as e:
            print(f"⚠️  Could not connect to PostgreSQL: {e}")
            
    if not pg_conn and os.path.exists(SQLITE_DB):
        sqlite_conn = sqlite3.connect(SQLITE_DB)
        print("🔗 Connected to SQLite for verification (ANSI SQL only).")
        
    if not pg_conn and not sqlite_conn:
        print("❌ Cannot verify: No database connections available.")
        print("   Run 'make up' and 'make init-db' OR 'make init-sqlite' first.")
        return
        
    sql_files = sorted(glob.glob("*.sql"))
    if not sql_files:
        print("❌ No .sql files found.")
        return
        
    print("\n" + "="*50)
    print("🚀 Starting SQL Verification")
    print("="*50 + "\n")
    
    total_files = 0
    failed_files = 0
    total_failures = 0
    
    for filepath in sql_files:
        total_files += 1
        passed, failures = verify_file(filepath, pg_conn, sqlite_conn)
        if not passed:
            failed_files += 1
            total_failures += failures
            
    print("\n" + "="*50)
    if failed_files == 0:
        print("🎉 ALL VERIFICATIONS PASSED!")
    else:
        print(f"⚠️  VERIFICATION FAILED: {failed_files}/{total_files} files had errors.")
        print(f"   Total failing queries: {total_failures}")
    print("="*50 + "\n")
    
    if pg_conn:
        pg_conn.close()
    if sqlite_conn:
        sqlite_conn.close()


if __name__ == "__main__":
    main()