TimeTracker/docker/verify-database.py

#!/usr/bin/env python3
"""
Database verification script for TimeTracker
This script thoroughly checks the database schema and reports any issues
"""

import os
import sys
from sqlalchemy import create_engine, text, inspect
from sqlalchemy.exc import OperationalError, ProgrammingError

def get_expected_schema():
    """Define the complete expected database schema"""
    return {
        'users': {
            'columns': ['id', 'username', 'role', 'created_at', 'last_login', 'is_active', 'updated_at'],
            'required_columns': ['id', 'username', 'role', 'created_at', 'is_active', 'updated_at'],
            'indexes': ['idx_users_username', 'idx_users_role'],
            'foreign_keys': []
        },
        'projects': {
            'columns': ['id', 'name', 'client', 'description', 'billable', 'hourly_rate', 'billing_ref', 'status', 'created_at', 'updated_at'],
            'required_columns': ['id', 'name', 'client', 'billable', 'status', 'created_at', 'updated_at'],
            'indexes': ['idx_projects_client', 'idx_projects_status'],
            'foreign_keys': []
        },
        'time_entries': {
            'columns': ['id', 'user_id', 'project_id', 'task_id', 'start_time', 'end_time', 'duration_seconds', 'notes', 'tags', 'source', 'billable', 'created_at', 'updated_at'],
            'required_columns': ['id', 'user_id', 'project_id', 'start_time', 'source', 'billable', 'created_at', 'updated_at'],
            'indexes': ['idx_time_entries_user_id', 'idx_time_entries_project_id', 'idx_time_entries_task_id', 'idx_time_entries_start_time', 'idx_time_entries_billable'],
            'foreign_keys': ['user_id', 'project_id', 'task_id']
        },
        'tasks': {
            'columns': ['id', 'project_id', 'name', 'description', 'status', 'priority', 'assigned_to', 'created_by', 'due_date', 'estimated_hours', 'actual_hours', 'started_at', 'completed_at', 'created_at', 'updated_at'],
            'required_columns': ['id', 'project_id', 'name', 'status', 'priority', 'created_by', 'created_at', 'updated_at'],
            'indexes': ['idx_tasks_project_id', 'idx_tasks_status', 'idx_tasks_assigned_to', 'idx_tasks_due_date'],
            'foreign_keys': ['project_id', 'assigned_to', 'created_by']
        },
        'settings': {
            'columns': ['id', 'timezone', 'currency', 'rounding_minutes', 'single_active_timer', 'allow_self_register', 'idle_timeout_minutes', 'backup_retention_days', 'backup_time', 'export_delimiter', 'allow_analytics', 'company_name', 'company_address', 'company_email', 'company_phone', 'company_website', 'company_logo_filename', 'company_tax_id', 'company_bank_info', 'invoice_prefix', 'invoice_start_number', 'invoice_terms', 'invoice_notes', 'created_at', 'updated_at'],
            'required_columns': ['id', 'timezone', 'currency', 'rounding_minutes', 'single_active_timer', 'allow_self_register', 'idle_timeout_minutes', 'backup_retention_days', 'backup_time', 'export_delimiter', 'allow_analytics', 'company_name', 'company_address', 'company_email', 'company_phone', 'company_website', 'company_logo_filename', 'company_tax_id', 'company_bank_info', 'invoice_prefix', 'invoice_start_number', 'invoice_terms', 'invoice_notes', 'created_at', 'updated_at'],
            'indexes': [],
            'foreign_keys': []
        },
        'invoices': {
            'columns': ['id', 'invoice_number', 'project_id', 'client_name', 'client_email', 'client_address', 'issue_date', 'due_date', 'status', 'subtotal', 'tax_rate', 'tax_amount', 'total_amount', 'notes', 'terms', 'created_by', 'created_at', 'updated_at'],
            'required_columns': ['id', 'invoice_number', 'client_name', 'issue_date', 'due_date', 'status', 'subtotal', 'tax_rate', 'tax_amount', 'total_amount', 'created_at', 'updated_at'],
            'indexes': ['idx_invoices_project_id', 'idx_invoices_status', 'idx_invoices_issue_date'],
            'foreign_keys': ['project_id', 'created_by']
        },
        'invoice_items': {
            'columns': ['id', 'invoice_id', 'description', 'quantity', 'unit_price', 'total_amount', 'time_entry_ids', 'created_at'],
            'required_columns': ['id', 'invoice_id', 'description', 'quantity', 'unit_price', 'total_amount', 'created_at'],
            'indexes': ['idx_invoice_items_invoice_id'],
            'foreign_keys': ['invoice_id']
        }
    }

def check_table_exists(engine, table_name):
    """Check if a table exists"""
    try:
        inspector = inspect(engine)
        existing_tables = inspector.get_table_names()
        return table_name in existing_tables
    except Exception as e:
        print(f"Error checking if table {table_name} exists: {e}")
        return False

def check_table_columns(engine, table_name, expected_columns, required_columns):
    """Check if a table has the expected columns"""
    try:
        inspector = inspect(engine)
        existing_columns = [col['name'] for col in inspector.get_columns(table_name)]
        
        missing_columns = [col for col in expected_columns if col not in existing_columns]
        missing_required = [col for col in required_columns if col not in existing_columns]
        
        return {
            'exists': True,
            'all_columns': len(missing_columns) == 0,
            'required_columns': len(missing_required) == 0,
            'missing_columns': missing_columns,
            'missing_required': missing_required,
            'existing_columns': existing_columns
        }
    except Exception as e:
        print(f"Error checking columns for table {table_name}: {e}")
        return {
            'exists': False,
            'all_columns': False,
            'required_columns': False,
            'missing_columns': expected_columns,
            'missing_required': required_columns,
            'existing_columns': []
        }

