Giới thiệu

Tôi đang lên kế hoạch tập trung vào việc bao phủ kiểm thử API, vì vậy cần xây dựng một framework kiểm thử. Sau khi cân nhắc kỹ lưỡng, tôi muốn tạo ra một giải pháp độc đáo hơn so với các hệ thống truyền thống.

Một số yêu cầu chính của hệ thống:

• Kiểm thử API yêu cầu hiệu suất cao để nhận phản hồi nhanh chóng, trong khi số lượng API thường rất lớn và ngày càng tăng, nên cải thiện tốc độ thực thi là điều cần thiết
• Các testcase API có thể được sử dụng cho các bài kiểm tra tải đơn giản, và kiểm thử tải cần hỗ trợ đồng thời
• Các testcase API chứa nhiều nội dung lặp lại, người kiểm thử nên chỉ cần tập trung vào thiết kế testcase, các công việc lặp lại nên được tự động hóa
• Tích hợp pytest và allure vì tính tiện dụng của chúng
• Sử dụng YAML cho testcase để dữ liệu có thể ánh xạ trực tiếp thành dữ liệu yêu cầu, giúp viết testcase giống như làm bài tập điền khuyết, thuận tiện cho việc giới thiệu cho các thành viên không có kinh nghiệm lập trình

Vì tôi rất hứng thú với coroutine trong Python và đã học một thời gian, hy vọng áp dụng kiến thức này vào thực tế, nên tôi quyết định sử dụng aiohttp cho các yêu cầu HTTP. Tuy nhiên, pytest không hỗ trợ event loop, nên việc kết hợp chúng cần thêm công sức.

Thiết kế hệ thống

Tôi chia toàn bộ quá trình thành hai phần chính:

1. Đọc testcase YAML, gửi yêu cầu HTTP đến API, thu thập dữ liệu kiểm thử
2. Dựa trên dữ liệu kiểm thử, sinh testcase động được pytest chấp nhận, sau đó thực thi và tạo báo cáo

Phần đầu tiên (Toàn bộ quy trình yêu cầu bất đồng bộ không chặn)

Định nghĩa mẫu testcase YAML

Đây là mẫu testcase đơn giản mà tôi thiết kế. Ưu điểm là tên tham số phù hợp trực tiếp với phương thức aiohttp.ClientSession().request(method,url,**kwargs), giúp truyền dữ liệu trực tiếp mà không cần chuyển đổi phức tạp.

args:
  - post
  - /api/create
kwargs:
  -
    caseName: Tạo mới tài nguyên
    data:
      title: ${generate_random_string(8)}
validator:
  -
    json:
      status: success

Hàm đọc file YAML bất đồng bộ

Sử dụng thư viện aiofiles để đọc file bất đồng bộ, đảm bảo tiến trình chính không bị chặn khi đọc testcase YAML.

async def load_yaml_config(directory='', filename=''):
    """
    Đọc file YAML bất đồng bộ và xử lý các giá trị đặc biệt
    :param filename: tên file
    :return: dữ liệu từ file YAML
    """
    if directory:
        filepath = os.path.join(directory, filename)
    
    async with aiofiles.open(filepath, 'r', encoding='utf-8', errors='ignore') as file_handle:
        content = await file_handle.read()

    parsed_data = yaml.safe_load(content)

    # Biểu thức regex để nhận diện cú pháp đặc biệt
    func_pattern = re.compile(r'^\${([A-Za-z_]+\w*\(.*\))}$')
    alt_func_pattern = re.compile(r'^\${(.*)}$')
    default_pattern = re.compile(r'^\$\((.*)\)$')

    def process_recursive(data_structure):
        """
        Duyệt đệ quy cấu trúc dữ liệu, xử lý các giá trị đặc biệt
        :param data_structure: cấu trúc dữ liệu đầu vào
        :return: dữ liệu đã xử lý
        """
        if isinstance(data_structure, (list, tuple)):
            for idx, item in enumerate(data_structure):
                data_structure[idx] = process_recursive(item) or item
        elif isinstance(data_structure, dict):
            for key, value in data_structure.items():
                data_structure[key] = process_recursive(value) or value
        elif isinstance(data_structure, (str, bytes)):
            match = func_pattern.match(data_structure)
            if not match:
                match = alt_func_pattern.match(data_structure)
            if match:
                return eval(match.group(1))
            if not match:
                match = default_pattern.match(data_structure)
            if match:
                parent_key, child_key = match.group(1).split(':')
                return config_store.default_configs.get(parent_key).get(child_key)
            return data_structure
        
        return data_structure

    processed_result = process_recursive(parsed_data)
    return CustomDict(processed_result)

