跳至主要內容

函数

薇念约 2520 字

函数

函数基础

1、简介

函数时组织好的，可重复使用的代码快，用于实现某个特定的功能

2、优点

提高代码复用性

使程序结构更清晰

降低维护成本

3、函数的基本结构

def 函数名(参数1, 参数2, ...):
    """文档字符串（可选）"""
    函数体
    return 返回值（可选）

参数类型

1、位置参数

def info(name, age):
    print(f"{name}今年{age}岁了")

info("小红", 20)

2、关键字参数（**kwargs）

def show_info(**kwargs):
    for key, value in kwargs.items():
        print(f"{key}: {value}")

show_info(name="小强", age=22, city="北京")

3、默认参数

def info(name, age=18):
    print(f"{name}今年{age}岁了")

info("小蓝")  # 使用默认年龄

4、可变参数（*args），也叫不定长参数

def total(*numbers):
    print("传入了：", numbers)
    print("总和为：", sum(numbers))

total(1, 2, 3, 4)

*args` —— 不定长位置参数

用星号 * 收集额外的位置参数，变成一个元组

常用于参数数量不确定的情况

def add_numbers(*args):
    print(args)             # args 是一个元组
    return sum(args)

print(add_numbers(1, 2, 3))       # 输出 (1, 2, 3)  → 6
print(add_numbers(5, 10))         # 输出 (5, 10)    → 15
print(add_numbers())              # 输出 ()         → 0

kwargs` —— 不定长关键字参数**

用两个星号 ** 收集额外的关键字参数，变成一个字典

def print_info(**kwargs):
    print(kwargs)           # kwargs 是一个字典
    for key, value in kwargs.items():
        print(f"{key}: {value}")

print_info(name="Alice", age=25, city="Shanghai")
# 输出：
# {'name': 'Alice', 'age': 25, 'city': 'Shanghai'}
# name: Alice
# age: 25
# city: Shanghai

混合使用，*args 必须在 **kwargs 前面：

def demo(a, b, *args, **kwargs):
    print("固定参数：", a, b)
    print("可变位置参数：", args)
    print("可变关键字参数：", kwargs)

demo(1, 2, 3, 4, x=5, y=6)
# 固定参数： 1 2
# 可变位置参数： (3, 4)
# 可变关键字参数： {'x': 5, 'y': 6}

简单来说：

  • *args → 多个位置参数 → 元组
  • **kwargs → 多个关键字参数 → 字典

返回值

作用域

1、作用域分类

局部变量：定义在函数内部，只在函数内部有效

全局变量：定义在函数外部，整个程序都能访问

x = 10  # 全局变量

def func():
    x = 5  # 局部变量
    print("函数内部 x =", x)

func()
print("函数外部 x =", x)

匿名函数lambda

常用于配合map()、filter()、sorted() 等函数

# 普通函数
def square(x):
    return x * x

# lambda 表达式
square2 = lambda x: x * x

print(square(5))   # 25
print(square2(5))  # 25

函数进阶

高阶函数

装饰器

def log(func):
    def wrapper():
        print("开始执行函数...")
        func()
        print("函数执行完毕")
    return wrapper

@log
def say_hello():
    print("Hello, world!")

say_hello()

实用技巧

项目实战

用函数结构搭建小项目（学生管理系统，数据分析流程）

小工具开发

需要实现的功能：

1、日志记录，记录函数执行的信息，包括参数，结果等

2、计时统计，记录函数的执行耗时

3、错误处理，捕获函数执行过程中的异常并输出

4、登录接口测试用例

5、加载yaml数据的测试脚本

6、日志输出

7、报告美化，高质量测试报告，支持用例详情，步骤截图，附件

8、测试覆盖率统计

9、mock外部依赖或接口

10、用例失败，自动重试

11、自动获取token，刷新并保存

12、接口依赖执行，控制先后执行接口依赖顺序

13、接口环境切换，支持多环境配置动态切换

14、动态数据生成，生成模拟用户，手机号，地址等数据

15、数据库验证，用于测试后检验数据库状态

16、redis验证，检验缓存更新是否生效

17、excel/csv读写：处理测试数据输入输出，批量执行用例 18、用例参数提取器，从接口响应中提取字段传入下一个接口用 conftest.py + pytest fixtures 实现

19、YAMl或excel转pytest用例生成器：读取测试用例数据自动转为函数代码，用openpyxl/pyyaml实现

20、三合一装饰器（日志、计时、异常）

import time
import traceback
from functools import wraps