def check_table_indexes(engine, table_name, expected_indexes):
    """Check if a table has the expected indexes"""
    try:
        inspector = inspect(engine)
        existing_indexes = [idx['name'] for idx in inspector.get_indexes(table_name)]
        
        missing_indexes = [idx for idx in expected_indexes if idx not in existing_indexes]
        
        return {
            'all_indexes': len(missing_indexes) == 0,
            'missing_indexes': missing_indexes,
            'existing_indexes': existing_indexes
        }
    except Exception as e:
        print(f"Error checking indexes for table {table_name}: {e}")
        return {
            'all_indexes': False,
            'missing_indexes': expected_indexes,
            'existing_indexes': []
        }

def check_foreign_keys(engine, table_name, expected_fks):
    """Check if a table has the expected foreign keys"""
    try:
        inspector = inspect(engine)
        existing_fks = [fk['constrained_columns'][0] for fk in inspector.get_foreign_keys(table_name)]
        
        missing_fks = [fk for fk in expected_fks if fk not in existing_fks]
        
        return {
            'all_fks': len(missing_fks) == 0,
            'missing_fks': missing_fks,
            'existing_fks': existing_fks
        }
    except Exception as e:
        print(f"Error checking foreign keys for table {table_name}: {e}")
        return {
            'all_fks': False,
            'missing_fks': expected_fks,
            'existing_fks': []
        }

def check_data_integrity(engine):
    """Check basic data integrity"""
    print("\n--- Checking Data Integrity ---")
    
    try:
        with engine.connect() as conn:
            # Check if admin user exists
            result = conn.execute(text("SELECT COUNT(*) FROM users WHERE role = 'admin'"))
            admin_count = result.scalar()
            print(f"Admin users: {admin_count}")
            
            # Check if default project exists
            result = conn.execute(text("SELECT COUNT(*) FROM projects WHERE name = 'General'"))
            project_count = result.scalar()
            print(f"Default projects: {project_count}")
            
            # Check if settings exist
            result = conn.execute(text("SELECT COUNT(*) FROM settings"))
            settings_count = result.scalar()
            print(f"Settings records: {settings_count}")
            
            # Check if allow_analytics column exists and has value
            try:
                result = conn.execute(text("SELECT allow_analytics FROM settings LIMIT 1"))
                analytics_setting = result.scalar()
                print(f"Analytics setting: {analytics_setting}")
            except Exception as e:
                print(f"Analytics setting check failed: {e}")
                
    except Exception as e:
        print(f"Error checking data integrity: {e}")

def main():
    """Main verification function"""
    url = os.getenv("DATABASE_URL", "")
    
    if not url.startswith("postgresql"):
        print("No PostgreSQL database configured, skipping verification")
        return
    
    print(f"Database URL: {url}")
    
    try:
        engine = create_engine(url, pool_pre_ping=True)
        
        # Test connection
        with engine.connect() as conn:
            conn.execute(text("SELECT 1"))
        print("✓ Database connection successful")
        
    except Exception as e:
        print(f"✗ Database connection failed: {e}")
        sys.exit(1)
    
    print("\n=== Starting Database Verification ===")
    
    expected_schema = get_expected_schema()
    verification_results = {}
    overall_status = True
    
    # Check each table
    for table_name, table_schema in expected_schema.items():
        print(f"\n--- Checking table: {table_name} ---")
        
        # Check if table exists
        table_exists = check_table_exists(engine, table_name)
        if not table_exists:
            print(f"✗ Table {table_name} does not exist")
            verification_results[table_name] = {'exists': False}
            overall_status = False
            continue
        
        print(f"✓ Table {table_name} exists")
        
        # Check columns
        column_check = check_table_columns(
            engine, table_name, 
            table_schema['columns'], 
            table_schema['required_columns']
        )
        
        if not column_check['required_columns']:
            print(f"✗ Table {table_name} missing required columns: {column_check['missing_required']}")
            overall_status = False
        elif not column_check['all_columns']:
            print(f"⚠ Table {table_name} missing optional columns: {column_check['missing_columns']}")
        else:
            print(f"✓ Table {table_name} has all expected columns")
        
        # Check indexes
        index_check = check_table_indexes(engine, table_name, table_schema['indexes'])
        if not index_check['all_indexes']:
            print(f"⚠ Table {table_name} missing indexes: {index_check['missing_indexes']}")
        
        # Check foreign keys
        fk_check = check_foreign_keys(engine, table_name, table_schema['foreign_keys'])
        if not fk_check['all_fks']:
            print(f"⚠ Table {table_name} missing foreign keys: {fk_check['missing_fks']}")
        
        verification_results[table_name] = {
            'exists': True,
            'columns': column_check,
            'indexes': index_check,
            'foreign_keys': fk_check
        }
    
    # Check data integrity
    check_data_integrity(engine)
    
    # Summary
    print("\n=== Verification Summary ===")
    if overall_status:
        print("✓ Database verification PASSED - All required tables and columns exist")
    else:
        print("✗ Database verification FAILED - Some required tables or columns are missing")
        print("\nIssues found:")
        for table_name, result in verification_results.items():
            if not result.get('exists', False):
                print(f"  - Table '{table_name}' is missing")
            elif not result.get('columns', {}).get('required_columns', False):
                print(f"  - Table '{table_name}' is missing required columns: {result['columns']['missing_required']}")
    
    return overall_status

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)