添加测试用例mark动态标记

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

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}'

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

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)

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)

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)

httpfpt/utils/case_auto_generator.py

@@ -68,21 +68,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='{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 @@
                     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:

httpfpt/utils/request/case_data_parse.py

@@ -2,22 +2,23 @@
 # -*- 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
 
 from httpfpt.common.errors import RequestDataParseError
 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

httpfpt/utils/request/ids_extract.py

@@ -1,22 +1,28 @@
 #!/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 数据
 
     :param request_data: 请求数据
     :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

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 为空'))