def log_timer_exception(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[START] {func.__name__}")
        start = time.time()
        try:
            result = func(*args, **kwargs)
            print(f"[SUCCESS] {func.__name__} 返回值：{result}")
            return result
        except Exception as e:
            print(f"[ERROR] {func.__name__} 报错：{e}")
            traceback.print_exc()
        finally:
            duration = time.time() - start
            print(f"[END] {func.__name__} 用时：{duration:.2f}s")
    return wrapper

用pytest对函数进行自动化测试

方案

requests → 进行接口调用

pytest → 编写测试用例

allure（可选）→ 生成测试报告

yaml/json → 进行数据驱动

logging → 输出日志

项目结构

api_test_project/
├── config/
│   └── config.yaml              # 全局配置，如base_url、headers等
├── data/
│   └── login_case.yaml          # 测试数据驱动文件
│   └── add_user.yaml            # 添加用户，用例数据
├── testcases/
│   └── test_login.py            # 用例文件（pytest写法）
├── common/
│   ├── request_handler.py       # 封装请求类,含token自动刷新
│   └── assert_handler.py        # 封装断言函数（通用）
│   └── token_manager.py         # token 读写 + 登录刷新
│   └── db_handler.py            # 数据库连接与查询
├── utils/
│   └── logger.py                # 日志封装
├── conftest.py                  # pytest钩子配置
└── pytest.ini                   # pytest运行配置
├── requirements.txt             # 依赖库列表
├── token.txt                    # 本地缓存token

安装依赖

pytest-allure-adaptor 或 allure-pytest:生成可视化测试报告,接口自动化核心插件
pytest-xdist:多线程并发执行用例	提升执行效率
pytest-html:生成 HTML 报告（简版）	快速出报告
pytest-rerunfailures:用例失败自动重跑	解决网络抖动或临时问题
pytest-ordering:控制测试用例执行顺序	有强依赖的场景
pytest-moc:mock 函数与对象	单元测试与接口依赖隔离
pytest-cov:代码覆盖率统计	结合单元测试使用
pytest-dependency:指定测试用例依赖关系	登录接口依赖场景常见
pytest-random-order:随机打乱用例执行顺序	检测测试间是否存在依赖
pytest-timeout:控制单个测试用例执行时间	防止卡死、超时阻塞
# 常用
pip install pytest
pip install allure-pytest（生成自定义的报告）
pip install pytest-xdist（多线程运行）
pip install pytest-html （生成html报告）
pip install pytest-rerunfailures（失败用例重跑）
pip install pytest-ordering（改变用例的执行顺序）


pip install requests pytest pyyaml allure-pytest


# 然后在虚拟环境执行命令，批量安装依赖
pip install -r requirement.txt

request封装:common/request_handler.py

import requests
import yaml

class RequestHandler:
    def __init__(self, base_url=None):
        self.base_url = base_url or ""

    def send(self, method, url, headers=None, params=None, data=None, json=None):
        full_url = self.base_url + url
        try:
            response = requests.request(
                method=method.upper(),
                url=full_url,
                headers=headers,
                params=params,
                data=data,
                json=json
            )
            return response
        except Exception as e:
            print(f"请求出错：{e}")
            return None

测试数据驱动:data/login_case.yaml

- case_name: 正确登录
  method: post
  url: /api/login
  json:
    username: admin
    password: 123456
  expected_status: 200
  expected_msg: "登录成功"

- case_name: 错误密码
  method: post
  url: /api/login
  json:
    username: admin
    password: wrongpwd
  expected_status: 401
  expected_msg: "密码错误"

接口用例编写:testcases/test_login.py

import pytest
import yaml
from common.request_handler import RequestHandler

handler = RequestHandler(base_url="http://your-api.com")

# 加载yaml测试数据
def load_login_data():
    with open("data/login_case.yaml", encoding="utf-8") as f:
        return yaml.safe_load(f)

@pytest.mark.parametrize("case", load_login_data())
def test_login(case):
    print(f"\n【执行用例】：{case['case_name']}")
    res = handler.send(
        method=case['method'],
        url=case['url'],
        json=case['json']
    )
    assert res.status_code == case['expected_status']
    assert case['expected_msg'] in res.text

运行方式

# 普通运行
pytest testcases/

# 加上报告（需要先安装 allure）
pytest testcases/ --alluredir=report/

# 打开allure报告（需先安装 allure 命令行工具）
allure serve report/

企业级接口测试：Token提取+过期自动刷新+多接口串联+数据库校验

# 场景设定（真实项目中非常常见）
# Token提取与共享，fixture实现
# 登录并提取token
# common/token_handler.py
import requests

def get_token():
    url = "http://your-api.com/api/login"
    payload = {"username": "admin", "password": "123456"}
    res = requests.post(url, json=payload)
    token = res.json().get("token")
    return token
    
# 在pytest中共享token，使用fixture实现
# conftest.py
import pytest
from common.token_handler import get_token

@pytest.fixture(scope="session")
def token():
    return get_token()

# token保存和加载工具：common/token_manager.py
import requests
import json
import os

TOKEN_FILE = 'token.txt'

def save_token(token):
    with open(TOKEN_FILE, 'w') as f:
        f.write(token)

def load_token():
    if os.path.exists(TOKEN_FILE):
        with open(TOKEN_FILE, 'r') as f:
            return f.read().strip()
    return None

def refresh_token():
    """重新登录获取 token 并保存"""
    url = "http://your-api.com/api/login"
    payload = {"username": "admin", "password": "123456"}
    res = requests.post(url, json=payload)
    token = res.json().get("token")
    if token:
        save_token(token)
    return token
# 自动加token并自动重试：封装请求类common/request_handler.py
from common.token_manager import load_token, refresh_token
import requests

class RequestHandler:
    def __init__(self, base_url=None):
        self.base_url = base_url or ""

    def send(self, method, url, headers=None, **kwargs):
        if headers is None:
            headers = {}
        token = load_token() or refresh_token()
        headers['Authorization'] = f'Bearer {token}'
        full_url = self.base_url + url

        response = requests.request(method, full_url, headers=headers, **kwargs)

        # 如果 token 失效，自动刷新后重试一次
        if response.status_code in (401, 403):
            print("⚠️ Token 失效，正在刷新...")
            new_token = refresh_token()
            headers['Authorization'] = f'Bearer {new_token}'
            response = requests.request(method, full_url, headers=headers, **kwargs)

        return response
# 用例中不再需要处理token，只需要调用封装类testcases/test_user_info.py
from common.request_handler import RequestHandler

handler = RequestHandler(base_url="http://your-api.com")

def test_get_user_info():
    res = handler.send("get", "/api/user/info")
    assert res.status_code == 200
    assert "用户信息" in res.text


# 添加用户
# data/add_user.yaml
- case_name: 添加普通用户
  url: /api/user/add
  method: post
  json:
    name: "张三"
    age: 28
  expected_status: 200
  expected_msg: "添加成功"

# 测试用例文件（自动加token）
# testcases/test_add_user.py
import pytest
import yaml
from common.request_handler import RequestHandler

handler = RequestHandler(base_url="http://your-api.com")

def load_case():
    with open("data/add_user.yaml", encoding="utf-8") as f:
        return yaml.safe_load(f)

@pytest.mark.parametrize("case", load_case())
def test_add_user(case, token):
    headers = {"Authorization": f"Bearer {token}"}
    res = handler.send(
        method=case["method"],
        url=case["url"],
        headers=headers,
        json=case["json"]
    )
    assert res.status_code == case["expected_status"]
    assert case["expected_msg"] in res.text
    
# 数据库连接：common/db_handler.py
import pymysql

def query_user_by_name(name):
    conn = pymysql.connect(
        host='localhost',
        user='root',
        password='123456',
        database='user_db'
    )
    cursor = conn.cursor()
    sql = f"SELECT * FROM users WHERE name = '{name}'"
    cursor.execute(sql)
    result = cursor.fetchone()
    cursor.close()
    conn.close()
    return result

# 加入到测试用例中作为断言
from common.db_handler import query_user_by_name

@pytest.mark.parametrize("case", load_case())
def test_add_user(case, token):
    headers = {"Authorization": f"Bearer {token}"}
    res = handler.send(
        method=case["method"],
        url=case["url"],
        headers=headers,
        json=case["json"]
    )
    assert res.status_code == case["expected_status"]
    assert case["expected_msg"] in res.text

    # ✅ 数据库断言
    db_user = query_user_by_name(case["json"]["name"])
    assert db_user is not None
    
# 完整自动化流程
Step 1：登录 → 获取 token → fixture 共享
Step 2：发送带 token 的接口请求（多接口串联）
Step 3：接口响应断言（状态码、文本）
Step 4：数据库查询 → 数据落库断言
Step 5：统一日志、报告、测试数据管理

# 多环境支持
env: dev

dev:
  base_url: http://dev-api.com
test:
  base_url: http://test-api.com
stage:
  base_url: http://stage-api.com

mysql:
  host: localhost
  user: root
  password: 123456
  database: user_db
    
# 动态加载base_url:
with open("config/config.yaml", encoding="utf-8") as f:
    config = yaml.safe_load(f)
env = config["env"]
base_url = config[env]["base_url"]
handler = RequestHandler(base_url=base_url)

# 多接口串联
def test_user_lifecycle(admin_token):
    headers = {"Authorization": f"Bearer {admin_token}"}

    # 1. 添加用户
    res_add = handler.send("post", "/api/user/add", headers=headers, json={"name": "test01", "age": 18})
    assert res_add.status_code == 200

    # 2. 查询用户
    res_get = handler.send("get", "/api/user/detail?name=test01", headers=headers)
    assert "test01" in res_get.text

    # 3. 删除用户
    res_del = handler.send("delete", "/api/user/delete", headers=headers, json={"name": "test01"})
    assert res_del.status_code == 200
    
# 失败重试机制（提高稳定性）pytest.ini增加配置
addopts = --reruns 2 --reruns-delay 2