HR支付模块接口测试实战（pytest+requests）

质量保障小乔

1022人浏览 · 2025-12-03 08:44:36

质量保障小乔 · 2025-12-03 08:44:36 发布

HR支付模块接口测试实战（pytest+requests）

HR支付模块接口测试核心聚焦业务规则校验、数据一致性、异常场景覆盖，结合pytest的Fixture、参数化特性，以及requests的接口请求能力，实现接口自动化测试的工程化落地。

一、项目基础架构设计

遵循“高复用、低耦合”原则，设计标准化项目结构，适配HR支付模块接口测试的可维护性与扩展性：

hr_payment_api_test/
├── conftest.py                # 全局Fixture（请求会话、环境配置、数据库连接）
├── pytest.ini                 # pytest全局配置（标记、报告、日志）
├── test_data/                 # 测试数据目录（结构化管理）
│   └── payment_api_data.json  # 接口测试用例数据（正向/异常场景）
├── test_case/                 # 测试用例目录（按接口模块拆分）
│   ├── test_salary_calc_api.py  # 薪资计算接口
│   ├── test_salary_pay_api.py   # 薪资支付接口
│   ├── test_ssc_pay_api.py      # 社保代缴接口
│   └── test_pay_query_api.py    # 支付结果查询接口
├── utils/                     # 工具类目录（通用能力封装）
│   ├── api_utils.py           # requests请求封装（含鉴权、日志）
│   ├── db_utils.py            # 数据库操作（校验支付结果）
│   └── log_utils.py           # 日志配置（统一输出格式）
└── reports/                   # 测试报告目录（自动生成）

二、环境准备与核心工具封装

1. 环境依赖安装

# 核心依赖
pip install pytest requests pytest-html pymysql python-dotenv loguru
# 可选：接口响应断言增强（简化复杂断言）
pip install jsonschema

2. 工具类封装（核心复用能力）

（1）日志工具（utils/log_utils.py）

统一日志格式，便于问题定位：

from loguru import logger
import os

# 日志配置
def init_log():
    # 日志文件路径
    log_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "logs")
    os.makedirs(log_dir, exist_ok=True)
    # 日志输出：控制台+文件
    logger.add(
        os.path.join(log_dir, "hr_payment_api_{time}.log"),
        rotation="1 day",  # 按天分割
        retention="7 days",  # 保留7天
        encoding="utf-8",
        format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {module}:{line} | {message}"
    )
    return logger

# 初始化日志
log = init_log()

（2）接口请求封装（utils/api_utils.py）

封装requests请求，统一处理鉴权、请求头、异常捕获：

import requests
from utils.log_utils import log
from dotenv import load_dotenv
import os

# 加载环境变量（区分测试/预生产环境）
load_dotenv()
BASE_URL = os.getenv("HR_PAYMENT_BASE_URL")

class HRPaymentApi:
    def __init__(self):
        self.session = requests.Session()
        # 统一请求头（含鉴权token，实际项目中可通过登录接口获取）
        self.session.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {os.getenv('HR_PAYMENT_TOKEN')}"
        }

    def send_request(self, method, url, params=None, json=None, headers=None):
        """
        通用请求方法
        :param method: 请求方法（get/post/put/delete）
        :param url: 接口路径（拼接BASE_URL）
        :param params: url参数
        :param json: 请求体
        :param headers: 自定义请求头（覆盖默认）
        :return: 响应对象
        """
        full_url = f"{BASE_URL}{url}"
        try:
            # 合并自定义请求头
            if headers:
                self.session.headers.update(headers)
            # 发送请求
            response = self.session.request(
                method=method,
                url=full_url,
                params=params,
                json=json,
                timeout=10  # 超时时间
            )
            log.info(f"请求成功 | 方法：{method} | URL：{full_url} | 请求体：{json} | 响应码：{response.status_code}")
            return response
        except requests.exceptions.Timeout:
            log.error(f"请求超时 | 方法：{method} | URL：{full_url}")
            raise
        except requests.exceptions.ConnectionError:
            log.error(f"连接失败 | 方法：{method} | URL：{full_url}")
            raise
        except Exception as e:
            log.error(f"请求异常 | 方法：{method} | URL：{full_url} | 异常：{str(e)}")
            raise

# 实例化（Fixture中复用）
payment_api = HRPaymentApi()

（3）数据库工具（utils/db_utils.py）

校验接口执行后的业务数据一致性（如薪资支付后数据库台账更新）：

import pymysql
from utils.log_utils import log
from dotenv import load_dotenv
import os

load_dotenv()

