diff --git a/httpfpt/data/test_data/test_project/only_skip.yml b/httpfpt/data/test_data/test_project/only_skip.yml index 1fa6c38..e72e126 100644 --- a/httpfpt/data/test_data/test_project/only_skip.yml +++ b/httpfpt/data/test_data/test_project/only_skip.yml @@ -30,6 +30,8 @@ test_steps: is_run: skip: True reason: 自定义跳过 + mark: + - test_api request: method: GET url: /skip diff --git a/httpfpt/run.py b/httpfpt/run.py index a40a85f..c1569fe 100644 --- a/httpfpt/run.py +++ b/httpfpt/run.py @@ -109,16 +109,17 @@ def startup( if '=' in i and k in i: run_args.remove(i) run_args.append(f'{k}={v}') - run_args = list(set(run_args)) format_run_args = [] for i in run_args: if '=' in i: i_split = i.split('=') - new_i = i.replace(i_split[1], '"' + f'{i_split[1]}' + '"') + new_i = i.replace(i_split[1], f'"{i_split[1]}"') format_run_args.append(new_i) else: format_run_args.append(i) - run_pytest_command_args = ' '.join(_ for _ in format_run_args) + run_pytest_command_args = ' '.join( + i if os.path.isdir(i) or i.startswith('-') else f'"{i}"' for i in format_run_args + ) log.info( f'开始运行项目：{httpfpt_config.PROJECT_NAME}' if run_path == default_case_path else f'开始运行：{run_path}' diff --git a/httpfpt/schemas/case_data.py b/httpfpt/schemas/case_data.py index 284c034..1cf8b8f 100644 --- a/httpfpt/schemas/case_data.py +++ b/httpfpt/schemas/case_data.py @@ -7,7 +7,10 @@ from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field from typing_extensions import Literal -__all__ = ['CaseData'] +__all__ = [ + 'CaseData', + 'CaseCacheData', +] class ConfigAllureData(BaseModel): @@ -30,6 +33,7 @@ class Config(BaseModel): allure: ConfigAllureData request: ConfigRequestData module: str + mark: list[str] | None = None class StepsRequestData(BaseModel): @@ -141,6 +145,7 @@ class Steps(BaseModel): case_id: str description: str is_run: bool | dict | None = None + mark: list[str] | None = None retry: int | None = None request: StepsRequestData setup: list[StepsSetUpData] | None = None @@ -152,5 +157,8 @@ class CaseData(BaseModel): config: Config test_steps: Steps | list[Steps] + + +class CaseCacheData(CaseData): filename: str | None = None file_hash: str | None = None diff --git a/httpfpt/testcases/test_project/test_api_testcase_template.py b/httpfpt/testcases/test_project/test_api_testcase_template.py index b3715f2..7e98552 100644 --- a/httpfpt/testcases/test_project/test_api_testcase_template.py +++ b/httpfpt/testcases/test_project/test_api_testcase_template.py @@ -4,21 +4,18 @@ import pytest from httpfpt.common.send_request import send_request -from httpfpt.utils.request.case_data_parse import get_request_data -from httpfpt.utils.request.ids_extract import get_ids +from httpfpt.utils.request.case_data_parse import get_testcase_data -request_data = get_request_data(filename='api_testcase_template.yaml') -allure_text = request_data[0]['config']['allure'] -request_ids = get_ids(request_data) +allure_data, ddt_data, ids = get_testcase_data(filename='api_testcase_template.yaml') -@allure.epic(allure_text['epic']) -@allure.feature(allure_text['feature']) +@allure.epic(allure_data['epic']) +@allure.feature(allure_data['feature']) class TestApiTestcaseTemplate: """ApicaseTemplate""" - @allure.story(allure_text['story']) - @pytest.mark.parametrize('data', request_data, ids=request_ids) + @allure.story(allure_data['story']) + @pytest.mark.parametrize('data', ddt_data, ids=ids) def test_api_testcase_template(self, data): """api_testcase_template""" send_request.send_request(data) diff --git a/httpfpt/testcases/test_project/test_only_skip.py b/httpfpt/testcases/test_project/test_only_skip.py index 518d3b4..9152243 100644 --- a/httpfpt/testcases/test_project/test_only_skip.py +++ b/httpfpt/testcases/test_project/test_only_skip.py @@ -4,21 +4,18 @@ import pytest from httpfpt.common.send_request import send_request -from httpfpt.utils.request.case_data_parse import get_request_data -from httpfpt.utils.request.ids_extract import get_ids +from httpfpt.utils.request.case_data_parse import get_testcase_data -request_data = get_request_data(filename='only_skip.yml') -allure_text = request_data[0]['config']['allure'] -request_ids = get_ids(request_data) +allure_data, ddt_data, ids = get_testcase_data(filename='only_skip.yml') -@allure.epic(allure_text['epic']) -@allure.feature(allure_text['feature']) +@allure.epic(allure_data['epic']) +@allure.feature(allure_data['feature']) class TestOnlySkip: """OnlySkip""" - @allure.story(allure_text['story']) - @pytest.mark.parametrize('data', request_data, ids=request_ids) + @allure.story(allure_data['story']) + @pytest.mark.parametrize('data', ddt_data, ids=ids) def test_only_skip(self, data): """only_skip""" send_request.send_request(data) diff --git a/httpfpt/testcases/test_project/test_upload_file.py b/httpfpt/testcases/test_project/test_upload_file.py index f8168de..68c3860 100644 --- a/httpfpt/testcases/test_project/test_upload_file.py +++ b/httpfpt/testcases/test_project/test_upload_file.py @@ -4,21 +4,18 @@ import pytest from httpfpt.common.send_request import send_request -from httpfpt.utils.request.case_data_parse import get_request_data -from httpfpt.utils.request.ids_extract import get_ids +from httpfpt.utils.request.case_data_parse import get_testcase_data -request_data = get_request_data(filename='upload_file.json') -allure_text = request_data[0]['config']['allure'] -request_ids = get_ids(request_data) +allure_data, ddt_data, ids = get_testcase_data(filename='upload_file.json') -@allure.epic(allure_text['epic']) -@allure.feature(allure_text['feature']) +@allure.epic(allure_data['epic']) +@allure.feature(allure_data['feature']) class TestUploadFile: """UploadFile""" - @allure.story(allure_text['story']) - @pytest.mark.parametrize('data', request_data, ids=request_ids) + @allure.story(allure_data['story']) + @pytest.mark.parametrize('data', ddt_data, ids=ids) def test_upload_file(self, data): """upload_file""" send_request.send_request(data) diff --git a/httpfpt/utils/case_auto_generator.py b/httpfpt/utils/case_auto_generator.py index f74d2bd..7359a54 100644 --- a/httpfpt/utils/case_auto_generator.py +++ b/httpfpt/utils/case_auto_generator.py @@ -68,21 +68,18 @@ def auto_generate_testcases(rewrite: bool = False) -> None: import pytest from httpfpt.common.send_request import send_request -from httpfpt.utils.request.case_data_parse import get_request_data -from httpfpt.utils.request.ids_extract import get_ids +from httpfpt.utils.request.case_data_parse import get_testcase_data -request_data = get_request_data(filename='{file_property[0]}') -allure_text = request_data[0]['config']['allure'] -request_ids = get_ids(request_data) +allure_data, ddt_data, ids = get_testcase_data(filename='{file_property[0]}') -@allure.epic(allure_text['epic']) -@allure.feature(allure_text['feature']) +@allure.epic(allure_data['epic']) +@allure.feature(allure_data['feature']) class {testcase_class_name}: """{testcase_class_name.replace('Test', '')}""" - @allure.story(allure_text['story']) - @pytest.mark.parametrize('data', request_data, ids=request_ids) + @allure.story(allure_data['story']) + @pytest.mark.parametrize('data', ddt_data, ids=ids) def {testcase_func_name}(self, data): """{create_file_root_name}""" send_request.send_request(data) @@ -98,7 +95,7 @@ def {testcase_func_name}(self, data): case_path = os.path.join( httpfpt_path.testcase_dir, httpfpt_config.PROJECT_NAME, new_testcase_filename ) - new_testcase_dir = Path(case_path).parent + new_testcase_dir = Path(case_path).parent # type: ignore if not new_testcase_dir.exists(): new_testcase_dir.mkdir(parents=True, exist_ok=True) with open(case_path, 'w', encoding='utf-8') as f: diff --git a/httpfpt/utils/request/case_data_parse.py b/httpfpt/utils/request/case_data_parse.py index b5d06c8..2bff358 100644 --- a/httpfpt/utils/request/case_data_parse.py +++ b/httpfpt/utils/request/case_data_parse.py @@ -2,12 +2,12 @@ # -*- coding: utf-8 -*- from __future__ import annotations -import copy import json import sys from collections import defaultdict -from typing import Any + +import pytest from pydantic import ValidationError @@ -15,9 +15,10 @@ from httpfpt.common.log import log from httpfpt.common.yaml_handler import read_yaml from httpfpt.db.redis_db import redis_client -from httpfpt.schemas.case_data import CaseData +from httpfpt.schemas.case_data import CaseCacheData from httpfpt.utils.file_control import get_file_hash, get_file_property, search_all_case_data_files from httpfpt.utils.pydantic_parser import parse_error +from httpfpt.utils.request.ids_extract import get_ids def clean_cache_data(clean_cache: bool) -> None: @@ -58,7 +59,7 @@ def case_data_init(pydantic_verify: bool) -> None: count: int = 0 for case_data in case_data_list: try: - CaseData.model_validate(json.loads(case_data)) + CaseCacheData.model_validate(json.loads(case_data)) except ValidationError as e: count += parse_error(e) if count > 0: @@ -118,9 +119,9 @@ def case_id_unique_verify() -> None: redis_client.rset(f'{redis_client.prefix}:case_id_list', str(all_case_id)) -def get_request_data(*, filename: str) -> list[dict[str, Any]]: +def get_testcase_data(*, filename: str) -> tuple[dict, list, list]: """ - 获取用于测试用例数据驱动的请求数据 + 获取测试用例数据 :param filename: 测试用例数据文件名称 :return: @@ -129,25 +130,60 @@ def get_request_data(*, filename: str) -> list[dict[str, Any]]: config_error = f'请求测试用例数据文件 {filename} 缺少 config 信息, 请检查测试用例文件内容' test_steps_error = f'请求测试用例数据文件 {filename} 缺少 test_steps 信息, 请检查测试用例文件内容' - if case_data.get('config') is None: + config = case_data.get('config') + if config is None: raise RequestDataParseError(config_error) - cases = case_data.get('test_steps') - if cases is None: + steps = case_data.get('test_steps') + if steps is None: raise RequestDataParseError(test_steps_error) - if isinstance(cases, dict): - return [case_data] - elif isinstance(cases, list): - case_list = [] - for case in cases: + allure_data = case_data['config']['allure'] + if isinstance(steps, dict): + ids = get_ids(case_data) + mark = get_testcase_mark(case_data) + if mark is not None: + ddt_data = pytest.param(case_data, marks=[getattr(pytest.mark, m) for m in mark]) + else: + ddt_data = case_data + return allure_data, [ddt_data], ids + elif isinstance(steps, list): + _ddt_data_list = [] + marked_ddt_data_list = [] + for case in steps: if isinstance(case, dict): - test_steps = {'test_steps': case} - data = copy.deepcopy(case_data) - data.update(test_steps) - case_list.append(data) + _case_data = {'config': config, 'test_steps': case} + _ddt_data_list.append(_case_data) + mark = get_testcase_mark(_case_data) + if mark is not None: + marked_ddt_data_list.append(pytest.param(_case_data, marks=[getattr(pytest.mark, m) for m in mark])) + else: + marked_ddt_data_list.append(_case_data) else: raise RequestDataParseError(test_steps_error) - return case_list + ids = get_ids(_ddt_data_list) + return allure_data, marked_ddt_data_list, ids else: raise RequestDataParseError(f'请求测试用例数据文件 {filename} 格式错误, 请检查用例数据文件内容') + + +def get_testcase_mark(case_data: dict) -> list[str] | None: + try: + mark = case_data['test_steps']['mark'] + except (KeyError, TypeError): + try: + mark = case_data['config']['mark'] + except (KeyError, TypeError): + mark = None + if mark is not None: + if not isinstance(mark, list): + raise RequestDataParseError( + '测试用例数据解析失败, 参数 test_steps:mark 或 config:mark 不是有效的 list 类型' + ) + else: + for m in mark: + if not isinstance(m, str): + raise RequestDataParseError( + '测试用例数据解析失败, 参数 test_steps:mark 或 config:mark 不是有效的 list[str] 类型' + ) + return mark diff --git a/httpfpt/utils/request/ids_extract.py b/httpfpt/utils/request/ids_extract.py index d7e3f8d..240d614 100644 --- a/httpfpt/utils/request/ids_extract.py +++ b/httpfpt/utils/request/ids_extract.py @@ -1,9 +1,11 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- +from __future__ import annotations + from httpfpt.common.errors import RequestDataParseError -def get_ids(request_data: list) -> list: +def get_ids(request_data: dict | list) -> list: """ 从请求数据获取数据驱动下的 ids 数据 @@ -11,12 +13,16 @@ def get_ids(request_data: list) -> list: :return: """ ids = [] - for data in request_data: - try: - module = data['config']['module'] - name = data['test_steps']['name'] - case_id = data['test_steps']['case_id'] - except KeyError as e: - raise RequestDataParseError('测试用例 ids 获取失败, 请检查测试用例数据是否符合规范: {}'.format(e)) - ids.append('module: {}, name: {}, case_id: {}'.format(module, name, case_id)) + try: + if isinstance(request_data, dict): + module = request_data['config']['module'] + case_id = request_data['test_steps']['case_id'] + ids.append(f'module: {module}, case_id: {case_id}') + else: + for data in request_data: + module = data['config']['module'] + case_id = data['test_steps']['case_id'] + ids.append(f'module: {module}, case_id: {case_id}') + except KeyError as e: + raise RequestDataParseError('测试用例 ids 获取失败, 请检查测试用例数据是否符合规范: {}'.format(e)) return ids diff --git a/httpfpt/utils/request/request_data_parse.py b/httpfpt/utils/request/request_data_parse.py index 4fd86d0..1a000e5 100644 --- a/httpfpt/utils/request/request_data_parse.py +++ b/httpfpt/utils/request/request_data_parse.py @@ -195,7 +195,7 @@ def retry(self) -> int | None: retry = None if retry is not None: if not isinstance(retry, int): - raise RequestDataParseError(_error_msg('参数 config:request:retry 不是有效的 int 类型')) + raise RequestDataParseError(_error_msg('参数 test_steps:retry 或 config:retry 不是有效的 int 类型')) return retry @property @@ -211,6 +211,20 @@ def module(self) -> str: raise RequestDataParseError(_error_msg('参数 config:module 不是有效的 str 类型')) return module + @property + def mark(self) -> list | None: + try: + mark = self.request_data['test_steps']['mark'] + except _RequestDataParamGetError: + try: + mark = self.request_data['config']['mark'] + except _RequestDataParamGetError: + mark = None + if mark is not None: + if not isinstance(mark, list): + raise RequestDataParseError(_error_msg('参数 test_steps:mark 或 config:mark 不是有效的 list 类型')) + return mark + @property def test_steps(self) -> dict | list: try: @@ -370,7 +384,9 @@ def headers(self) -> dict | None: headers = None else: if not isinstance(headers, dict): - raise RequestDataParseError(_error_msg('参数 test_steps:request:headers 不是有效的 dict 类型')) + raise RequestDataParseError( + _error_msg('参数 test_steps:request:headers 或 config:request:headers 不是有效的 dict 类型') + ) if headers is not None: if len(headers) == 0: raise RequestDataParseError(_error_msg('参数 test_steps:request:headers 为空'))