python中动态创建函数

发表于 | 分类于
字数统计: 1.2k 字 | 阅读时长 ≈ 5 分钟

动态创建一个函数（带有参数变量）

python中动态创建函数。有以下几种方案可以进行动态创建：

  1. 可以使用闭包
  2. lambda表达式
  3. 可以使用types模块中的FunctionType

其中第三种比较复杂。

在进行接口自动化测试框架的搭建过程中，需要将测试用例数据传入一个名称以test_开头的方法模板中，然后将该方法模板通过反射注入到测试模块当中。当运行pytest时就可以执行该方法。

此前已经通过模拟闭包解决了方法模板中未正确引用测试数据的问题，但是仍有一个问题，那就是目前定义的方法模板中无法加入参数（测试数据参数是在函数内调用的，此处指的是无法在函数上进行传参）。

如果能够实现在函数上传参，就可以在conftest.py文件中定义fixture，然后将该fixture在方法模板的函数上作为参数名称进行引用，达到使用fixture的目的。

以下代码的原理参考：https://zhuanlan.zhihu.com/p/386276353

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
# conftest.py
import os
import sys
import types
from inspect import Signature, Parameter
from pathlib import Path
from typing import Sequence, Callable, Mapping, Any

import allure
import pytest
from _pytest.python import Module

from utils.path_util import path_util
from utils.yaml_util import YamlUtil
# 创建函数的方法
def create_function_parameters(
        func: Callable[[Mapping[str, Any]], Any],
        parameters: Sequence[Parameter],
        documentation=None,
        func_name=None,
        func_filename=None):
    '''
    :param func: 函数体内执行的方法
    :param parameters: 函数上的参数
    :param documentation: 函数的注释
    :param func_name: 函数的名称
    :param func_filename: 函数所属的文件名称
    :return:
    '''
    new_signature = Signature(parameters) # Checks the parameter consistency
    def pass_locals():
        return dict_func(locals()) # noqa: F821 TODO
    code = pass_locals.__code__
    mod_co_argcount = len(parameters)
    mod_co_nlocals = len(parameters)
    mod_co_varnames = tuple(param.name for param in parameters)
    mod_co_name = func_name or code.co_name
    if func_filename:
        mod_co_filename = func_filename
        mod_co_firstlineno = 1
    else:
        mod_co_filename = code.co_filename
        mod_co_firstlineno = code.co_firstlineno
    if sys.version_info >= (3, 8):
        modified_code = code.replace(
            co_argcount=mod_co_argcount,
            co_nlocals=mod_co_nlocals,
            co_varnames=mod_co_varnames,
            co_filename=mod_co_filename,
            co_name=mod_co_name,
            co_firstlineno=mod_co_firstlineno,
            )
    else:
        modified_code = types.CodeType(
            mod_co_argcount,
            code.co_kwonlyargcount,
            mod_co_nlocals,
            code.co_stacksize,
            code.co_flags,
            code.co_code,
            code.co_consts,
            code.co_names,
            mod_co_varnames,
            mod_co_filename,
            mod_co_name,
            mod_co_firstlineno,
            code.co_lnotab
            )
    default_arg_values = tuple(p.default for p in parameters if p.default != Parameter.empty) #!argdefs "starts from the right"/"is right-aligned"
    modified_func = types.FunctionType(modified_code, {'dict_func': func, 'locals': locals}, name=func_name,argdefs=default_arg_values)
    modified_func.__doc__ = documentation
    modified_func.__signature__ = new_signature
    return modified_func

