要在需要多用户身份验证的测试场景中应用此方法,可以通过以下步骤来实现:

  1. 定义多用户凭据:在 conftest.py 中定义多个用户的凭据。
  2. 创建一个 Fixture 用于管理用户身份验证:根据需要动态选择用户并获取相应的令牌或 API 密钥。
  3. 在测试用例中使用动态用户身份:通过参数化测试用例,允许使用不同的用户进行测试。

1. 更新 conftest.py

在 conftest.py 中,定义多个用户的凭据,并创建一个 Fixture 来处理身份验证:

import pytest
import requests

# 定义多个用户的凭据
USER_CREDENTIALS = {
    "user1": {
        "client_id": "user1_client_id",
        "client_secret": "user1_client_secret",
        "token_url": "https://example.com/oauth/token"
    },
    "user2": {
        "client_id": "user2_client_id",
        "client_secret": "user2_client_secret",
        "token_url": "https://example.com/oauth/token"
    }
}

@pytest.fixture(scope='session', params=list(USER_CREDENTIALS.keys()))
def oauth_token(request):
    """获取 OAuth 2.0 访问令牌,支持多用户。"""
    user = request.param
    credentials = USER_CREDENTIALS[user]
    client_id = credentials["client_id"]
    client_secret = credentials["client_secret"]
    token_url = credentials["token_url"]

    payload = {
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret
    }

    response = requests.post(token_url, data=payload)
    response.raise_for_status()
    return response.json()['access_token'], user  # 返回令牌和用户标识

@pytest.fixture(scope='session')
def api_key():
    """返回 API 密钥。"""
    return "your_api_key"  # 替换为实际的 API 密钥

2. 更新测试框架代码

在 main.py 中,更新 test_api 函数以接受 oauth_token 的用户信息:

import yaml
import requests
import pytest
import os
from threading import Lock
from jsonschema import validate, ValidationError

class Context:
    def __init__(self):
        self.data = {}
        self.lock = Lock()

    def update(self, key, value):
        with self.lock:
            self.data[key] = value

    def get(self, key):
        with self.lock:
            return self.data.get(key)

def find_yaml_files(directory, target_file=None):
    """遍历目录,获取指定的 YAML 文件。"""
    yaml_files = []
    
    for root, _, files in os.walk(directory):
        for file in files:
            if file.endswith('.yml') or file.endswith('.yaml'):
                file_path = os.path.join(root, file)
                if target_file is None or target_file in file:
                    yaml_files.append(file_path)

    return yaml_files

def load_test_cases(file_path):
    """从指定的 YAML 文件加载测试用例。"""
    with open(file_path, 'r') as file:
        data = yaml.safe_load(file)
        return data['tests']

def resolve_placeholders(data, context):
    """解析 YAML 数据中的占位符。"""
    if isinstance(data, dict):
        return {key: resolve_placeholders(value, context) for key, value in data.items()}
    elif isinstance(data, list):
        return [resolve_placeholders(item, context) for item in data]
    elif isinstance(data, str) and data.startswith("${") and data.endswith("}"):
        placeholder = data[2:-1]
        return context.get(placeholder, data)
    return data

def run_test_case(base_url, test_case, context, api_key=None, oauth_token=None):
    """执行单个测试用例并验证响应。"""
    url = f"{base_url}{test_case['endpoint']}"
    method = test_case['method']
    expected_status = test_case['expected_status']

    headers = {}
    if api_key:
        headers['Authorization'] = f"Bearer {api_key}"
    elif oauth_token:
        headers['Authorization'] = f"Bearer {oauth_token}"

    for key in context.data:
        if f'{{{key}}}' in url:
            url = url.replace(f'{{{key}}}', str(context.get(key)))

    response = requests.request(method, url, json=test_case.get('payload', {}), headers=headers)
    assert response.status_code == expected_status

    if 'id' in test_case:
        context.update(test_case['id'], response.json().get('id'))

    if 'validate' in test_case:
        validate_response(response.json(), test_case['validate'])

def validate_response(response_data, validation):
    """验证响应数据中的字段值是否满足特定条件。"""
    validation_type = validation.get('type')
    
    if validation_type == 'json':
        schema = validation['schema']
        try:
            validate(instance=response_data, schema=schema)
        except ValidationError as e:
            pytest.fail(f"JSON validation failed: {e.message}")

# 指定要加载的 YAML 文件
def get_test_cases_from_files(directory, target_file=None):
    yaml_files = find_yaml_files(directory, target_file)
    tests = []
    for file_path in yaml_files:
        tests.extend(load_test_cases(file_path))
    return tests

@pytest.mark.parametrize("test_case", get_test_cases_from_files('tests', 'user'))  # 指定文件名部分
def test_api(test_case, oauth_token, api_key):
    """主测试函数,执行指定 YAML 文件中的测试用例。"""
    base_url = "http://localhost:5000"  # 替换为你的 API 基础 URL
    context = Context()
    oauth_token_value, user = oauth_token  # 解包令牌和用户信息

    print(f"Running tests for {user}")

    if 'depends_on' in test_case:
        dependency = test_case['depends_on']
        if context.get(dependency) is None:
            pytest.skip(f"Skipping '{test_case['name']}' due to unmet dependency '{dependency}'.")

    if 'payload' in test_case:
        test_case['payload'] = resolve_placeholders(test_case['payload'], context)

    run_test_case(base_url, test_case, context, api_key=api_key, oauth_token=oauth_token_value)

3. YAML 文件示例

确保测试用例 YAML 文件不需要特别的修改,因为用户凭据在 conftest.py 中处理。

4. 运行测试

使用以下命令运行测试:

pytest main.py

总结

通过这些步骤,你可以在测试框架中支持多用户身份验证。每个测试用例都可以使用不同的用户凭据进行测试,从而验证不同用户的访问权限和操作。通过这种方式,测试变得更加灵活且易于扩展。你可以根据需要进一步调整用户凭据和验证逻辑,以满足特定的测试需求。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