#!/usr/bin/env python3 """ Python-based database setup for E2E tests. Replaces: go run ./cmd/setup_db && go run ./cmd/setup apply This script handles: 1. Database creation (if not exists) 2. Schema bootstrap (creates all 94 tables from system_tables.json) 3. Bootstrap state management 4. ALTER TABLE statements for additional columns Note: The 'go run ./cmd/setup apply' command (seed data) is still needed and should be run separately after this script completes. """ import json import os import sys import time import re try: import mysql.connector except ImportError: import subprocess subprocess.run([sys.executable, "-m", "pip", "install", "mysql-connector-python", "-q"], check=True) import mysql.connector def get_tidb_config(): """Get TiDB connection config from environment or /tmp/tidb_instance.json""" host = os.environ.get("TIDB_HOST", "127.0.0.1") port = int(os.environ.get("TIDB_PORT", "4000")) user = os.environ.get("TIDB_USER", "root") password = os.environ.get("TIDB_PASSWORD", "") database = os.environ.get("TIDB_DATABASE", "nexuscrm") return host, port, user, password, database def connect_tidb(with_db=True): """Connect to TiDB with automatic TLS for remote instances""" host, port, user, password, database = get_tidb_config() # mysql-connector-python auto-negotiates TLS with TiDB Cloud serverless config = { 'host': host, 'port': port, 'user': user, 'password': password, } if with_db and database: config['database'] = database return mysql.connector.connect(**config) def create_database(): """Create the database if it doesn't exist""" host, port, user, password, database = get_tidb_config() print(f"Connecting to {host}:{port} as {user}...") conn = mysql.connector.connect(host=host, port=port, user=user, password=password) cursor = conn.cursor() print(f"Creating database '{database}' if not exists...") cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{database}`") conn.commit() cursor.close() conn.close() print(f"Database '{database}' ready.") def load_system_tables(): """Load system_tables.json""" # Try multiple paths paths = [ 'backend/internal/bootstrap/system_tables.json', '/home/openclaw/github/bingwang-MetaDrivenPlatform4AI/MetaDrivenPlatform4AI/backend/internal/bootstrap/system_tables.json', ] for path in paths: try: with open(path) as f: tables = json.load(f) print(f"Loaded {len(tables)} table definitions from {path}") return tables except FileNotFoundError: continue raise FileNotFoundError(f"Could not find system_tables.json in any of: {paths}") def is_numeric_type(sql_type): """Check if a SQL type is numeric""" sql_type_upper = sql_type.upper() numeric_prefixes = ('INT', 'TINYINT', 'SMALLINT', 'MEDIUMINT', 'BIGINT', 'FLOAT', 'DOUBLE', 'DECIMAL', 'NUMERIC', 'BIT') return any(sql_type_upper.startswith(p) for p in numeric_prefixes) def format_default_value(default, sql_type): """ Format default value properly for MySQL. - Numeric types: no quotes (e.g., 0, 1) - String types: quoted (e.g., 'value') - Already quoted strings: keep as-is """ if default is None: return None # If already quoted (starts and ends with single quotes), keep as-is if isinstance(default, str) and default.startswith("'") and default.endswith("'"): return default # If numeric type, no quotes if is_numeric_type(sql_type): return str(default) # Otherwise, add quotes return f"'{default}'" def generate_create_table_ddl(table_def): """Generate CREATE TABLE DDL from table definition""" columns = [] primary_keys = [] for col in table_def.get('columns', []): col_name = col.get('name', '') col_type = col.get('type', 'VARCHAR(255)') nullable = col.get('nullable', False) default = col.get('default') unique = col.get('unique', False) auto_inc = col.get('autoIncrement', False) primary = col.get('primaryKey', False) parts = [f"`{col_name}`", col_type] if not nullable: parts.append('NOT NULL') if default is not None: formatted_default = format_default_value(default, col_type) parts.append(f"DEFAULT {formatted_default}") if unique and not primary: parts.append('UNIQUE') if auto_inc: parts.append('AUTO_INCREMENT') columns.append(' '.join(parts)) if primary: primary_keys.append(col_name) # Add PRIMARY KEY if we have one if primary_keys: columns.append(f"PRIMARY KEY ({', '.join(['`' + pk + '`' for pk in primary_keys])})") table_name = table_def['tableName'] ddl = f"CREATE TABLE IF NOT EXISTS `{table_name}` (\n " + ',\n '.join(columns) + "\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin" return ddl def ensure_bootstrap_state_table(conn): """Create the bootstrap state table""" ddl = """CREATE TABLE IF NOT EXISTS `_System_BootstrapState` ( `StateKey` VARCHAR(128) NOT NULL PRIMARY KEY, `RunID` VARCHAR(64) NOT NULL, `Phase` VARCHAR(128) NOT NULL, `Status` VARCHAR(32) NOT NULL, `LastError` TEXT NULL, `CreatedDate` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `LastModifiedDate` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin""" cursor = conn.cursor() cursor.execute(ddl) conn.commit() cursor.close() print("Bootstrap state table ready.") def mark_bootstrap_complete(conn, phase="completed"): """Mark bootstrap as complete""" cursor = conn.cursor() run_id = f"python-bootstrap-{int(time.time())}" cursor.execute(""" INSERT INTO `_System_BootstrapState` (`StateKey`, `RunID`, `Phase`, `Status`, `LastError`, `CreatedDate`, `LastModifiedDate`) VALUES ('schema_initialize', %s, %s, 'completed', NULL, NOW(), NOW()) ON DUPLICATE KEY UPDATE `RunID`=VALUES(`RunID`), `Phase`=VALUES(`Phase`), `Status`=VALUES(`Status`), `LastModifiedDate`=NOW() """, (run_id, phase)) conn.commit() cursor.close() def run_alter_statements(conn): """Run ALTER TABLE statements for additional columns""" alter_statements = [ "ALTER TABLE `_System_Object` ADD COLUMN IF NOT EXISTS `DeleteMode` VARCHAR(50) NULL", "ALTER TABLE `_System_Session` ADD COLUMN IF NOT EXISTS `GrantType` VARCHAR(50) NULL", ] cursor = conn.cursor() for ddl in alter_statements: try: cursor.execute(ddl) conn.commit() print(f" Executed: {ddl[:60]}...") except mysql.connector.Error as e: # Ignore "duplicate column" errors if "Duplicate column" in str(e): print(f" (Already exists, skipping): {ddl[:60]}...") else: print(f" Warning: {e}") cursor.close() def bootstrap_schema(conn): """Create all tables from system_tables.json""" tables = load_system_tables() cursor = conn.cursor() created = 0 errors = 0 print(f"Creating {len(tables)} tables...") for table_def in tables: table_name = table_def['tableName'] ddl = generate_create_table_ddl(table_def) try: cursor.execute(ddl) created += 1 if created % 20 == 0: print(f" Created {created}/{len(tables)} tables...") except mysql.connector.Error as e: print(f" Error creating {table_name}: {e}") errors += 1 conn.commit() cursor.close() print(f"Schema bootstrap complete: {created} created, {errors} errors.") def main(): print("=" * 60) print("Python Database Setup for E2E Tests") print("=" * 60) # Step 1: Create database print("\n[1/5] Creating database...") create_database() # Step 2: Connect to database print("\n[2/5] Connecting to database...") conn = connect_tidb(with_db=True) print("Connected!") # Step 3: Bootstrap schema print("\n[3/5] Bootstrapping schema...") bootstrap_schema(conn) # Step 4: Ensure bootstrap state table print("\n[4/5] Setting up bootstrap state...") ensure_bootstrap_state_table(conn) mark_bootstrap_complete(conn) # Step 5: Run ALTER statements print("\n[5/5] Running ALTER TABLE statements...") run_alter_statements(conn) conn.close() print("\n" + "=" * 60) print("Database setup complete!") print("=" * 60) if __name__ == "__main__": main()