def get_db_conn():
    """获取数据库连接（HR支付台账库）"""
    try:
        conn = pymysql.connect(
            host=os.getenv("DB_HOST"),
            port=int(os.getenv("DB_PORT")),
            user=os.getenv("DB_USER"),
            password=os.getenv("DB_PASSWORD"),
            db=os.getenv("DB_NAME"),
            charset="utf8"
        )
        log.info("数据库连接成功")
        return conn
    except Exception as e:
        log.error(f"数据库连接失败：{str(e)}")
        raise

def query_salary_pay_record(emp_id, pay_month):
    """查询员工薪资支付记录"""
    conn = get_db_conn()
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cursor:
            sql = """
                SELECT gross_salary, social_insurance, tax, net_salary, pay_status 
                FROM salary_pay_ledger 
                WHERE emp_id = %s AND pay_month = %s
            """
            cursor.execute(sql, (emp_id, pay_month))
            result = cursor.fetchone()
            log.info(f"查询薪资支付记录 | 员工ID：{emp_id} | 月份：{pay_month} | 结果：{result}")
            return result
    finally:
        conn.close()

def clear_test_data(emp_id):
    """清理测试数据（用例执行后）"""
    conn = get_db_conn()
    try:
        with conn.cursor() as cursor:
            sql = "DELETE FROM salary_pay_ledger WHERE emp_id = %s AND emp_id LIKE 'test_%'"
            cursor.execute(sql, (emp_id,))
            conn.commit()
            log.info(f"清理测试数据 | 员工ID：{emp_id}")
    finally:
        conn.close()

3. 环境配置文件（.env）

区分环境配置，避免硬编码：

# 接口基础地址
HR_PAYMENT_BASE_URL=http://test.hr-system.com/api/v1/payment
# 接口鉴权Token
HR_PAYMENT_TOKEN=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
# 数据库配置
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=test
DB_PASSWORD=123456
DB_NAME=hr_payment_db

4. pytest全局配置（pytest.ini）

[pytest]
# 测试用例路径
testpaths = test_case
# 自定义标记（按业务模块/优先级）
markers =
    salary_calc: 薪资计算接口
    salary_pay: 薪资支付接口
    ssc_pay: 社保代缴接口
    p0: 核心P0用例
    p1: 次要P1用例
# 默认参数（日志、报告）
addopts = -v -s --html=reports/hr_payment_api_report.html --self-contained-html
# 测试文件/函数命名规则
python_files = test_*.py
python_functions = test_*
python_classes = Test*

三、核心Fixture设计（conftest.py）

复用测试资源，统一处理“前置准备”和“后置清理”：

import pytest
from utils.api_utils import payment_api
from utils.db_utils import get_db_conn, clear_test_data
from utils.log_utils import log
import json

# 加载测试数据
def load_test_data():
    with open("test_data/payment_api_data.json", "r", encoding="utf8") as f:
        return json.load(f)

# Fixture：接口请求实例（会话级，全局复用）
@pytest.fixture(scope="session")
def api_client():
    log.info("===== 初始化接口请求客户端 =====")
    yield payment_api
    log.info("===== 销毁接口请求客户端 =====")

# Fixture：数据库连接（函数级，每个用例独立）
@pytest.fixture(scope="function")
def db_conn():
    conn = get_db_conn()
    yield conn
    conn.close()

# Fixture：测试数据（会话级）
@pytest.fixture(scope="session")
def test_data():
    return load_test_data()

# Fixture：测试员工ID（函数级，用例执行后清理数据）
@pytest.fixture(scope="function")
def test_emp_id():
    emp_id = "test_1001"  # 测试专用员工ID
    yield emp_id
    # 后置清理：删除测试数据
    clear_test_data(emp_id)
    log.info(f"测试用例执行完成，清理员工{emp_id}数据")

四、核心接口测试用例实现

1. 测试数据文件（test_data/payment_api_data.json）

{
  "salary_calc": {
    "normal_case": {
      "emp_id": "test_1001",
      "pay_month": "2024-10",
      "gross_salary": 10000.0,
      "social_insurance": 800.0,
      "special_deduction": 1000.0,
      "expected_tax": 120.0,
      "expected_net_salary": 9080.0
    },
    "boundary_case": {
      "emp_id": "test_1001",
      "pay_month": "2024-10",
      "gross_salary": 5000.0,
      "social_insurance": 500.0,
      "special_deduction": 0.0,
      "expected_tax": 0.0,
      "expected_net_salary": 4500.0
    },
    "abnormal_case": {
      "emp_id": "test_1001",
      "pay_month": "2024-10",
      "gross_salary": -1000.0,  # 异常：薪资为负
      "social_insurance": 800.0,
      "special_deduction": 1000.0,
      "expected_code": 400,
      "expected_msg": "应发薪资不能为负数"
    }
  },
  "salary_pay": {
    "success_case": {
      "emp_id": "test_1001",
      "pay_month": "2024-10",
      "bank_card": "6222021234567890",
      "expected_pay_status": "SUCCESS"
    },
    "balance_not_enough": {
      "emp_id": "test_1001",
      "pay_month": "2024-10",
      "bank_card": "6222021234567890",
      "expected_code": 500,
      "expected_msg": "企业账户余额不足"
    },
    "invalid_card": {
      "emp_id": "test_1001",
      "pay_month": "2024-10",
      "bank_card": "1234567890",  # 异常：卡号格式错误
      "expected_code": 400,
      "expected_msg": "银行卡号格式错误"
    }
  }
}

