Encapsulating MySQL Database Assertions in Pytest for Automated API Testing

In API automation testing, database assertions verify that the database state after an interface operation matches expectations; this guide details how to encapsulate MySQL database assertions within the Pytest framework, covering connection handling, assertion utilities, parameterized tests, JSON validation, and logging.

1. Encapsulating MySQL Database Connection

pip install pymysql

import pymysql

class MySQLDatabase:
    def __init__(self, host, user, password, db):
        self.host = host
        self.user = user
        self.password = password
        self.db = db
        self.connection = None

    def connect(self):
        self.connection = pymysql.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            db=self.db,
            charset='utf8mb4',
            cursorclass=pymysql.cursors.DictCursor
        )

    def execute_query(self, query):
        if not self.connection:
            self.connect()
        with self.connection.cursor() as cursor:
            cursor.execute(query)
            result = cursor.fetchall()
        return result

    def close(self):
        if self.connection:
            self.connection.close()

2. Encapsulating Database Assertion Function

def assert_db_query(query, expected_result):
    """
    Verify that the database query result matches the expected result.

    :param query: SQL query string
    :param expected_result: Expected query result
    """
    db = MySQLDatabase(host='localhost', user='root', password='password', db='testdb')
    db.connect()
    result = db.execute_query(query)
    db.close()
    assert result == expected_result, f"数据库查询结果不匹配。预期：{expected_result}，实际：{result}"

3. Parameterized Test Cases

test_data = [
    ("SELECT * FROM users WHERE id=1", [{"id": 1, "name": "John Doe"}]),
    ("SELECT * FROM users WHERE id=2", [{"id": 2, "name": "Jane Doe"}]),
]

import pytest

@pytest.mark.parametrize("query, expected_result", test_data)
def test_database(query, expected_result):
    assert_db_query(query, expected_result)

4. JSON Response Assertion

def assert_json(response, expected_json):
    """
    Verify that the JSON data returned by an API matches the expected JSON.

    :param response: API response object
    :param expected_json: Expected JSON data
    """
    actual_json = response.json()
    assert actual_json == expected_json, f"JSON 数据不匹配。预期：{expected_json}，实际：{actual_json}"

5. Logging Integration with Loguru

pip install loguru

from loguru import logger

logger.add("logs/{time}.log", rotation="1 day", compression="zip")

Using the logger inside the combined test:

def test_api_and_db():
    logger.info("开始测试接口和数据库")
    response = requests.get("https://api.example.com/users/1")
    logger.info("接口请求完成")
    assert_json(response, {"id": 1, "name": "John Doe"})

    query = "SELECT * FROM users WHERE id=1"
    expected_result = [{"id": 1, "name": "John Doe"}]
    assert_db_query(query, expected_result)
    logger.info("数据库验证完成")

6. Conclusion

By encapsulating MySQL connection handling, assertion utilities, parameterized test data, JSON validation, and logging, developers can significantly improve the efficiency, readability, and maintainability of API automation tests that involve database verification.