Chương trình học

• Thực hành kiểm thử giao diện Python 1 (Phần trên) - Lý thuyết kiểm thử giao diện
• Thực hành kiểm thử giao diện Python 1 (Phần dưới) - Sử dụng công cụ kiểm thử
• Thực hành kiểm thử giao diện Python 2 - Gửi yêu cầu bằng Python
• Thực hành kiểm thử giao diện Python 3 (Phần trên) - Thao tác cơ sở dữ liệu bằng Python
• Thực hành kiểm thử giao diện Python 3 (Phần dưới) - Khung kiểm thử unittest
• Thực hành kiểm thử giao diện Python 4 (Phần trên) - Xây dựng khung kiểm thử
• Thực hành kiểm thử giao diện Python 4 (Phần dưới) - Hoàn thiện khung: lớp cơ sở, nhãn kiểm thử, chạy lại trường hợp thất bại
• Thực hành kiểm thử giao diện Python 5 (Phần trên) - Git và tích hợp liên tục Jenkins
• Thực hành kiểm thử giao diện Python 5 (Phần dưới) - RESTful, Web Service và Mock Server

Nội dung bài học

• Thao tác cơ sở dữ liệu
• Đóng gói thao tác cơ sở dữ liệu

Dẫn nhập

Trong kiểm thử chức năng và giao diện, việc thao tác cơ sở dữ liệu thường cần thiết để chuẩn bị dữ liệu, kiểm tra môi trường và xác minh tính chính xác của thao tác cơ sở dữ liệu. Trong kiểm thử tự động, chúng ta cần sử dụng mã nguồn để kết nối cơ sở dữ liệu, tự động hoàn thành chuẩn bị dữ liệu, kiểm tra môi trường và xác minh cơ sở dữ liệu. Để thao tác MySQL bằng Python, chúng ta sử dụng thư viện thứ ba pymysql.

Cách cài đặt: pip install pymysql

Thao tác cơ sở dữ liệu

1. Tạo kết nối cơ sở dữ liệu conn = pymysql.connect()
2. Tạo con trỏ từ kết nối cur = conn.cursor()
3. Thực thi SQL (đọc/ghi) cur.execute(sql)
4. Lấy kết quả (đọc)/Gửi thay đổi (ghi) cur.fetchall()/conn.commit()
5. Đóng con trỏ và kết nối cur.close();conn.close()

import pymysql

# Tạo kết nối
db_connection = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123456',
    database='api_test',
    charset='utf8'
)

# Tạo con trỏ
cursor = db_connection.cursor()

# Truy vấn dữ liệu
cursor.execute("SELECT * FROM users WHERE username='NguyenVanA'")

# Lấy kết quả
results = cursor.fetchall()
print(results)

# Xóa dữ liệu
cursor.execute("DELETE FROM users WHERE username='TranVanB'")

# Gửi thay đổi
db_connection.commit()

# Đóng tài nguyên
cursor.close()
db_connection.close()

Thao tác truy vấn

Sau khi thực thi cur.execute(), phương thức này trả về số hàng bị ảnh hưởng chứ không phải kết quả truy vấn. Cần sử dụng các phương thức sau để lấy kết quả:

• cur.fetchone(): Lấy 1 bản ghi (kiểu tuple)
• cur.fetchmany(3): Lấy nhiều bản ghi (tuple lồng nhau)
• cur.fetchall(): Lấy tất cả bản ghi

Thao tác chỉnh sửa

Thay đổi dữ liệu sẽ không có hiệu lực cho đến khi gọi conn.commit(). Hỗ trợ transaction và rollback:

try:
    cursor.execute("INSERT INTO users (username, password) VALUES ('LeVanC', '123')") 
    cursor.execute("INSERT INTO users (username, password) VALUES ('PhamVanD', '123')") 
    db_connection.commit()
except Exception as error:
    db_connection.rollback()
    print(str(error))

Đóng gói thao tác cơ sở dữ liệu

Phương pháp hàm

def create_connection():
    return pymysql.connect(
        host='127.0.0.1',
        port=3306,
        user='test',
        password='123456',
        database='api_test',
        charset='utf8'
    )

def execute_query(sql):
    connection = create_connection()
    cursor = connection.cursor()
    cursor.execute(sql)
    result = cursor.fetchall()
    cursor.close()
    connection.close()
    return result

def modify_data(sql):
    connection = create_connection()
    cursor = connection.cursor()
    try:
        cursor.execute(sql)
        connection.commit()
    except Exception as e:
        connection.rollback()
    finally:
        cursor.close()
        connection.close()

def check_user_exists(name):
    sql = "SELECT * FROM users WHERE username='{}'".format(name)
    return bool(execute_query(sql))

def add_new_user(username, pwd):
    sql = "INSERT INTO users (username, password) VALUES ('{}','{}')".format(username, pwd)
    modify_data(sql)

def remove_user(name):
    sql = "DELETE FROM users WHERE username='{}'".format(name)
    modify_data(sql)

Phương pháp hướng đối tượng

class DatabaseHandler:
    def __init__(self):
        self.connection = pymysql.connect(
            host='127.0.0.1',
            port=3306,
            user='admin',
            password='123456',
            database='test_db'
        )
        self.cursor = self.connection.cursor()
    
    def __del__(self):
        self.cursor.close()
        self.connection.close()
    
    def run_query(self, query):
        self.cursor.execute(query)
        return self.cursor.fetchall()
    
    def execute_command(self, command):
        try:
            self.cursor.execute(command)
            self.connection.commit()
        except Exception as error:
            self.connection.rollback()
            print(str(error))
    
    def verify_user(self, name):
        result = self.run_query(f"SELECT * FROM users WHERE username='{name}'")
        return bool(result)
    
    def delete_record(self, name):
        self.execute_command(f"DELETE FROM users WHERE username='{name}'")

Lời khuyên

• Thông tin kết nối nên lưu trong file cấu hình
• Kiểm tra cú pháp SQL bằng công cụ chuyên dụng trước khi sử dụng
• Thao tác thay đổi cơ sở dữ liệu cần thận trọng