Xử lý yêu cầu HTTP bất đồng bộ

Sử dụng aiohttp để thực hiện yêu cầu HTTP, đảm bảo không bị chặn trong quá trình gửi yêu cầu mạng.

async def execute_http_request(base_url, *request_args, **request_kwargs):
    """
    Xử lý yêu cầu HTTP
    :param base_url: địa chỉ dịch vụ
    :param request_args: các đối số vị trí
    :param request_kwargs: các đối số từ khóa
    :return: dữ liệu phản hồi
    """
    http_method, endpoint = request_args
    payload = request_kwargs.get('payload') or request_kwargs.get('query_params') or request_kwargs.get('json_payload') or {}

    # Thêm token vào header
    request_kwargs.setdefault('headers', {}).update({'auth_token': auth_manager.token})
    
    full_url = ''.join([base_url, endpoint])

    async with ClientSession() as client_session:
        async with client_session.request(http_method, full_url, **request_kwargs) as response:
            processed_response = await handle_response(response)
            return {
                'response_data': processed_response,
                'full_url': full_url,
                'request_payload': payload
            }

Thu thập dữ liệu kiểm thử

Sử dụng asyncio.Semaphore để kiểm soát mức độ đồng thời nhằm tránh quá tải hệ thống.

async def execution_entrypoint(test_scenarios, event_loop, concurrency_limit=None):
    """
    Điểm vào thực thi HTTP
    :param test_scenarios: danh sách testcase
    :param concurrency_limit: giới hạn đồng thời
    :return: kết quả thực thi
    """
    results = CustomDict()
    
    async with ClientSession(loop=event_loop, cookie_jar=CookieJar(unsafe=True), headers={'auth_token': auth_manager.token}) as session:
        await authenticate_system(session)
        
        if concurrency_limit:
            async with concurrency_limit:
                for scenario in test_scenarios:
                    execution_data = await single_execution(session, case_name=scenario)
                    results.setdefault(execution_data.pop('case_directory'), CustomList()).append(execution_data)
        else:
            for scenario in test_scenarios:
                execution_data = await single_execution(session, case_name=scenario)
                results.setdefault(execution_data.pop('case_directory'), CustomList()).append(execution_data)

        return results


async def single_execution(session, case_directory='', case_name=''):
    """
    Thực thi một testcase hoàn chỉnh: đọc .yml, gửi yêu cầu HTTP, trả về kết quả
    Tất cả thao tác đều bất đồng bộ không chặn
    :param session: phiên làm việc
    :param case_directory: thư mục testcase
    :param case_name: tên testcase
    :return: kết quả thực thi
    """
    project_identifier = case_name.split(os.sep)[1]
    target_domain = config_store.service_urls.get(project_identifier)
    test_configuration = await load_yaml_config(dir=case_directory, file=case_name)
    
    execution_result = CustomDict({
        'case_directory': os.path.dirname(case_name),
        'endpoint_path': test_configuration.args[1].replace('/', '_'),
    })
    
    if isinstance(test_configuration.kwargs, list):
        for index, individual_data in enumerate(test_configuration.kwargs):
            step_description = individual_data.pop('caseName')
            response = await execute_http_request(session, target_domain, *test_configuration.args, **individual_data)
            response.update({'case_description': step_description})
            execution_result.setdefault('responses', CustomList()).append({
                'response': response,
                'validation_rules': test_configuration.validation_rules[index]
            })
    else:
        step_description = test_configuration.kwargs.pop('caseName')
        response = await execute_http_request(session, target_domain, *test_configuration.args, **test_configuration.kwargs)
        response.update({'case_description': step_description})
        execution_result.setdefault('responses', CustomList()).append({
            'response': response,
            'validation_rules': test_configuration.validation_rules
        })

    return execution_result

