feat: add database profiler support for multiple databases

examples/pytest/qase.config.json

@@ -1,6 +1,7 @@
 {
 	"mode": "testops",
 	"fallback": "report",
+	"profilers": ["db"],
 	"report": {
 		"driver": "local",
 		"connection": {

examples/pytest/requirements.txt

@@ -3,3 +3,7 @@ qase-pytest==6.0.0
 qase-python-commons==3.0.2
 qase-api-client==1.0.1
 qase-api-v2-client==1.0.0
+psycopg2-binary>=2.9.0
+pymysql>=1.0.0
+pymongo>=4.0.0
+redis>=4.0.0

examples/pytest/tests/test_mongodb_profiler.py

@@ -0,0 +1,119 @@
+"""
+Simple test for MongoDB database profiler
+"""
+import pytest
+import os
+from qase.pytest import qase
+
+
+@qase.id(203)
+@qase.title("MongoDB Database Profiler Test")
+@qase.description("Simple test to verify MongoDB database profiler functionality")
+@qase.severity("normal")
+@qase.priority("high")
+def test_mongodb_profiler():
+    """Test MongoDB database operations with profiler."""
+    try:
+        from pymongo import MongoClient
+    except ImportError:
+        pytest.skip("pymongo not installed")
+
+    # Get connection parameters from environment variables or use defaults
+    db_host = os.getenv("MONGODB_HOST", "localhost")
+    db_port = int(os.getenv("MONGODB_PORT", "27017"))
+    db_name = os.getenv("MONGODB_DB", "testdb")
+    db_user = os.getenv("MONGODB_USER", None)
+    db_password = os.getenv("MONGODB_PASSWORD", None)
+
+    client = None
+
+    try:
+        # Connect to MongoDB
+        with qase.step("Connect to MongoDB database"):
+            if db_user and db_password:
+                client = MongoClient(
+                    host=db_host,
+                    port=db_port,
+                    username=db_user,
+                    password=db_password,
+                    authSource="admin"
+                )
+            else:
+                client = MongoClient(
+                    host=db_host,
+                    port=db_port
+                )
+
+        db = client[db_name]
+        collection = db.users
+
+        # Insert data
+        with qase.step("Insert test users"):
+            result1 = collection.insert_one({
+                "name": "John Doe",
+                "email": "john@example.com"
+            })
+            result2 = collection.insert_one({
+                "name": "Jane Smith",
+                "email": "jane@example.com"
+            })
+            assert result1.inserted_id is not None
+            assert result2.inserted_id is not None
+
+        # Find all documents
+        with qase.step("Query all users"):
+            users = list(collection.find())
+            assert len(users) == 2, f"Expected 2 users, got {len(users)}"
+
+        # Find specific document
+        with qase.step("Query user by name"):
+            user = collection.find_one({"name": "John Doe"})
+            assert user is not None
+            assert user["email"] == "john@example.com"
+
+        # Update document
+        with qase.step("Update user email"):
+            result = collection.update_one(
+                {"name": "John Doe"},
+                {"$set": {"email": "john.doe@example.com"}}
+            )
+            assert result.modified_count == 1
+
+        # Verify update
+        with qase.step("Query updated user"):
+            user = collection.find_one({"name": "John Doe"})
+            assert user is not None
+            assert user["email"] == "john.doe@example.com"
+
+        # Delete document
+        with qase.step("Delete user"):
+            result = collection.delete_one({"name": "Jane Smith"})
+            assert result.deleted_count == 1
+
+        # Verify final state
+        with qase.step("Verify final user count"):
+            count = collection.count_documents({})
+            assert count == 1, f"Expected 1 user, got {count}"
+
+    except Exception as e:
+        # Check if it's a connection error
+        error_msg = str(e).lower()
+        if "connection" in error_msg or "timeout" in error_msg or "network" in error_msg:
+            pytest.skip(f"MongoDB not available: {e}")
+        else:
+            pytest.fail(f"Test failed with error: {e}")
+    finally:
+        # Clean up: drop collection and close connection
+        if client is not None:
+            try:
+                db = client[db_name]
+                collection = db.users
+                collection.drop()
+            except Exception:
+                pass
+            finally:
+                try:
+                    client.close()
+                except Exception:
+                    pass
+

examples/pytest/tests/test_mysql_profiler.py

@@ -0,0 +1,133 @@
+"""
+Simple test for MySQL database profiler
+"""
+import pytest
+import os
+from qase.pytest import qase
+
+
+@qase.id(202)
+@qase.title("MySQL Database Profiler Test")
+@qase.description("Simple test to verify MySQL database profiler functionality")
+@qase.severity("normal")
+@qase.priority("high")
+def test_mysql_profiler():
+    """Test MySQL database operations with profiler."""
+    try:
+        import pymysql
+    except ImportError:
+        pytest.skip("pymysql not installed")
+
+    # Get connection parameters from environment variables or use defaults
+    db_host = os.getenv("MYSQL_HOST", "localhost")
+    db_port = int(os.getenv("MYSQL_PORT", "3306"))
+    db_name = os.getenv("MYSQL_DB", "testdb")
+    db_user = os.getenv("MYSQL_USER", "testuser")
+    db_password = os.getenv("MYSQL_PASSWORD", "testpass")
+
+    conn = None
+    cursor = None
+
+    try:
+        # Connect to MySQL
+        with qase.step("Connect to MySQL database"):
+            conn = pymysql.connect(
+                host=db_host,
+                port=db_port,
+                database=db_name,
+                user=db_user,
+                password=db_password
+            )
+
+        cursor = conn.cursor()
+
+        # Create table
+        with qase.step("Create users table"):
+            cursor.execute("""
+                CREATE TABLE IF NOT EXISTS users (
+                    id INT AUTO_INCREMENT PRIMARY KEY,
+                    name VARCHAR(100) NOT NULL,
+                    email VARCHAR(100) UNIQUE NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+
+        # Insert data
+        with qase.step("Insert test users"):
+            cursor.execute(
+                "INSERT INTO users (name, email) VALUES (%s, %s)",
+                ("John Doe", "john@example.com")
+            )
+            cursor.execute(
+                "INSERT INTO users (name, email) VALUES (%s, %s)",
+                ("Jane Smith", "jane@example.com")
+            )
+
+        # Select data
+        with qase.step("Query all users"):
+            cursor.execute("SELECT * FROM users ORDER BY id")
+            users = cursor.fetchall()
+            assert len(users) == 2, f"Expected 2 users, got {len(users)}"
+
+        # Update data
+        with qase.step("Update user email"):
+            cursor.execute(
+                "UPDATE users SET email = %s WHERE name = %s",
+                ("john.doe@example.com", "John Doe")
+            )
+            assert cursor.rowcount == 1
+
+        # Select specific user
+        with qase.step("Query updated user"):
+            cursor.execute("SELECT * FROM users WHERE name = %s", ("John Doe",))
+            user = cursor.fetchone()
+            assert user is not None
+            assert user[2] == "john.doe@example.com"
+
+        # Delete data
+        with qase.step("Delete user"):
+            cursor.execute("DELETE FROM users WHERE name = %s", ("Jane Smith",))
+            assert cursor.rowcount == 1
+
+        # Verify final state
+        with qase.step("Verify final user count"):
+            cursor.execute("SELECT COUNT(*) FROM users")
+            count = cursor.fetchone()[0]
+            assert count == 1, f"Expected 1 user, got {count}"
+
+        # Commit transaction
+        conn.commit()
+
+    except pymysql.err.OperationalError as e:
+        pytest.skip(f"MySQL not available: {e}")
+    except Exception as e:
+        pytest.fail(f"Test failed with error: {e}")
+    finally:
+        # Clean up: drop table and close connections
+        if cursor is not None:
+            try:
+                # Try to drop table only if connection is still valid
+                if conn is not None and conn.open:
+                    try:
+                        cursor.execute("DROP TABLE IF EXISTS users")
+                        conn.commit()
+                    except Exception:
+                        # If drop fails, try to rollback
+                        try:
+                            conn.rollback()
+                        except Exception:
+                            pass
+            except Exception:
+                pass
+            finally:
+                try:
+                    cursor.close()
+                except Exception:
+                    pass
+        if conn is not None:
+            try:
+                if conn.open:
+                    conn.close()
+            except Exception:
+                pass
+