def pytest_collect_file(file_path: Path, parent):
    if file_path.suffix == ".yaml" and "mall" in str(file_path):
        # 得到文件名称（不含后缀）
        yaml_file = os.path.splitext(os.path.basename(file_path))[0]
        pytest_module = Module.from_parent(parent, path=file_path)
        # 动态创建module
        myModule = types.ModuleType(file_path.stem)
        # 解析yaml内容
        yaml_info_dict = YamlUtil(file_path).read_yaml()
        # 读取fixture的值
        p_list = yaml_info_dict["config"]["fixture"]
        # 将fixture加入到数组中
        parameters = []
        for p in p_list:
            parameters.append(Parameter(f'{p}', Parameter.POSITIONAL_OR_KEYWORD))
        # 获取测试步骤数据
        yaml_steps = yaml_info_dict["teststeps"]
        # 遍历测试用例步骤
        # flag用于递增，加在方法名称上用来区分不同的方法名
        flag = 1
        for step in yaml_steps:
            # 定义要动态创建的方法，使用了闭包，传递step参数
            def function_outer(**kwargs):
                def function_template(*arg, **kwarg):
                    print(kwargs)
                return function_template

            func = function_outer(**step)
            # 创建函数，主要用于动态传递参数
            f = create_function_from_parameters(func=func,
                                           parameters=parameters,
                                           documentation=None,
                                           func_name="test_"+yaml_file+str(flag),
                                           func_filename="test_case.py")
            setattr(myModule, "test_"+yaml_file+str(flag), f)
            flag = flag+1
        # 重写_getobj属性，使用了lambda匿名函数
        pytest_module._getobj = lambda: myModule
        return pytest_module


@pytest.fixture(scope="function")
def request_hook():
    print("--------------request------")


@pytest.fixture(scope="function")
def response_hook():
    print("--------------response------")

yaml数据如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
config:
  variables:
    password: macro123
    username: admin
  fixture: [request_hook, response_hook]
teststeps:
-
  func_name: login
  skip: 手动跳过
  epic: mall商城
  feature: 登录
  story: 登录
  title: 正常账号密码登录
  requests:
    url: ${host}/admin/login
    method: post
    headers:
    params:
    data:
    json:
      password: ${password}
      username: ${username}
#  setup_hooks: ${__setup(requests_info)}
  teardown_hooks:
  extract:
    token: ['jsonpath', '$.data.token']
  validate:
    code: 20022
    message: 成功111
-
  func_name: login
  skip: 手动跳过2
  epic: mall商城
  feature: 登录
  story: 登录
  title: 正常账号密码登录
  requests:
    url: ${host}/admin/login
    method: post
    headers:
    params:
    data:
    json:
      password: ${password}
  teardown_hooks:
  extract:
    token: ['jsonpath', '$.data.token']
  validate:
    code: 20022
    message: 成功111

运行后结果如下：

1
2
3
4
5
6
7
8
data/mall/login.yaml::test_login1 --------------request------
--------------response------
{'func_name': 'login', 'skip': '手动跳过', 'epic': 'mall商城', 'feature': '登录', 'story': '登录', 'title': '正常账号密码登录', 'requests': {'url': '${host}/admin/login', 'method': 'post', 'headers': None, 'params': None, 'data': None, 'json': {'password': '${password}', 'username': '${username}'}}, 'teardown_hooks': None, 'extract': {'token': ['jsonpath', '$.data.token']}, 'validate': {'code': 20022, 'message': '成功111'}}
PASSED
data/mall/login.yaml::test_login2 --------------request------
--------------response------
{'func_name': 'login', 'skip': '手动跳过2', 'epic': 'mall商城', 'feature': '登录', 'story': '登录', 'title': '正常账号密码登录', 'requests': {'url': '${host}/admin/login', 'method': 'post', 'headers': None, 'params': None, 'data': None, 'json': {'password': '${password}'}}, 'teardown_hooks': None, 'extract': {'token': ['jsonpath', '$.data.token']}, 'validate': {'code': 20022, 'message': '成功111'}}
PASSED

在yaml文件的config下定义了fixture，然后在conftest.py文件中写了对应的两个fixture，运行后控制台打印输入了fixture中的内容，说明调用fixture成功了。将打印测试数据换成对测试数据进行前置处理、发送请求、后置处理、断言等逻辑代码，即可执行接口自动化测试。

------------- End -------------