Chạy vòng lặp sự kiện chính

Vòng lặp sự kiện chịu trách nhiệm thực thi các coroutine và trả về kết quả.

def main_execution(test_scenarios):
    """
    Hàm chính của vòng lặp sự kiện, chịu trách nhiệm thực thi tất cả yêu cầu API
    :param test_scenarios: danh sách testcase
    :return: kết quả thực thi
    """
    event_loop = asyncio.get_event_loop()
    rate_limiter = asyncio.Semaphore(config_store.concurrent_requests)
    
    task = event_loop.create_task(execution_entrypoint(test_scenarios, event_loop, rate_limiter))
    
    try:
        event_loop.run_until_complete(task)
    finally:
        event_loop.close()

    return task.result()

Phần thứ hai: Sinh testcase pytest động

Cơ chế hoạt động của pytest

Pytest tìm kiếm file conftest.py trong thư mục hiện tại, sau đó tìm các file .py bắt đầu hoặc kết thúc bằng 'test' theo tham số dòng lệnh, phân tích fixture, và tìm kiếm lớp, phương thức theo quy tắc cụ thể.

File bootstrap để kích hoạt pytest

Tạo file hướng dẫn để kích hoạt pytest, sử dụng pytest.skip() để bỏ qua phương thức kiểm thử.

# test_initializer.py
import pytest

class TestBootstrap(object):

    def test_initialize(self):
        pytest.skip('Đây là phương thức khởi tạo kiểm thử, không thực thi')

Sử dụng fixture để sinh testcase động

Sử dụng fixture có phạm vi session để sinh testcase trước khi pytest bắt đầu thực thi.

# conftest.py
@pytest.fixture(scope='session')
def generate_test_scenarios(request):
    """
    Xử lý sinh testcase
    :param request: đối tượng yêu cầu
    :return: danh sách testcase
    """
    target_directory = request.config.getoption("--rootdir")
    specific_file = request.config.getoption("--tf")
    environment = request.config.getoption("--te")
    
    scenarios = []
    if specific_file:
        scenarios = [specific_file]
    else:
        if os.path.isdir(target_directory):
            for root_path, subdirs, files_list in os.walk(target_directory):
                if re.match(r'\w+', root_path):
                    if files_list:
                        scenarios.extend([os.path.join(root_path, file) for file in files_list if file.endswith('yml')])

    test_data = main_execution(scenarios)

    template_content = """
import allure

from conftest import TestCaseMetaClass


@allure.feature('{} API kiểm thử (dự án {})')
class Test{}API(object, metaclass=TestCaseMetaClass):

    scenario_data = {}
"""
    
    generated_files = []
    if os.path.isdir(target_directory):
        for root_path, subdirs, files_list in os.walk(target_directory):
            if not ('.' in root_path or '__' in root_path):
                if files_list:
                    case_identifier = os.path.basename(root_path)
                    project_identifier = os.path.basename(os.path.dirname(root_path))
                    output_file = os.path.join(root_path, 'test_{}.py'.format(case_identifier))
                    
                    with open(output_file, 'w', encoding='utf-8') as file_writer:
                        file_writer.write(template_content.format(case_identifier, project_identifier, case_identifier.title(), test_data.get(root_path)))
                    
                    generated_files.append(output_file)

    if specific_file:
        temp_path = os.path.dirname(specific_file)
        python_file = os.path.join(temp_path, 'test_{}.py'.format(os.path.basename(temp_path)))
    else:
        python_file = target_directory

    pytest.main([
        '-v',
        python_file,
        '--alluredir',
        'report',
        '--te',
        environment,
        '--capture',
        'no',
        '--disable-warnings',
    ])

    for file in generated_files:
        os.remove(file)

    return generated_files

Meta class để sinh testcase động

CaseMetaClass là một meta class đọc thuộc tính scenario_data và sinh phương thức động.

function_template = """
def {}(self, response, validation_rule):
    with allure.step(response.pop('case_description')):
        validate_response(response, validation_rule)"""


