使用pytest对数据库进行校验

背景

• 业务中需要与其他方进行对接，对方提供数据
• 由于有异常的数据会对业务操作非常大的影响
• 需要将异常数据进行前置检查，未通过的数据不允许入库

pytest介绍

• pytest 是一个用于 Python 的测试框架,支持简单的单元测试和复杂的功能测试。它以其简单、易用、灵活的特点,受到了许
  多开发者的青睐。
• pytest 框架可以轻松编写小型、可读的测试,并可以扩展支持应用程序和库的复杂功能测试
• pytest 官网：https://docs.pytest.org/en/8.2.x/

pytest特点

1. 简洁的语法:无需继承特定的测试类,只需使用简单的函数即可编写测试。
2. 强大的断言:内置丰富的断言方法,提供详细的失败信息。
3. 自动发现:自动发现测试文件和测试函数,无需显式地注册测试。
4. 插件系统:丰富的插件生态系统,支持扩展和定制。

实践使用

1、安装引用

requirements.txt 文件

psycopg2-binary==2.9.9
pytest==8.2.1
pytest-xdist==3.6.1
pytest-html==4.1.1

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

2、建立数据库连接

import os
import pytest
import psycopg2

pg_conf = {
    'host': os.getenv("PG_HOST", '192.168.1.81') ,
    'port': os.getenv("PG_PORT", '5432'),
    'user': os.getenv("PG_USER", 'postgres'),
    'password': os.getenv("PG_PASSWORD", 'postgres'),
    'dbname': os.getenv("PG_DBNAME", 'postgres')
}

@pytest.fixture(scope="session")
def db_conn():
    conn = psycopg2.connect(**pg_conf)
    yield conn
    conn.close()

3、编写测试用例

可以建立一个tests的文件夹，专门存放测试用例的代码

import pytest
import psycopg2


class TestLicense():

    def test_001(self, db_conn):
        cur = db_conn.cursor()
        cur.execute("SELECT COUNT(*) FROM t_license_version")
        count = cur.fetchone()[0]
        assert count >= 200, f"assert {count} >= 200, 许可证数据量应大于 200"
        cur.close()
    
    def test_002(self, db_conn):
        cur = db_conn.cursor()
        cur.execute("SELECT COUNT(*) FROM t_license_version WHERE risk_level NOT IN ('unknown','low','medium','high')")
        count = cur.fetchone()[0]
        assert count == 0, f"assert {count} == 0, 风险等级 risk_level 应包含 ('unknown','low','medium','high')"
        cur.close()

    def test_003(self, db_conn):
        cur = db_conn.cursor()
        cur.execute("SELECT COUNT(*) FROM t_license_version WHERE conditions_use IS NULL")
        count = cur.fetchone()[0]
        assert count == 0, f"assert {count} == 0, 限制条件 conditions_use 不能为 NULL"
        cur.close()

4、运行测试

import pytest

if __name__ == '__main__':
    pytest.main(['-n', 'auto', '--html=report.html', '--self-contained-html'])

5、运行测试

python main.py

运行完成后会在当前目录产生检测包括，是html格式的，可以在浏览器直接打开

构建集成

1、docker运行

• 由于连接数据库需要在服务器上安装 postgresql-client，使用docker可以避开一些环境问题
• 方便集成到jk流水线上进行CI/CD实践

Dockerfile

FROM python:3.10-slim-buster

WORKDIR /app

COPY . /app

RUN echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ buster main non-free contrib" > /etc/apt/sources.list && \
    echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian/ buster-updates main non-free contrib" >> /etc/apt/sources.list && \
    echo "deb https://mirrors.tuna.tsinghua.edu.cn/debian-security buster/updates main non-free contrib" >> /etc/apt/sources.list && \
    apt-get update && \
    apt-get install -y postgresql-client && \
    pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

CMD ["python3", "main.py"]

构建使用

docker build -t app-test:latest .

docker run --rm -it -e PG_HOST=192.168.1.81 -e PG_PORT=5432 -e PG_USER=postgres -e PG_PASSWORD=xxxxxx -e PG_DBNAME=postgres -v $(pwd):/app app-test:latest

2、增加 Makefile

VENV_NAME?=venv
PYTHON=python3
VERSION?=latest
APP_NAME=app-test

dep: venv
	@pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

venv:
	${PYTHON} -m venv venv
	source $(VENV_NAME)/bin/activate

test: venv
	${PYTHON} -m pytest

lint: venv
	${PYTHON} -m pylint
	${PYTHON} -m mypy

run: venv
	${PYTHON} main.py

docker-build:
	docker build -t ${APP_NAME}:${VERSION} .

docker-test:
	docker run --rm -it -v $(shell pwd):/app ${APP_NAME}:${VERSION}

clean:
	@find . -name '*.pyc' -delete
	@find . -name '__pycache__' -type d | xargs rm -fr
	@find . -name '.pytest_cache' -type d | xargs rm -fr