feat(parser): 支持解析模块/类/函数多层 mark 并去重

- 改进 parse_case_attributes 使用 item.iter_markers() 包含全层级 mark
- 实现 tags 的多层级收集（模块/类/函数）与去重逻辑
- 支持 owner/extra_attributes/coding_testcase_id 的正确解析
- 移除 getattr/hasattr，改用 try/except 与直接属性访问
- 新增测试用例验证多层级 mark 解析功能
- 添加集成测试在 test_collector.py 中验证实际解析结果

Author: zixingdeng

pytest/src/testsolar_pytestx/collector.py

@@ -88,7 +88,9 @@ def collect_testcases(
     run_all_cases = os.getenv("TESTSOLAR_TTP_RUNALLCASES", "").lower() in ["1", "true"]
     if run_all_cases:
         logger.info("[Load] 执行一条命令运行当前项目的所有用例")
-        load_result.Tests.extend([TestCase(Name=selector, Attributes={}) for selector in entry_param.TestSelectors])
+        load_result.Tests.extend(
+            [TestCase(Name=selector, Attributes={}) for selector in entry_param.TestSelectors]
+        )
         reporter = FileReporter(Path(entry_param.FileReportPath))
         reporter.report_load_result(load_result)
         return

pytest/src/testsolar_pytestx/parser.py

@@ -1,6 +1,8 @@
 import json
 import re
-from typing import Dict, List, Optional
+from typing import Dict, Iterable, List, Optional, Set
+
+from _pytest.mark.structures import Mark
 from pytest import Item
 
 
@@ -17,37 +19,135 @@
 # - tag
 # - owner
 # - extra_attributes
+
+
+def _get_desc(item: Item) -> str:
+    try:
+        doc = item.function.__doc__  # type: ignore
+    except AttributeError:
+        return ""
+    return str(doc).strip() if doc else ""
+
+
+def _iter_markers(item: Item) -> List[Mark]:
+    try:
+        return list(item.iter_markers())
+    except AttributeError:
+        return list(item.own_markers or [])
+
+
+def _is_tag_mark(mark: Mark) -> bool:
+    if mark.name in {"attributes"}:
+        return False
+    if mark.args or mark.kwargs:
+        return False
+    return True
+
+
+def _collect_tags(markers: Iterable[Mark]) -> List[str]:
+    """从 markers 中提取 tags（去重），只收集“无 args 且无 kwargs”的 mark.name。
+
+    示例：
+        pytestmark = [pytest.mark.mod_tag]
+
+        @pytest.mark.class_tag
+        class TestX:
+            @pytest.mark.func_tag
+            def test_ok(self):
+                pass
+
+        => tags 包含：mod_tag / class_tag / func_tag
+
+    反例（不会进 tags）：
+        @pytest.mark.slow(reason="too slow")  # 有 kwargs
+        @pytest.mark.owner("alice")          # 有 args
+        @pytest.mark.attributes({"k": "v"})  # 显式排除
+    """
+    tags: List[str] = []
+    seen: Set[str] = set()
+    for mark in markers:
+        name_str = str(mark.name)
+        if not _is_tag_mark(mark) or name_str in seen:
+            continue
+        seen.add(name_str)
+        tags.append(name_str)
+    return tags
+
+
+def _try_set_owner(attributes: Dict[str, str], mark: Mark) -> None:
+    """解析 owner：仅支持 `@pytest.mark.owner("alice")`。
+
+    示例：
+        @pytest.mark.owner("alice")
+        def test_ok():
+            pass
+
+        => attributes["owner"] == "alice"
+    """
+    if "owner" in attributes:
+        return
+    if mark.name != "owner" or not mark.args:
+        return
+    attributes["owner"] = str(mark.args[0])
+
+
+def _try_set_extra_attributes(attributes: Dict[str, str], mark: Mark) -> None:
+    """解析 extra_attributes：仅支持 `@pytest.mark.extra_attributes({"k": "v"})`。
+
+    示例：
+        @pytest.mark.extra_attributes({"env": "AA", "skip": None})
+        def test_ok():
+            pass
+
+        => attributes["extra_attributes"] == '[{"env": "AA"}]'  # None 会被过滤
+    """
+    if mark.name != "extra_attributes" or not mark.args:
+        return
+    data = mark.args[0]
+    if not isinstance(data, dict):
+        return
+    attr_list = [{k: v} for k, v in data.items() if v is not None]
+    attributes["extra_attributes"] = json.dumps(attr_list)
+
+
+def _try_set_coding_testcase_id(attributes: Dict[str, str], item: Item, mark: Mark) -> None:
+    """解析 coding_testcase_id：面向参数化用例，通过参数 id 映射到外部用例 ID。
+
+    示例：
+        @pytest.mark.coding_testcase_id({"case1": "CID-001"})
+        @pytest.mark.parametrize("x", [pytest.param(1, id="case1")])
+        def test_p(x):
+            pass
+
+        => item.name == "test_p[case1]"
+        => attributes["coding_testcase_id"] == "CID-001"
+    """
+    if mark.name != "coding_testcase_id" or not mark.args:
+        return
+    if "[" not in item.name:
+        return
+    data = mark.args[0]
+    if not isinstance(data, dict):
+        return
+    case_data_name = item.name.split("[", 1)[1][:-1]
+    if case_data_name in data:
+        attributes["coding_testcase_id"] = str(data[case_data_name])
+
+
 def parse_case_attributes(item: Item, comment_fields: Optional[List[str]] = None) -> Dict[str, str]:
     """parse testcase attributes"""
-    desc: str = (str(item.function.__doc__) if item.function.__doc__ else "").strip()  # type: ignore
+    desc = _get_desc(item)
     attributes: Dict[str, str] = {"description": desc}
     if comment_fields:
         attributes.update(scan_comment_fields(desc, comment_fields))
 
-    if not item.own_markers:
-        return attributes
-    tags = []
-    for mark in item.own_markers:
-        if not mark.args and mark.name != "attributes":
-            tags.append(mark.name)
-        elif mark.args and mark.name == "owner":
-            attributes["owner"] = str(mark.args[0])
-        elif mark.name == "extra_attributes":
-            extra_attr = {}
-            attr_list = []
-            for key in mark.args[0]:
-                if mark.args[0][key] is None:
-                    continue
-                extra_attr[key] = mark.args[0][key]
-                attr_list.append(extra_attr)
-            attributes["extra_attributes"] = json.dumps(attr_list)
-        elif mark.name == "coding_testcase_id":
-            case_data_name = item.name.split("[")[1][:-1]
-            for data_name in mark.args[0]:
-                if data_name == case_data_name:
-                    attributes["coding_testcase_id"] = mark.args[0][data_name]
-
-    attributes["tags"] = json.dumps(tags)
+    markers = _iter_markers(item)
+    for mark in markers:
+        _try_set_owner(attributes, mark)
+        _try_set_extra_attributes(attributes, mark)
+        _try_set_coding_testcase_id(attributes, item, mark)
+
+    attributes["tags"] = json.dumps(_collect_tags(markers))
     return attributes
 
 

pytest/testdata/test_mark_layers_case.py

@@ -0,0 +1,23 @@
+import pytest
+
+pytestmark = [
+    pytest.mark.mod_tag,
+    pytest.mark.slow(reason="ignored for tags"),
+    pytest.mark.owner("module_owner"),
+    pytest.mark.extra_attributes({"layer": "module", "none": None}),
+]
+
+
+@pytest.mark.class_tag
+@pytest.mark.owner("class_owner")
+class TestMarkLayers:
+    @pytest.mark.func_tag
+    @pytest.mark.owner("func_owner")
+    def test_layers(self):
+        assert True
+
+
+@pytest.mark.coding_testcase_id({"case1": "CID-001"})
+@pytest.mark.parametrize("x", [pytest.param(1, id="case1")])
+def test_coding_id_layers(x):
+    assert x == 1

pytest/tests/test_collector.py

@@ -1,3 +1,4 @@
+import json
 import tempfile
 import unittest
 from pathlib import Path
@@ -257,8 +258,7 @@ def test_collect_testcases_with_coding_testcase_id(self):
             )
             self.assertEqual(re.Tests[1].Attributes["coding_testcase_id"], "789")
 