class TestCaseMetaClass(type):
    """
    Tự động sinh testcase dựa trên kết quả gọi API
    """

    def __new__(cls, class_name, base_classes, attributes):
        scenario_data = attributes.pop('scenario_data')
        
        for item in scenario_data:
            endpoint = item.pop('endpoint_path')
            method_name = 'test_' + endpoint
            test_input_data = [tuple(x.values()) for x in item.get('responses')]
            
            function_obj = create_function_from_string(function_template.format(method_name),
                                                      namespace={'validate_response': validate_response, 'allure': allure})
            
            decorated_function = allure.story('{}'.format(endpoint.replace('_', '/')))(function_obj)
            attributes[method_name] = pytest.mark.parametrize('response,validation_rule', test_input_data)(decorated_function)

        return super().__new__(cls, class_name, base_classes, attributes)

Hàm tạo hàm từ chuỗi

Hàm này tạo đối tượng hàm từ biểu thức chuỗi.

def create_function_from_string(func_expression, namespace={}):
    """
    Tạo đối tượng hàm động, thiết lập phạm vi hàm mặc định là builtins.__dict__, sau đó hợp nhất namespace tùy chỉnh
    :param func_expression: biểu thức hàm, ví dụ 'def sample_func(): return "result"'
    :return: đối tượng hàm
    """
    builtins.__dict__.update(namespace)
    module_compiled = compile(func_expression, '', 'exec')
    function_compiled = [code for code in module_compiled.co_consts if isinstance(code, types.CodeType)][0]
    return types.FunctionType(function_compiled, builtins.__dict__)

Tiện ích tự động tạo file YAML

Giải pháp tự động sinh testcase từ Swagger

Để giảm thiểu công việc lặp lại trong quá trình tạo testcase, tôi phát triển chức năng tự động tạo file YAML từ dữ liệu Swagger.

import re
import os
import sys
from requests import Session

yaml_template = """
args:
  - {http_method}
  - {api_endpoint}
kwargs:
  -
    caseName: {description}
    {payload_type}:
        {parameter_definitions}
validator:
  -
    json:
      status: success
"""


def generate_swagger_testcases(swagger_endpoint, project_identifier):
    """
    Tự động tạo template testcase YAML từ dữ liệu JSON của Swagger
    :param swagger_endpoint: URL Swagger
    :param project_identifier: tên dự án
    :return: None
    """
    response = Session().request('get', swagger_endpoint).json()
    endpoints_data = response.get('paths')

    working_directory = os.getcwd()
    project_path = os.path.join(working_directory, project_identifier)

    if not os.path.exists(project_path):
        os.mkdir(project_path)

    for path, methods in endpoints_data.items():
        path_parts = re.split(r'[/]+', path)
        directory, *filename = path_parts[1:]

        if filename:
            filename = ''.join([part.title() for part in filename])
        else:
            filename = directory

        filename += '.yml'

        directory_path = os.path.join(project_path, directory)

        if not os.path.exists(directory_path):
            os.mkdir(directory_path)

        os.chdir(directory_path)

        if len(methods) > 1:
            methods = {'post': methods.get('post')}
        
        for method_name, method_info in methods.items():
            http_method = method_name
            api_endpoint = path
            description = method_info.get('summary') or method_info.get('description')
            payload_type = 'query_params' if method_name == 'get' else 'payload'
            parameters = method_info.get('parameters')

            parameter_string = ''
            try:
                for param in parameters:
                    parameter_string += param.get('name')
                    parameter_string += ': \n'
                    parameter_string += ' ' * 8
            except TypeError:
                parameter_string += '{}'

        output_filepath = os.path.join(directory_path, filename)

        with open(output_filepath, 'w', encoding='utf-8') as file_writer:
            file_writer.write(yaml_template.format(
                http_method=http_method,
                api_endpoint=api_endpoint,
                description=description,
                payload_type=payload_type,
                parameter_definitions=parameter_string
            ))

        os.chdir(project_path)

Bây giờ khi bắt đầu kiểm thử API cho một dự án có tích hợp Swagger, hệ thống có thể tự động tạo khung dự án trong vài giây, nhân viên kiểm thử chỉ cần tập trung vào thiết kế testcase cụ thể, điều này rất có ý nghĩa cho việc推广应用 trong đội ngũ kiểm thử.