Upload 10 files

main.py

@@ -675,6 +675,243 @@ async def health_check():
         "database_type": getattr(app, "db_type", "unknown")
     }
 
+# Database test endpoint
+@app.get("/test-db", tags=["General"])
+async def test_database():
+    """
+    Test endpoint to verify database operations.
+    This is for debugging purposes only.
+    """
+    import time
+
+    # Generate a unique test ID
+    test_id = f"test_{int(time.time())}"
+    logger.info(f"[{test_id}] Starting database test")
+
+    results = {
+        "connection_type": getattr(app, "db_type", "unknown"),
+        "connection_object_type": type(app.db_conn).__name__ if hasattr(app, "db_conn") else "None",
+        "operations": []
+    }
+
+    if hasattr(app, "last_successful_connection_method"):
+        results["connection_method"] = app.last_successful_connection_method
+
+    if not hasattr(app, "db_conn"):
+        logger.error(f"[{test_id}] Database connection not available")
+        results["operations"].append({
+            "name": "Database connection check",
+            "success": False,
+            "error": "Database connection not available"
+        })
+        return results
+
+    try:
+        # Test 1: Simple SELECT
+        logger.info(f"[{test_id}] Test 1: Simple SELECT")
+        test_query = "SELECT 1 as test"
+        result = app.db_conn.execute(test_query).fetchone()
+        results["operations"].append({
+            "name": "Simple SELECT",
+            "success": result is not None,
+            "result": str(result) if result is not None else None
+        })
+        logger.info(f"[{test_id}] Test 1 result: {result}")
+
+        # Test 2: Create a temporary table
+        logger.info(f"[{test_id}] Test 2: Create a temporary table")
+        create_temp_table = """
+        CREATE TABLE IF NOT EXISTS test_table (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT NOT NULL,
+            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+        app.db_conn.execute(create_temp_table)
+        app.db_conn.commit()
+        results["operations"].append({
+            "name": "Create temporary table",
+            "success": True
+        })
+        logger.info(f"[{test_id}] Test 2 completed successfully")
+
+        # Test 3: Insert into the temporary table
+        logger.info(f"[{test_id}] Test 3: Insert into the temporary table")
+        test_name = f"test_user_{int(time.time())}"
+        insert_query = "INSERT INTO test_table (name) VALUES (?)"
+        cursor = app.db_conn.execute(insert_query, (test_name,))
+        app.db_conn.commit()
+
+        # Check if lastrowid is available
+        last_id = None
+        try:
+            last_id = cursor.lastrowid
+            logger.info(f"[{test_id}] Got lastrowid: {last_id}")
+        except Exception as e:
+            logger.warning(f"[{test_id}] Could not get lastrowid: {str(e)}")
+
+            # Try to get the ID using a query
+            try:
+                id_query = "SELECT id FROM test_table WHERE name = ? ORDER BY id DESC LIMIT 1"
+                id_result = app.db_conn.execute(id_query, (test_name,)).fetchone()
+                if id_result:
+                    last_id = id_result[0]
+                    logger.info(f"[{test_id}] Got ID from query: {last_id}")
+            except Exception as e2:
+                logger.error(f"[{test_id}] Error getting ID from query: {str(e2)}")
+
+        results["operations"].append({
+            "name": "Insert into temporary table",
+            "success": True,
+            "last_id": last_id
+        })
+        logger.info(f"[{test_id}] Test 3 completed successfully. Last ID: {last_id}")
+
+        # Test 4: Select from the temporary table
+        logger.info(f"[{test_id}] Test 4: Select from the temporary table")
+        select_query = "SELECT * FROM test_table WHERE name = ?"
+        result = app.db_conn.execute(select_query, (test_name,)).fetchone()
+        results["operations"].append({
+            "name": "Select from temporary table",
+            "success": result is not None,
+            "result": str(result) if result is not None else None
+        })
+        logger.info(f"[{test_id}] Test 4 result: {result}")
+
+        # Test 5: Check if users table exists and has the expected structure
+        logger.info(f"[{test_id}] Test 5: Check users table structure")
+        try:
+            table_info = app.db_conn.execute("PRAGMA table_info(users)").fetchall()
+            results["operations"].append({
+                "name": "Check users table structure",
+                "success": len(table_info) > 0,
+                "columns": [col[1] for col in table_info] if table_info else []
+            })
+            logger.info(f"[{test_id}] Test 5 result: {table_info}")
+        except Exception as e:
+            logger.error(f"[{test_id}] Error checking users table structure: {str(e)}")
+            results["operations"].append({
+                "name": "Check users table structure",
+                "success": False,
+                "error": str(e)
+            })
+
+        # Test 6: List all users
+        logger.info(f"[{test_id}] Test 6: List all users")
+        try:
+            all_users = app.db_conn.execute("SELECT id, email FROM users").fetchall()
+            results["operations"].append({
+                "name": "List all users",
+                "success": True,
+                "count": len(all_users) if all_users else 0,
+                "users": [{"id": user[0], "email": user[1]} for user in all_users] if all_users else []
+            })
+            logger.info(f"[{test_id}] Test 6 result: {all_users}")
+        except Exception as e:
+            logger.error(f"[{test_id}] Error listing all users: {str(e)}")
+            results["operations"].append({
+                "name": "List all users",
+                "success": False,
+                "error": str(e)
+            })
+
+        logger.info(f"[{test_id}] Database test completed successfully")
+        return results
+    except Exception as e:
+        logger.error(f"[{test_id}] Database test failed: {str(e)}")
+        results["operations"].append({
+            "name": "Error during tests",
+            "success": False,
+            "error": str(e)
+        })
+        return results
+
+# Test user creation endpoint (direct SQL)
+@app.get("/create-test-user", tags=["General"])
+async def create_test_user():
+    """
+    Create a test user with email [email protected].
+    This is for testing purposes only.
+
+    Returns:
+        dict: Information about the created test user.
+    """
+    import time
+
+    # Generate a unique test ID
+    test_id = f"test_{int(time.time())}"
+    logger.info(f"[{test_id}] Starting test user creation (direct SQL)")
+
+    if not hasattr(app, "db_conn"):
+        logger.error(f"[{test_id}] Database connection not available")
+        return {
+            "success": False,
+            "error": "Database connection not available"
+        }
+
+    try:
+        # First check if the test user already exists
+        logger.info(f"[{test_id}] Checking if test user already exists")
+        check_query = "SELECT id FROM users WHERE email = ?"
+        existing_user = app.db_conn.execute(check_query, ("[email protected]",)).fetchone()
+
+        if existing_user:
+            logger.info(f"[{test_id}] Test user already exists with ID: {existing_user[0]}")
+            return {
+                "success": True,
+                "user_id": existing_user[0],
+                "email": "[email protected]",
+                "status": "already_exists"
+            }
+
+        # Use a pre-hashed password to avoid dependency on passlib
+        logger.info(f"[{test_id}] Using pre-hashed password for test user")
+        # This is a pre-hashed version of "TestPassword123!" using Argon2
+        hashed_password = "$argon2id$v=19$m=65536,t=3,p=4$NElQRUZCWDRZSHpIWWRGSA$TYU8R7EfXGgEu9FWZGMX9AVwmMwpSKECCZMXgbzr6JE"
+
+        # Insert the test user
+        logger.info(f"[{test_id}] Inserting test user with email: [email protected]")
+        insert_query = "INSERT INTO users (email, hashed_password) VALUES (?, ?)"
+        cursor = app.db_conn.execute(insert_query, ("[email protected]", hashed_password))
+        app.db_conn.commit()
+        logger.info(f"[{test_id}] Committed test user insert")
+
+        # Try to get the user ID
+        user_id = None
+        try:
+            user_id = cursor.lastrowid
+            logger.info(f"[{test_id}] Got test user lastrowid: {user_id}")
+        except Exception as e:
+            logger.warning(f"[{test_id}] Could not get test user lastrowid: {str(e)}")
+
+        # Verify the insert with a separate query
+        logger.info(f"[{test_id}] Verifying test user was created")
+        verify_query = "SELECT id, email, created_at FROM users WHERE email = ?"
+        verify_result = app.db_conn.execute(verify_query, ("[email protected]",)).fetchone()
+
+        if verify_result:
+            user_id = verify_result[0]
+            logger.info(f"[{test_id}] Verified test user with ID: {user_id}")
+            return {
+                "success": True,
+                "user_id": user_id,
+                "email": "[email protected]",
+                "created_at": verify_result[2] if len(verify_result) > 2 else None,
+                "status": "created"
+            }
+        else:
+            logger.error(f"[{test_id}] Failed to verify test user after insert")
+            return {
+                "success": False,
+                "error": "Failed to verify user after insert"
+            }
+    except Exception as e:
+        logger.error(f"[{test_id}] Error creating test user: {str(e)}")
+        return {
+            "success": False,
+            "error": str(e)
+        }
+
 # Import and include routers
 from app.api.routes import auth_router, projects_router, journals_router