2. 薪资计算接口测试（test_case/test_salary_calc_api.py）

覆盖正向、边界、异常场景，校验计算结果准确性：

import pytest
from utils.log_utils import log

@pytest.mark.salary_calc
class TestSalaryCalcApi:
    @pytest.mark.p0
    def test_salary_calc_normal(self, api_client, test_data, test_emp_id):
        """正向场景：正常薪资计算"""
        # 1. 准备测试数据
        data = test_data["salary_calc"]["normal_case"]
        # 2. 发送请求
        response = api_client.send_request(
            method="POST",
            url="/salary/calc",
            json={
                "emp_id": data["emp_id"],
                "pay_month": data["pay_month"],
                "gross_salary": data["gross_salary"],
                "social_insurance": data["social_insurance"],
                "special_deduction": data["special_deduction"]
            }
        )
        # 3. 断言
        assert response.status_code == 200, "响应码非200"
        resp_json = response.json()
        assert resp_json["code"] == 0, "业务码非成功"
        assert resp_json["data"]["tax"] == data["expected_tax"], "个税计算错误"
        assert resp_json["data"]["net_salary"] == data["expected_net_salary"], "实发薪资计算错误"
        log.info("正向薪资计算接口测试通过")

    @pytest.mark.p0
    def test_salary_calc_boundary(self, api_client, test_data, test_emp_id):
        """边界场景：薪资=5000（个税起征点）"""
        data = test_data["salary_calc"]["boundary_case"]
        response = api_client.send_request(
            method="POST",
            url="/salary/calc",
            json={
                "emp_id": data["emp_id"],
                "pay_month": data["pay_month"],
                "gross_salary": data["gross_salary"],
                "social_insurance": data["social_insurance"],
                "special_deduction": data["special_deduction"]
            }
        )
        resp_json = response.json()
        assert response.status_code == 200
        assert resp_json["data"]["tax"] == data["expected_tax"]
        assert resp_json["data"]["net_salary"] == data["expected_net_salary"]
        log.info("边界值薪资计算接口测试通过")

    @pytest.mark.p1
    def test_salary_calc_abnormal(self, api_client, test_data, test_emp_id):
        """异常场景：应发薪资为负数"""
        data = test_data["salary_calc"]["abnormal_case"]
        response = api_client.send_request(
            method="POST",
            url="/salary/calc",
            json={
                "emp_id": data["emp_id"],
                "pay_month": data["pay_month"],
                "gross_salary": data["gross_salary"],
                "social_insurance": data["social_insurance"],
                "special_deduction": data["special_deduction"]
            }
        )
        resp_json = response.json()
        assert response.status_code == data["expected_code"]
        assert resp_json["msg"] == data["expected_msg"], "异常提示错误"
        log.info("异常薪资计算接口测试通过")

3. 薪资支付接口测试（test_case/test_salary_pay_api.py）

校验支付结果与数据库一致性，覆盖异常场景：

import pytest
from utils.log_utils import log
from utils.db_utils import query_salary_pay_record

