""" Database checks module """ import os import sys import subprocess import difflib import re import sqlparse from pathlib import Path from tests.utils.logger import Logger from tests.utils.schema_analyzer import SchemaAnalyzer from tests.utils.docker_utils import DockerUtils from tests.utils.model_schema_analyzer import ModelSchemaAnalyzer from tests.utils.db_constants import INDEX_DEFINITIONS, TRIGGER_DEFINITIONS from typing import Dict, List, Optional class DatabaseChecks: """Check database schema and connection""" def __init__(self, logger: Logger, project_dir: str, version: str = "unknown"): self.logger = logger self.version = version self.project_dir = project_dir self.backend_dir = os.path.join(project_dir, "backend") self.schema_analyzer = SchemaAnalyzer(project_dir) self.docker_utils = DockerUtils(project_dir) self.model_schema_analyzer = ModelSchemaAnalyzer(project_dir) # Add backend to path for imports if self.backend_dir not in sys.path: sys.path.insert(0, self.backend_dir) def _get_expected_tables_from_models(self) -> set: """Dynamically extract table names from SQLAlchemy models""" expected_tables = set() models_dir = Path(self.backend_dir) / "app" / "models" if not models_dir.exists(): # Fallback to hardcoded list if models dir doesn't exist return { "users", "sessions", "streams", "comments", "tags", "stream_tags", "processing_jobs", "activity_log", "chat_messages", "chapters", "watch_history", "twitch_oauth", "site_settings" } try: for model_file in models_dir.glob("*.py"): if model_file.name == "__init__.py": continue content = model_file.read_text(encoding='utf-8') # Look for __tablename__ = "table_name" tablename_matches = re.findall(r'__tablename__\s*=\s*["\']([^"\']+)["\']', content) expected_tables.update(tablename_matches) # Also look for association tables in Table definitions table_matches = re.findall(r'\b(\w+)\s*=\s*Table\s*\(', content) expected_tables.update(table_matches) except Exception as e: self.logger.warning("DATABASE", f"Failed to scan models dynamically: {e}") # Fallback to hardcoded list return { "users", "sessions", "streams", "comments", "tags", "stream_tags", "processing_jobs", "activity_log", "chat_messages", "chapters", "watch_history", "twitch_oauth", "site_settings" } return expected_tables def run_all_checks(self, auto_fix: bool = True) -> bool: """Run all database checks and optionally fix issues""" self.logger.subheader(f"Database Checks (v{self.version})") # Check database connection if not self._check_database_connection(): return False # Compare schema with models if not self._compare_schema_with_models(auto_fix): return False # Check indexes if not self._check_indexes(auto_fix): return False # Check triggers if not self._check_triggers(auto_fix): return False # Check initial data if not self._check_initial_data(): return False # Check migration system if not self.check_migration_system(): return False # Check model-schema coverage (new) if not self._check_model_schema_coverage(): return False # Check migration consistency (new) if not self._check_migration_consistency(auto_fix): return False # Check all model tables exist (new) if not self._check_all_model_tables_exist(): return False self.logger.success("DATABASE", "All database checks passed") return True def _check_database_connection(self) -> bool: """Check if database connection works using docker exec""" self.logger.info("DATABASE", "Checking database connection...") try: import subprocess # Use docker exec to check database connection result = subprocess.run( ["docker", "exec", "streama2026-postgres", "pg_isready", "-U", "streama", "-d", "streama2026"], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: self.logger.success("DATABASE", "Database connection successful") return True else: self.logger.error("DATABASE", f"Database connection failed: {result.stderr}") return False except subprocess.TimeoutExpired: self.logger.error("DATABASE", "Database connection check timed out") return False except Exception as e: self.logger.error("DATABASE", f"Database connection check failed: {e}") return False def _compare_schema_with_models(self, auto_fix: bool) -> bool: """Compare database tables with expected tables using docker exec""" self.logger.info("DATABASE", "Comparing tables with models...") try: import subprocess # Use docker exec to get list of tables from database result = subprocess.run( ["docker", "exec", "streama2026-postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", "\\dt"], capture_output=True, text=True, timeout=10 ) if result.returncode != 0: self.logger.error("DATABASE", f"Failed to get table list: {result.stderr}") return False # Parse table list from psql output db_tables = set() for line in result.stdout.split('\n'): if line.strip() and not line.startswith('|') and 'public' in line: parts = line.split('|') if len(parts) >= 2: table_name = parts[1].strip() if table_name and table_name != 'name': db_tables.add(table_name) # Expected tables from models (dynamically extracted) model_tables = self._get_expected_tables_from_models() all_ok = True # Check for missing tables missing_tables = model_tables - db_tables if missing_tables: self.logger.error("DATABASE", f"Missing tables: {', '.join(missing_tables)}") all_ok = False if auto_fix: self.logger.info("DATABASE", "Attempting to fix missing tables...") if self._fix_missing_tables(missing_tables): self.logger.success("DATABASE", "Missing tables fixed") else: self.logger.error("DATABASE", "Failed to fix missing tables") # Check for extra tables extra_tables = db_tables - model_tables if extra_tables: # alembic_version and flyway_schema_history are expected for migrations if extra_tables == {"alembic_version"}: self.logger.info("DATABASE", f"Extra table found: alembic_version (expected for migrations)") elif extra_tables == {"flyway_schema_history"}: self.logger.info("DATABASE", f"Extra table found: flyway_schema_history (expected for migrations)") elif extra_tables == {"alembic_version", "flyway_schema_history"}: self.logger.info("DATABASE", f"Extra tables found: alembic_version, flyway_schema_history (expected for migrations)") else: self.logger.warning("DATABASE", f"Extra tables found: {', '.join(extra_tables)}") # Check each expected table for table_name in model_tables: if table_name in db_tables: self.logger.success("DATABASE", f"Table {table_name}: exists") else: self.logger.error("DATABASE", f"Table {table_name}: missing") all_ok = False # Check columns for each existing table using schema analyzer if db_tables: models_info = self.schema_analyzer.analyze_models() self.logger.info("DATABASE", f"Found {len(models_info)} models in backend/app/models") self.logger.info("DATABASE", "Schema comparison with init-db.sql skipped (Flyway is now responsible for migrations)") return all_ok except subprocess.TimeoutExpired: self.logger.error("DATABASE", "Schema comparison timed out") return False except Exception as e: self.logger.error("DATABASE", f"Schema comparison failed: {e}") return False def _auto_create_and_apply_migration(self) -> bool: """Automatically create and apply Flyway migration""" # Flyway is now responsible for database migrations self.logger.info("DATABASE", "Skipping auto-migration (Flyway is now responsible for migrations)") return True def _check_indexes(self, auto_fix: bool = False) -> bool: """Check if required indexes exist using docker exec""" self.logger.info("DATABASE", "Checking indexes...") try: import subprocess # Use docker exec to get list of indexes from database result = subprocess.run( ["docker", "exec", "streama2026-postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", "\\di"], capture_output=True, text=True, timeout=10 ) if result.returncode != 0: self.logger.warning("DATABASE", f"Failed to get index list: {result.stderr}") return True # Not critical, continue with tests # Parse index list from psql output db_indexes = set() for line in result.stdout.split('\n'): if line.strip() and ('idx_' in line or 'ix_' in line): parts = line.split('|') if len(parts) >= 2: index_name = parts[1].strip() if index_name and index_name != 'name': db_indexes.add(index_name) # Expected indexes (using idx_ prefix as defined in V1__baseline.sql) expected_indexes = { "idx_sessions_user_id", "idx_sessions_expires_at", "idx_streams_status", "idx_streams_created_by", "idx_comments_stream_id", "idx_comments_user_id", "idx_comments_parent_id", "idx_processing_jobs_stream_id", "idx_processing_jobs_status", "idx_activity_log_user_id", "idx_activity_log_created_at", "idx_watch_history_user_id", "idx_watch_history_stream_id", "idx_watch_history_last_watched_at", "idx_twitch_oauth_user_id", "idx_chapters_stream_id", "idx_chapters_game_display_name", "idx_chat_messages_stream_id", "idx_site_settings_key" } all_ok = True for index_name in expected_indexes: if index_name in db_indexes: self.logger.success("DATABASE", f"Index {index_name}: exists") else: self.logger.info("DATABASE", f"Index {index_name}: missing (may be in migration)") # Not critical - indexes can be added via migration missing_indexes = set(expected_indexes) - set(db_indexes) if missing_indexes: self.logger.info("DATABASE", f"Missing indexes: {', '.join(missing_indexes)} (may be in migration)") # Auto-fix missing indexes if auto_fix and missing_indexes: self.logger.info("DATABASE", "Auto-fix enabled - creating migration for missing indexes...") if not self._auto_create_index_migration(list(missing_indexes)): self.logger.warning("DATABASE", "Failed to create index migration - indexes may be added manually") # Don't return False - indexes are not critical return True except subprocess.TimeoutExpired: self.logger.warning("DATABASE", "Index check timed out") return True # Not critical except Exception as e: self.logger.warning("DATABASE", f"Index check failed: {e}") return True # Not critical def _check_triggers(self, auto_fix: bool = False) -> bool: """Check if required triggers exist using docker exec""" self.logger.info("DATABASE", "Checking triggers...") try: import subprocess # Use docker exec to get list of triggers from database result = subprocess.run( ["docker", "exec", "streama2026-postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", "SELECT trigger_name FROM information_schema.triggers WHERE trigger_schema = 'public';"], capture_output=True, text=True, ) db_triggers = set() if result.returncode == 0: # Parse output more carefully lines = result.stdout.strip().split('\n') for line in lines: # Skip header lines, empty lines, and summary lines like (5 rows) if line.strip() and not line.startswith('-') and not line.startswith('trigger_name') and '(' not in line and '_' in line: trigger_name = line.strip() if trigger_name: db_triggers.add(trigger_name) expected_triggers = [ "update_users_updated_at", "update_streams_updated_at", "update_comments_updated_at", "update_twitch_oauth_updated_at", "update_site_settings_updated_at" ] all_ok = True missing_triggers = [] for trigger_name in expected_triggers: if trigger_name in db_triggers: self.logger.success("DATABASE", f"Trigger {trigger_name}: exists ✓") else: self.logger.info("DATABASE", f"Trigger {trigger_name}: missing (may be in migration)") all_ok = False missing_triggers.append(trigger_name) if all_ok: self.logger.success("DATABASE", "All triggers exist ✓") else: self.logger.error("DATABASE", "Some triggers missing") # Auto-fix: Create migration for missing triggers if auto_fix and missing_triggers: self.logger.info("DATABASE", "Attempting to auto-create migration for missing triggers...") if self._auto_create_trigger_migration(missing_triggers): self.logger.success("DATABASE", "Trigger migration created and applied successfully") all_ok = True else: self.logger.error("DATABASE", "Failed to create/apply trigger migration") return all_ok except Exception as e: self.logger.error("DATABASE", f"Failed to check triggers: {e}") return False def _auto_create_and_apply_triggers_migration(self) -> bool: """Automatically create and apply triggers migration""" # Skip Alembic migrations - Flyway is now responsible for database migrations self.logger.info("DATABASE", "Skipping Alembic triggers migration (Flyway is now responsible for migrations)") return True def _auto_create_index_migration(self, missing_indexes: list) -> bool: """Automatically create and apply migration for missing indexes using docker exec""" # Skip Alembic migrations - Flyway is now responsible for database migrations self.logger.info("DATABASE", "Skipping Alembic indexes migration (Flyway is now responsible for migrations)") return True def _check_initial_data(self) -> bool: """Check if initial data exists using docker exec""" self.logger.info("DATABASE", "Checking initial data...") try: import subprocess # Use docker exec to check site settings count result = subprocess.run( ["docker", "exec", "streama2026-postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", "SELECT COUNT(*) FROM site_settings;"], capture_output=True, text=True, ) if result.returncode == 0: # Parse output to get count (skip header and separator) lines = result.stdout.strip().split('\n') if len(lines) >= 3: try: site_settings_count = int(lines[2].strip()) if site_settings_count == 0: self.logger.error("DATABASE", "Site settings table is empty - missing initial data") return False elif site_settings_count >= 5: self.logger.success("DATABASE", f"Site settings: {site_settings_count} rows ✓") return True else: self.logger.warning("DATABASE", f"Site settings: only {site_settings_count} rows (expected at least 5)") return False except ValueError: self.logger.error("DATABASE", "Failed to parse site settings count") return False else: self.logger.error("DATABASE", f"Failed to check initial data: {result.stderr}") return False except Exception as e: self.logger.error("DATABASE", f"Failed to check initial data: {e}") return False # Not critical def check_migration_system(self) -> bool: """Check if Flyway migration system is set up""" self.logger.info("DATABASE", "Checking migration system (Flyway)...") # Check if Flyway SQL directory exists try: from pathlib import Path project_root = Path(self.project_dir) flyway_dir = project_root / "flyway" / "sql" if not flyway_dir.exists(): self.logger.warning("DATABASE", "Flyway SQL directory not found") return True # Not critical # Check for baseline migration baseline = flyway_dir / "V1__baseline.sql" if not baseline.exists(): self.logger.warning("DATABASE", "Flyway baseline migration not found") return True # Not critical self.logger.success("DATABASE", "Flyway migration system is set up") return True except Exception as e: self.logger.error("DATABASE", f"Failed to check Flyway migration system: {e}") return True # Not critical def _check_model_schema_coverage(self) -> bool: """Check model-schema coverage using ModelSchemaAnalyzer""" self.logger.info("DATABASE", "Checking model-schema coverage...") try: coverage_report = self.model_schema_analyzer.compare_model_schema_coverage() self.logger.info("DATABASE", f"Model-Schema coverage: {coverage_report['coverage_percentage']}%") self.logger.info("DATABASE", f"Total model fields: {coverage_report['total_model_fields']}") self.logger.info("DATABASE", f"Covered fields: {coverage_report['covered_fields']}") if coverage_report["missing_from_schemas"]: missing_count = sum(len(fields) for fields in coverage_report["missing_from_schemas"].values()) self.logger.warning("DATABASE", f"Missing from schemas: {missing_count} fields") for model_name, fields in coverage_report["missing_from_schemas"].items(): self.logger.info("DATABASE", f" {model_name}: {', '.join(f['name'] for f in fields)}") if coverage_report["extra_in_schemas"]: extra_count = sum(len(fields) for fields in coverage_report["extra_in_schemas"].values()) self.logger.info("DATABASE", f"Extra in schemas: {extra_count} fields (may be intentional)") # Check for type mismatches if coverage_report.get("type_mismatches"): type_mismatch_count = sum(len(mismatches) for mismatches in coverage_report["type_mismatches"].values()) if type_mismatch_count > 0: self.logger.warning("DATABASE", f"Type mismatches found: {type_mismatch_count}") for table_name, mismatches in coverage_report["type_mismatches"].items(): self.logger.info("DATABASE", f" {table_name}: {', '.join(mismatches)}") # Not blocking - just informational return True except Exception as e: self.logger.error("DATABASE", f"Failed to check model-schema coverage: {e}") return True # Not blocking def _check_migration_consistency(self, auto_fix: bool = False) -> bool: """Check migration consistency using Flyway""" self.logger.info("DATABASE", "Checking migration consistency...") try: # Flyway manages migrations, so we just check if Flyway SQL directory exists flyway_dir = Path(self.project_dir) / "flyway" / "sql" if not flyway_dir.exists(): self.logger.warning("DATABASE", "Flyway SQL directory not found") return True # Not critical # Count migration files migration_files = list(flyway_dir.glob("*.sql")) if migration_files: self.logger.success("DATABASE", f"Flyway migrations found: {len(migration_files)} files") self.logger.info("DATABASE", "Flyway manages migration consistency automatically") else: self.logger.warning("DATABASE", "No Flyway migration files found") return True except Exception as e: self.logger.error("DATABASE", f"Failed to check migration consistency: {e}") return True # Not blocking def _check_all_model_tables_exist(self) -> bool: """Check if all model tables exist in database""" self.logger.info("DATABASE", "Checking all model tables exist...") try: import subprocess # Get list of tables from database result = subprocess.run( ["docker", "exec", "streama2026-postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", "\\dt"], capture_output=True, text=True, timeout=10 ) if result.returncode != 0: self.logger.error("DATABASE", f"Failed to get table list: {result.stderr}") return False # Parse table list db_tables = set() for line in result.stdout.split('\n'): if line.strip() and not line.startswith('|') and 'public' in line: parts = line.split('|') if len(parts) >= 2: table_name = parts[1].strip() if table_name and table_name != 'name': db_tables.add(table_name) # Expected tables from models model_tables = { "users", "sessions", "streams", "comments", "tags", "stream_tags", "processing_jobs", "activity_log", "chat_messages", "chapters", "watch_history", "twitch_oauth", "site_settings" } all_ok = True missing_tables = model_tables - db_tables if missing_tables: self.logger.error("DATABASE", f"Missing tables in database: {', '.join(missing_tables)}") all_ok = False else: self.logger.success("DATABASE", "All model tables exist in database") return all_ok except subprocess.TimeoutExpired: self.logger.error("DATABASE", "Table check timed out") return False except Exception as e: self.logger.error("DATABASE", f"Failed to check model tables: {e}") return False