IRIS Legacy

#!/usr/bin/env python3 """ Test IRIS Database Models Test basic CRUD operations """ import sys import os from datetime import datetime import uuid # Add src to path sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src')) from src.database import get_db_session, User, Session, Message, File, SystemLog, DatabaseManager import logging # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def test_user_operations(): """Test User model CRUD operations""" logger.info("🧪 Testing User operations...") with get_db_session() as db: # Create test user test_user = User( telegram_id=123456789, username="test_user", first_name="Test", last_name="User", language_code="en", preferences={"theme": "dark", "notifications": True} ) db.add(test_user) db.flush() # Get the ID logger.info(f"✅ Created user: {test_user}") # Read user found_user = db.query(User).filter(User.telegram_id == 123456789).first() assert found_user is not None assert found_user.username == "test_user" logger.info(f"✅ Found user: {found_user}") # Update user found_user.last_activity = datetime.utcnow() found_user.preferences = {"theme": "light", "notifications": False} # Test session creation test_session = Session( user_id=found_user.id, session_id=str(uuid.uuid4()), context={"conversation_started": True}, title="Test Conversation" ) db.add(test_session) db.flush() logger.info(f"✅ Created session: {test_session}") # Test message creation test_message = Message( session_id=test_session.id, role="user", content="Hello, this is a test message!", message_type="text" ) db.add(test_message) logger.info(f"✅ Created message: {test_message}") # Test file creation test_file = File( user_id=found_user.id, file_id=str(uuid.uuid4()), filename="test_document.pdf", original_filename="My Document.pdf", file_type="pdf", mime_type="application/pdf", file_size=1024000, analysis_status="pending", tags=["important", "test"] ) db.add(test_file) logger.info(f"✅ Created file: {test_file}") # Test system log test_log = SystemLog( level="INFO", message="Test log entry", component="test_script", user_id=found_user.id, log_metadata={"test": True} ) db.add(test_log) logger.info(f"✅ Created log: {test_log}") return found_user.id def test_relationships(): """Test model relationships""" logger.info("🔗 Testing relationships...") with get_db_session() as db: # Get user with relationships user = db.query(User).filter(User.telegram_id == 123456789).first() # Test user -> sessions relationship sessions = user.sessions logger.info(f"✅ User has {len(sessions)} sessions") # Test user -> files relationship files = user.files logger.info(f"✅ User has {len(files)} files") # Test session -> messages relationship if sessions: messages = sessions[0].messages logger.info(f"✅ Session has {len(messages)} messages") def test_database_stats(): """Test database statistics""" logger.info("📊 Testing database statistics...") stats = DatabaseManager.get_stats() logger.info(f"✅ Database stats: {stats}") assert stats["users"] > 0 assert stats["sessions"] > 0 assert stats["messages"] > 0 assert stats["files"] > 0 assert stats["logs"] > 0 def cleanup_test_data(): """Clean up test data""" logger.info("🧹 Cleaning up test data...") with get_db_session() as db: # Delete test user (need to delete logs first due to FK constraint) test_user = db.query(User).filter(User.telegram_id == 123456789).first() if test_user: # Delete system logs first db.query(SystemLog).filter(SystemLog.user_id == test_user.id).delete() # Now delete user (cascade will handle sessions, messages, files) db.delete(test_user) logger.info("✅ Test data cleaned up") def main(): """Run all tests""" try: logger.info("🚀 Starting IRIS Database Tests...") # Test database health if not DatabaseManager.health_check(): logger.error("❌ Database health check failed!") return False logger.info("✅ Database health check passed") # Run tests user_id = test_user_operations() test_relationships() test_database_stats() # Cleanup cleanup_test_data() logger.info("🎉 All database tests passed!") return True except Exception as e: logger.error(f"❌ Database tests failed: {e}") return False if __name__ == "__main__": success = main() sys.exit(0 if success else 1)