@pytest.mark.salary_pay
class TestSalaryPayApi:
    @pytest.mark.p0
    def test_salary_pay_success(self, api_client, test_data, test_emp_id, db_conn):
        """正向场景：薪资支付成功（校验数据库台账）"""
        # 1. 先调用薪资计算接口（依赖）
        calc_data = test_data["salary_calc"]["normal_case"]
        api_client.send_request(
            method="POST",
            url="/salary/calc",
            json={
                "emp_id": calc_data["emp_id"],
                "pay_month": calc_data["pay_month"],
                "gross_salary": calc_data["gross_salary"],
                "social_insurance": calc_data["social_insurance"],
                "special_deduction": calc_data["special_deduction"]
            }
        )
        # 2. 调用薪资支付接口
        pay_data = test_data["salary_pay"]["success_case"]
        response = api_client.send_request(
            method="POST",
            url="/salary/pay",
            json={
                "emp_id": pay_data["emp_id"],
                "pay_month": pay_data["pay_month"],
                "bank_card": pay_data["bank_card"]
            }
        )
        # 3. 接口响应断言
        assert response.status_code == 200
        resp_json = response.json()
        assert resp_json["code"] == 0
        assert resp_json["data"]["pay_status"] == pay_data["expected_pay_status"]
        # 4. 数据库台账断言
        db_record = query_salary_pay_record(pay_data["emp_id"], pay_data["pay_month"])
        assert db_record is not None, "数据库无支付记录"
        assert db_record["pay_status"] == pay_data["expected_pay_status"], "支付状态未更新"
        assert db_record["net_salary"] == calc_data["expected_net_salary"], "实发薪资不一致"
        log.info("薪资支付成功接口测试通过")

    @pytest.mark.p1
    def test_salary_pay_balance_not_enough(self, api_client, test_data, test_emp_id):
        """异常场景：企业账户余额不足"""
        data = test_data["salary_pay"]["balance_not_enough"]
        response = api_client.send_request(
            method="POST",
            url="/salary/pay",
            json={
                "emp_id": data["emp_id"],
                "pay_month": data["pay_month"],
                "bank_card": data["bank_card"]
            }
        )
        resp_json = response.json()
        assert response.status_code == data["expected_code"]
        assert resp_json["msg"] == data["expected_msg"]
        # 校验数据库：无支付记录/支付状态为失败
        db_record = query_salary_pay_record(data["emp_id"], data["pay_month"])
        assert db_record is None or db_record["pay_status"] == "FAIL"
        log.info("薪资支付余额不足异常测试通过")

    @pytest.mark.p1
    @pytest.mark.parametrize("invalid_card", ["1234567890", "", "622202123456789012345"])
    def test_salary_pay_invalid_card(self, api_client, test_data, test_emp_id, invalid_card):
        """参数化：银行卡号格式错误（空/过短/过长）"""
        data = test_data["salary_pay"]["invalid_card"]
        response = api_client.send_request(
            method="POST",
            url="/salary/pay",
            json={
                "emp_id": data["emp_id"],
                "pay_month": data["pay_month"],
                "bank_card": invalid_card
            }
        )
        resp_json = response.json()
        assert response.status_code == data["expected_code"]
        assert resp_json["msg"] == data["expected_msg"]
        log.info(f"银行卡号{invalid_card}格式错误测试通过")

五、测试执行与报告

1. 执行测试用例

# 执行所有接口测试用例
pytest
# 仅执行薪资支付P0用例
pytest -m "salary_pay and p0"
# 执行指定文件
pytest test_case/test_salary_pay_api.py -v

2. 测试报告

执行完成后，reports/目录下生成hr_payment_api_report.html报告，包含：

用例执行统计（总数、通过数、失败数、跳过数）；
每个用例的执行时间、响应详情、断言结果；
失败用例的详细堆栈信息（便于定位问题）。

六、实战优化与注意事项

1. 关键优化点

用例解耦：避免用例强依赖（如薪资支付依赖计算接口，可在Fixture中提前执行计算接口）；
参数化扩展：复杂场景（如多薪资档位、多社保基数）用@pytest.mark.parametrize批量覆盖；
接口幂等性：支付接口需保证幂等（重复调用不重复支付），用例中需校验“重复支付返回‘已支付’”；

异步接口处理：若支付接口为异步（如提交后需轮询结果），封装轮询工具函数：

def poll_pay_result(api_client, emp_id, pay_month, timeout=30):
    """轮询支付结果"""
    start_time = time.time()
    while time.time() - start_time < timeout:
        response = api_client.send_request(
            method="GET",
            url="/salary/pay/query",
            params={"emp_id": emp_id, "pay_month": pay_month}
        )
        if response.json()["data"]["pay_status"] == "SUCCESS":
            return True
        time.sleep(1)
    return False

2. 注意事项

环境隔离：测试环境与生产环境严格隔离，使用测试专用员工ID/银行卡号；
数据清理：每个用例执行后清理测试数据（避免脏数据影响后续用例）；
鉴权处理：Token过期自动刷新（可在api_utils.py中添加Token刷新逻辑）；

异常重试：非核心异常（如网络抖动）添加重试机制（用pytest-rerunfailures插件）：

# 安装插件
pip install pytest-rerunfailures
# 执行时重试失败用例（最多2次）
pytest --reruns 2 --reruns-delay 1

七、总结

HR支付模块接口测试的核心是**“业务规则校验+数据一致性”**，通过pytest的Fixture实现资源复用，requests封装统一请求逻辑，参数化覆盖多场景，最终实现接口测试的自动化、工程化。落地时需优先覆盖P0核心场景（如薪资计算准确性、支付状态更新），再扩展异常场景，同时通过日志、报告、数据库校验保障测试结果的可信度。