-    def test_collect_testcases_file_mode_with_directory(self):
-        """测试文件模式：扫描目录"""
+    def test_collect_testcases_with_mark_layers(self):
         with tempfile.TemporaryDirectory() as tmpdir:
             report_file = Path(tmpdir) / "result.json"
             entry = EntryParam(
@@ -418,3 +418,42 @@ def test_scan_pytest_files_excludes_hidden_and_cache_dirs(self):
             # 验证只包含有效的测试文件
             self.assertEqual(len(test_files), 1)
             self.assertIn("test_valid.py", test_files)
+
+    def test_collect_testcases_file_mode_with_directory(self):
+        """测试文件模式：扫描目录"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            report_file = Path(tmpdir) / "result.json"
+            entry = EntryParam(
+                TaskId="aa",
+                ProjectPath=self.testdata_dir,
+                TestSelectors=[
+                    "test_mark_layers_case.py",
+                ],
+                FileReportPath=str(report_file),
+            )
+
+            collect_testcases(entry)
+
+            re = read_file_load_result(report_file)
+            self.assertEqual(len(re.LoadErrors), 0)
+            self.assertEqual(len(re.Tests), 2)
+
+            tests_by_name = {it.Name: it for it in re.Tests}
+
+            class_case_name = "test_mark_layers_case.py?TestMarkLayers/test_layers"
+            self.assertIn(class_case_name, tests_by_name)
+            class_attrs = tests_by_name[class_case_name].Attributes
+            self.assertEqual(class_attrs["owner"], "func_owner")
+            self.assertEqual(class_attrs["extra_attributes"], '[{"layer": "module"}]')
+            self.assertEqual(
+                set(json.loads(class_attrs["tags"])),
+                {"mod_tag", "class_tag", "func_tag"},
+            )
+
+            coding_case_name = "test_mark_layers_case.py?test_coding_id_layers/[case1]"
+            self.assertIn(coding_case_name, tests_by_name)
+            coding_attrs = tests_by_name[coding_case_name].Attributes
+            self.assertEqual(coding_attrs["owner"], "module_owner")
+            self.assertEqual(coding_attrs["extra_attributes"], '[{"layer": "module"}]')
+            self.assertEqual(coding_attrs["coding_testcase_id"], "CID-001")
+            self.assertEqual(set(json.loads(coding_attrs["tags"])), {"mod_tag"})

pytest/tests/test_parser.py

@@ -18,8 +18,7 @@ def test_handle_str_param():
         "owner": "test_owner",
         "tag": "test_tag",
     }
-    result = handle_str_param(desc)
-    assert result == expected
+    assert handle_str_param(desc) == expected
 
 
 def test_scan_comment_fields():
@@ -54,6 +53,7 @@ def test_parse_case_attributes():
         "description": "This is a test function",
         "owner": "test_owner",
         "coding_testcase_id": '["123", "456"]',
+        "tags": "[]",
     }
     assert parse_case_attributes(item, comment_fields) == expected