Build a Reusable Python API Test Framework with Assertions, DB Checks, and Pytest

Goal

✅ Encapsulate generic API response assertion logic ✅ Encapsulate generic database query and assertion logic ✅ Support chainable calls, logging, exception handling, and other advanced features ✅ Integrate with pytest / unittest testing frameworks

Required Dependencies (pip install)

pip install httpx pymysql SQLAlchemy pytest

Suggested Directory Structure

test_framework/
│
├── utils/
│   ├── assert_utils.py       # Assertion utilities (API + DB)
│   └── db_utils.py          # Database connection utilities
│
├── testcases/
│   └── test_api_with_assert.py
│
└── config/
    └── db_config.py         # Database configuration

Core Code Implementation

1️⃣ Database configuration: config/db_config.py

# config/db_config.py
DB_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "your_password",
    "database": "test_db",
    "charset": "utf8mb4"
}

2️⃣ Database utility class: utils/db_utils.py

# utils/db_utils.py
import pymysql
from config.db_config import DB_CONFIG

class DBUtils:
    def __init__(self, db_config=None):
        self.db_config = db_config or DB_CONFIG
        self.connection = None

    def connect(self):
        self.connection = pymysql.connect(**self.db_config)
        return self.connection.cursor()

    def query(self, sql: str):
        cursor = self.connect()
        try:
            cursor.execute(sql)
            result = cursor.fetchone() if "SELECT" in sql.upper() else None
            self.connection.commit()
            return result
        finally:
            cursor.close()
            self.connection.close()

    def execute(self, sql: str):
        cursor = self.connect()
        try:
            cursor.execute(sql)
            self.connection.commit()
        finally:
            cursor.close()
            self.connection.close()

3️⃣ Advanced assertion utility class: utils/assert_utils.py

# utils/assert_utils.py
import logging
import json
from typing import Any, Union

logger = logging.getLogger(__name__)

class AssertUtils:
    def __init__(self):
        self._failures = []

    def assert_equal(self, actual: Any, expected: Any, msg: str = "") -> None:
        try:
            assert actual == expected, f"{msg} | Expected: {expected}, Actual: {actual}"
        except AssertionError as e:
            self._failures.append(str(e))
            logger.error(f"Assertion failed: {e}")

    def assert_in(self, actual: str, expected: str, msg: str = "") -> None:
        try:
            assert expected in actual, f"{msg} | Expected substring '{expected}' not in '{actual}'"
        except AssertionError as e:
            self._failures.append(str(e))
            logger.error(f"Assertion failed: {e}")

    def assert_status_code(self, response, expected_code: int) -> None:
        self.assert_equal(response.status_code, expected_code, "HTTP status code mismatch")

    def assert_json_key(self, response, key: str) -> None:
        try:
            json_data = response.json()
            assert key in json_data, f"JSON response missing key: {key}"
        except Exception as e:
            self._failures.append(str(e))
            logger.error(f"Assertion failed: {e}")

    def assert_json_value(self, response, key: str, expected_value: Any) -> None:
        try:
            json_data = response.json()
            actual_value = json_data.get(key)
            assert actual_value == expected_value, f"JSON field value mismatch: {key} | Expected: {expected_value}, Actual: {actual_value}"
        except Exception as e:
            self._failures.append(str(e))
            logger.error(f"Assertion failed: {e}")

    def assert_db_value(self, db_utils, sql: str, expected_value: Any) -> None:
        try:
            result = db_utils.query(sql)
            actual_value = result[0] if result else None
            assert actual_value == expected_value, f"Database value mismatch | SQL: {sql} | Expected: {expected_value}, Actual: {actual_value}"
        except Exception as e:
            self._failures.append(str(e))
            logger.error(f"Assertion failed: {e}")

    def assert_all(self):
        if self._failures:
            error_msg = "
".join(self._failures)
            raise AssertionError(f"The following assertions failed:
{error_msg}")
        else:
            logger.info("All assertions passed")

4️⃣ Example test case: testcases/test_api_with_assert.py

# testcases/test_api_with_assert.py
import httpx
from utils.assert_utils import AssertUtils
from utils.db_utils import DBUtils
import pytest

@pytest.fixture
def client():
    return httpx.Client(base_url="https://jsonplaceholder.typicode.com")

@pytest.fixture
def assert_utils():
    return AssertUtils()

@pytest.fixture
def db_utils():
    return DBUtils()

def test_get_user_and_assert(client, assert_utils, db_utils):
    response = client.get("/users/1")
    # API assertions
    assert_utils.assert_status_code(response, 200)
    assert_utils.assert_json_key(response, "name")
    assert_utils.assert_json_value(response, "id", 1)
    # Database assertion (assuming a corresponding user table)
    user_id = 1
    sql = f"SELECT name FROM users WHERE id = {user_id}"
    assert_utils.assert_db_value(db_utils, sql, "Leanne Graham")
    # Verify all assertions
    assert_utils.assert_all()

Run Tests & View Results

pytest testcases/test_api_with_assert.py -v

Sample output:

========= test session starts ==========
collected 1 item
test_api_with_assert.py::test_get_user_and_assert PASSED    [100%]
========= 1 passed in 0.5s ===========

Advantages Summary

(illustrative image omitted)

Optional Advanced Features

(illustrative image omitted)

Packaging & Release Suggestions

You can package these utility classes as a Python package for reuse across projects:

pip install setuptools wheel
python setup.py sdist bdist_wheel