diff --git a/pytest/CHANGELOG.md b/pytest/CHANGELOG.md index 6118249..27123fa 100644 --- a/pytest/CHANGELOG.md +++ b/pytest/CHANGELOG.md @@ -12,4 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Refactor reporter implementation to use FileReporter instead of Reporter ### Fixed -- Improve test case collection and execution with better file handling \ No newline at end of file +- Improve test case collection and execution with better file handling + +### Added +- Support for HTTP request header injection during test execution +- Add test case identification to API requests via X-Testsolar-Testcase header \ No newline at end of file diff --git a/pytest/pyproject.toml b/pytest/pyproject.toml index 838119b..4786bb8 100644 --- a/pytest/pyproject.toml +++ b/pytest/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "testsolar-pytestx" -version = "0.1.58" +version = "0.1.69" description = "Pytest tool for TestSolar" authors = [ {name = "asiazhang2002", email = "asiazhang2002@gmail.com"}, diff --git a/pytest/src/testsolar_pytestx/conftest_generator.py b/pytest/src/testsolar_pytestx/conftest_generator.py new file mode 100644 index 0000000..21e3a5a --- /dev/null +++ b/pytest/src/testsolar_pytestx/conftest_generator.py @@ -0,0 +1,175 @@ +""" +conftest.py 生成器模块 + +用于在测试目录中生成 conftest.py 文件，确保 xdist worker 进程也能正确初始化 header injection。 + +当使用 pytest.main(plugins=[PytestExecutor]) 方式运行测试时， +xdist worker 进程不会继承主进程的 plugins 参数， +因此需要通过 conftest.py 来确保每个 worker 进程都能执行初始化。 +""" + +from pathlib import Path +from typing import Optional + +from loguru import logger + +# conftest.py 的内容模板 +# 注意：这个模板会被写入到测试项目目录中 +CONFTEST_TEMPLATE = '''""" +Auto-generated conftest.py for testsolar header injection support. +This file ensures that HTTP request header injection works correctly +even when using pytest-xdist for parallel test execution. + +DO NOT MODIFY THIS FILE - it will be regenerated on each test run. +""" +import os + +# 仅在启用 API 收集时才初始化 +if os.environ.get("ENABLE_API_COLLECTING") == "1": + try: + from testsolar_pytestx.header_injection import ( + initialize_header_injection, + set_current_test_nodeid, + ) + from testsolar_pytestx.converter import normalize_testcase_name + + # 在模块导入时就初始化，确保在 httpx 被导入之前完成 patch + initialize_header_injection() + + def pytest_runtest_setup(item): + """在每个测试开始前设置 nodeid""" + testcase_name = normalize_testcase_name(item.nodeid) + testcase_class_name = testcase_name.split("?", 1)[-1] + set_current_test_nodeid(testcase_class_name) + + def pytest_runtest_teardown(item): + """在每个测试结束后清除 nodeid""" + set_current_test_nodeid(None) + + except ImportError: + # testsolar_pytestx 未安装，跳过 + pass +''' + +# 用于标识自动生成的 conftest.py 的标记 +CONFTEST_MARKER = "Auto-generated conftest.py for testsolar header injection support." + + +def _is_testsolar_conftest(conftest_path: Path) -> bool: + """检查 conftest.py 是否是由 testsolar 生成的""" + if not conftest_path.exists(): + return False + + try: + content = conftest_path.read_text(encoding="utf-8") + return CONFTEST_MARKER in content + except Exception: + return False + + +def _backup_existing_conftest(conftest_path: Path) -> Optional[Path]: + """备份已存在的 conftest.py（如果不是 testsolar 生成的）""" + if not conftest_path.exists(): + return None + + if _is_testsolar_conftest(conftest_path): + # 是 testsolar 生成的，不需要备份 + return None + + # 创建备份 + backup_path = conftest_path.with_suffix(".py.testsolar_backup") + try: + import shutil + + shutil.copy2(conftest_path, backup_path) + logger.info(f"Backed up existing conftest.py to {backup_path}") + return backup_path + except Exception as e: + logger.error(f"Failed to backup conftest.py: {e}") + return None + + +def _merge_conftest_content(existing_content: str, template_content: str) -> str: + """合并已有的 conftest.py 内容和模板内容""" + # 如果已有内容中已经包含了我们的标记，说明已经合并过了 + if CONFTEST_MARKER in existing_content: + return existing_content + + # 将模板内容添加到已有内容的开头 + merged = ( + template_content + "\n\n# === Original conftest.py content below ===\n\n" + existing_content + ) + return merged + + +def generate_conftest_for_header_injection(project_path: str) -> Optional[str]: + """ + 在项目目录中生成支持 header injection 的 conftest.py + + Args: + project_path: 项目根目录路径 + + Returns: + 生成的 conftest.py 路径，如果生成失败则返回 None + """ + conftest_path = Path(project_path) / "conftest.py" + + try: + if conftest_path.exists(): + if _is_testsolar_conftest(conftest_path): + # 已经是 testsolar 生成的，直接覆盖更新 + logger.debug(f"Updating existing testsolar conftest.py at {conftest_path}") + conftest_path.write_text(CONFTEST_TEMPLATE, encoding="utf-8") + else: + # 已有用户的 conftest.py，需要合并 + logger.info(f"Found existing conftest.py at {conftest_path}, merging content") + existing_content = conftest_path.read_text(encoding="utf-8") + merged_content = _merge_conftest_content(existing_content, CONFTEST_TEMPLATE) + + # 备份原文件 + _backup_existing_conftest(conftest_path) + + # 写入合并后的内容 + conftest_path.write_text(merged_content, encoding="utf-8") + logger.info("Merged testsolar header injection into existing conftest.py") + else: + # 不存在 conftest.py，直接创建 + logger.info(f"Creating conftest.py at {conftest_path}") + conftest_path.write_text(CONFTEST_TEMPLATE, encoding="utf-8") + + return str(conftest_path) + + except Exception as e: + logger.error(f"Failed to generate conftest.py: {e}") + return None + + +def cleanup_generated_conftest(project_path: str) -> None: + """ + 清理生成的 conftest.py + + 如果 conftest.py 是纯粹由 testsolar 生成的，则删除它。 + 如果是合并的，则恢复原始内容。 + + Args: + project_path: 项目根目录路径 + """ + conftest_path = Path(project_path) / "conftest.py" + backup_path = conftest_path.with_suffix(".py.testsolar_backup") + + try: + if backup_path.exists(): + # 存在备份，恢复原始文件 + import shutil + + shutil.move(str(backup_path), str(conftest_path)) + logger.info("Restored original conftest.py from backup") + elif conftest_path.exists() and _is_testsolar_conftest(conftest_path): + # 是纯粹由 testsolar 生成的，检查是否只有模板内容 + content = conftest_path.read_text(encoding="utf-8") + if "Original conftest.py content below" not in content: + # 没有合并用户内容，可以安全删除 + conftest_path.unlink() + logger.info("Removed generated conftest.py") + except Exception as e: + logger.warning(f"Failed to cleanup conftest.py: {e}") diff --git a/pytest/src/testsolar_pytestx/executor.py b/pytest/src/testsolar_pytestx/executor.py index 00bea78..8419223 100644 --- a/pytest/src/testsolar_pytestx/executor.py +++ b/pytest/src/testsolar_pytestx/executor.py @@ -38,12 +38,20 @@ from .parser import parse_case_attributes from .stream import pytest_main_with_output +# from .header_injection import set_current_test_nodeid +from .conftest_generator import generate_conftest_for_header_injection + class RunMode(Enum): SINGLE = "single" BATCH = "batch" +def should_enable_header_injection() -> bool: + """判断是否需要启用请求头注入功能""" + return os.environ.get("ENABLE_API_COLLECTING", "") == "1" + + class PytestExecutor: def __init__( self, @@ -67,6 +75,13 @@ def pytest_runtest_logstart(self, nodeid: str, location: Any) -> None: # 通知ResultHouse用例开始运行 testcase_name = normalize_testcase_name(nodeid, self.data_drive_key) + # 设置当前测试用例的nodeid到上下文 + # 注意：这里使用处理后的 testcase_class_name，与 conftest.py 中的格式可能不同 + # conftest.py 中使用原始 nodeid，这里使用处理后的名称 + # if should_enable_header_injection(): + # testcase_class_name = testcase_name.split("?", 1)[-1] + # set_current_test_nodeid(testcase_class_name) + test_result = TestResult( Test=TestCase(Name=testcase_name), ResultType=ResultType.RUNNING, @@ -210,6 +225,11 @@ def pytest_runtest_logfinish(self, nodeid: str, location: Any) -> None: # 上报完成后测试记录就没有用了，删除以节省内存 self.testdata.pop(testcase_name, None) + + # 清除当前测试用例的nodeid + # if should_enable_header_injection(): + # set_current_test_nodeid(None) + logger.info(f"E {nodeid} runtest_logfinish") def pytest_sessionfinish(self, session: Session, exitstatus: int) -> None: @@ -240,6 +260,13 @@ def run_testcases( if entry.ProjectPath not in sys.path: sys.path.insert(0, entry.ProjectPath) + # 如果启用了 header injection，生成 conftest.py 以支持 xdist + # 这确保了 xdist worker 进程也能正确初始化 header injection + if should_enable_header_injection(): + conftest_path = generate_conftest_for_header_injection(entry.ProjectPath) + if conftest_path: + logger.info(f"Generated conftest.py for header injection at {conftest_path}") + valid_selectors, _ = filter_invalid_selector_path( workspace=entry.ProjectPath, selectors=entry.TestSelectors, diff --git a/pytest/src/testsolar_pytestx/header_injection.py b/pytest/src/testsolar_pytestx/header_injection.py new file mode 100644 index 0000000..1417cf7 --- /dev/null +++ b/pytest/src/testsolar_pytestx/header_injection.py @@ -0,0 +1,202 @@ +""" +HTTP请求头注入模块 + +该模块用于在pytest测试执行过程中，自动向HTTP请求注入测试用例标识信息。 +支持requests和httplib/http.client两种HTTP库。 +""" + +import threading +from typing import Optional, Callable, Any, Dict + +from loguru import logger + + +# ============ 上下文管理 ============ +# 用于存储当前测试用例的nodeid +_test_context = threading.local() + +# 原始请求方法引用 +_original_requests_request: Optional[Callable[..., Any]] = None +_original_httplib_request: Optional[Callable[..., Any]] = None +_original_httpx_client_request: Optional[Callable[..., Any]] = None +_original_httpx_async_client_request: Optional[Callable[..., Any]] = None + + +def get_current_test_nodeid() -> Optional[str]: + """获取当前测试用例的nodeid""" + return getattr(_test_context, "nodeid", None) + + +def set_current_test_nodeid(nodeid: Optional[str]) -> None: + """设置当前测试用例的nodeid""" + _test_context.nodeid = nodeid + + +# ============ Header 注入逻辑 ============ +def _inject_header_to_dict(headers: Optional[Dict[str, Any]], nodeid: str) -> Dict[str, Any]: + """ + 向请求头字典注入X-Testsolar-Testcase + + Args: + headers: 原始请求头字典 + nodeid: 测试用例的nodeid + + Returns: + 注入后的请求头字典 + """ + if headers is None: + headers = {} + elif not isinstance(headers, dict): + # 如果headers不是dict，创建新dict + headers = dict(headers) + else: + # 创建副本避免修改原始dict + headers = headers.copy() + + headers["X-Testsolar-Testcase"] = nodeid + return headers + + +# ============ Monkey Patch 实现 ============ +def _patch_requests() -> None: + """Patch requests库的Session.request方法""" + try: + import requests # type: ignore[import] + + global _original_requests_request + + if _original_requests_request is not None: + # 已经patch过了 + return + + _original_requests_request = requests.Session.request + + def _patched_request(self: Any, method: str, url: str, **kwargs: Any) -> Any: + nodeid = get_current_test_nodeid() + if nodeid: + headers = kwargs.get("headers") + kwargs["headers"] = _inject_header_to_dict(headers, nodeid) + + return _original_requests_request(self, method, url, **kwargs) + + requests.Session.request = _patched_request + logger.info("Successfully patched requests.Session.request") + + except ImportError: + logger.debug("requests library not installed, skip patching") + except Exception as e: + logger.error(f"Failed to patch requests: {e}") + + +def _patch_httplib() -> None: + """Patch httplib/http.client的HTTPConnection.request方法""" + try: + # Python 2/3 兼容 + try: + import http.client as httplib + except ImportError: + import httplib # type: ignore + + global _original_httplib_request + + if _original_httplib_request is not None: + # 已经patch过了 + return + + _original_httplib_request = httplib.HTTPConnection.request + + def _patched_httplib_request( + self: Any, + method: str, + url: str, + body: Any = None, + headers: Any = None, + **kwargs: Any, + ) -> Any: + # 通过判定函数检查是否需要注入 + final_headers: Any = headers + nodeid = get_current_test_nodeid() + if nodeid: + # httplib的headers参数默认是{}而不是None + if final_headers is None: + final_headers = {"X-Testsolar-Testcase": nodeid} + else: + # 创建副本避免修改原始headers + new_headers = dict(final_headers) + new_headers["X-Testsolar-Testcase"] = nodeid + final_headers = new_headers + + return _original_httplib_request(self, method, url, body, final_headers, **kwargs) + + httplib.HTTPConnection.request = _patched_httplib_request # type: ignore[method-assign] + logger.info("Successfully patched httplib.HTTPConnection.request") + + except ImportError: + logger.debug("httplib library not available, skip patching") + except Exception as e: + logger.error(f"Failed to patch httplib: {e}") + + +def _patch_httpx() -> None: + """Patch httpx库的Client.request和AsyncClient.request方法""" + try: + import httpx # type: ignore[import] + + global _original_httpx_client_request + global _original_httpx_async_client_request + + # Patch Sync Client + if _original_httpx_client_request is None: + _original_httpx_client_request = httpx.Client.request + + def _patched_httpx_request(self: Any, method: str, url: str, **kwargs: Any) -> Any: + nodeid = get_current_test_nodeid() + if nodeid: + headers = kwargs.get("headers") + kwargs["headers"] = _inject_header_to_dict(headers, nodeid) + + if _original_httpx_client_request: + return _original_httpx_client_request(self, method, url, **kwargs) + return None + + httpx.Client.request = _patched_httpx_request + logger.info("Successfully patched httpx.Client.request") + + # Patch Async Client + if _original_httpx_async_client_request is None: + _original_httpx_async_client_request = httpx.AsyncClient.request + + async def _patched_httpx_async_request( + self: Any, method: str, url: str, **kwargs: Any + ) -> Any: + nodeid = get_current_test_nodeid() + if nodeid: + headers = kwargs.get("headers") + kwargs["headers"] = _inject_header_to_dict(headers, nodeid) + + if _original_httpx_async_client_request: + return await _original_httpx_async_client_request(self, method, url, **kwargs) + return None + + httpx.AsyncClient.request = _patched_httpx_async_request + logger.info("Successfully patched httpx.AsyncClient.request") + + except ImportError: + logger.debug("httpx library not installed, skip patching") + except Exception as e: + logger.error(f"Failed to patch httpx: {e}") + + +# ============ 公共接口 ============ +def initialize_header_injection() -> None: + """ + 初始化HTTP请求头注入功能 + + 注意：此函数可以被多次调用（在主进程和xdist子进程中）， + 内部会检查是否已经patch过，避免重复patch + """ + + logger.info("Initializing API header injection...") + _patch_requests() + _patch_httplib() + _patch_httpx() diff --git a/pytest/tests/test_conftest_generator.py b/pytest/tests/test_conftest_generator.py new file mode 100644 index 0000000..cc62af4 --- /dev/null +++ b/pytest/tests/test_conftest_generator.py @@ -0,0 +1,187 @@ +"""Tests for conftest_generator module.""" + +import tempfile +from pathlib import Path + +from src.testsolar_pytestx.conftest_generator import ( + CONFTEST_MARKER, + CONFTEST_TEMPLATE, + _is_testsolar_conftest, + _backup_existing_conftest, + _merge_conftest_content, + generate_conftest_for_header_injection, + cleanup_generated_conftest, +) + + +class TestIsTestsolarConftest: + """Test cases for _is_testsolar_conftest function.""" + + def test_returns_false_for_nonexistent_file(self) -> None: + """Test _is_testsolar_conftest returns False for non-existent file.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + assert _is_testsolar_conftest(conftest_path) is False + + def test_returns_true_for_testsolar_generated(self) -> None: + """Test _is_testsolar_conftest returns True for testsolar-generated file.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + conftest_path.write_text(CONFTEST_TEMPLATE, encoding="utf-8") + assert _is_testsolar_conftest(conftest_path) is True + + def test_returns_false_for_user_conftest(self) -> None: + """Test _is_testsolar_conftest returns False for user-created conftest.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + conftest_path.write_text("# user conftest\nimport pytest\n", encoding="utf-8") + assert _is_testsolar_conftest(conftest_path) is False + + +class TestBackupExistingConftest: + """Test cases for _backup_existing_conftest function.""" + + def test_returns_none_for_nonexistent(self) -> None: + """Test _backup_existing_conftest returns None for non-existent file.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + result = _backup_existing_conftest(conftest_path) + assert result is None + + def test_returns_none_for_testsolar_generated(self) -> None: + """Test _backup_existing_conftest returns None for testsolar-generated file.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + conftest_path.write_text(CONFTEST_TEMPLATE, encoding="utf-8") + result = _backup_existing_conftest(conftest_path) + assert result is None + + def test_creates_backup_for_user_conftest(self) -> None: + """Test _backup_existing_conftest creates backup for user conftest.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + user_content = "# user conftest\nimport pytest\n" + conftest_path.write_text(user_content, encoding="utf-8") + + result = _backup_existing_conftest(conftest_path) + + assert result is not None + backup_path = Path(result) + assert backup_path.exists() + assert backup_path.read_text(encoding="utf-8") == user_content + + +class TestMergeConftestContent: + """Test cases for _merge_conftest_content function.""" + + def test_adds_template_to_existing(self) -> None: + """Test _merge_conftest_content adds template to existing content.""" + existing_content = "# user conftest\nimport pytest\n" + merged = _merge_conftest_content(existing_content, CONFTEST_TEMPLATE) + + assert CONFTEST_MARKER in merged + assert "# user conftest" in merged + assert "Original conftest.py content below" in merged + + def test_does_not_duplicate(self) -> None: + """Test _merge_conftest_content does not duplicate if already merged.""" + existing_content = CONFTEST_TEMPLATE + "\n# user code" + merged = _merge_conftest_content(existing_content, CONFTEST_TEMPLATE) + + # Should return the existing content unchanged + assert merged == existing_content + + +class TestGenerateConftestForHeaderInjection: + """Test cases for generate_conftest_for_header_injection function.""" + + def test_creates_new_file(self) -> None: + """Test generate_conftest_for_header_injection creates new conftest.py.""" + with tempfile.TemporaryDirectory() as tmpdir: + result = generate_conftest_for_header_injection(tmpdir) + + assert result is not None + conftest_path = Path(tmpdir) / "conftest.py" + assert conftest_path.exists() + content = conftest_path.read_text(encoding="utf-8") + assert CONFTEST_MARKER in content + + def test_updates_existing_testsolar_conftest(self) -> None: + """Test generate_conftest_for_header_injection updates existing testsolar conftest.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + # Write an old version of testsolar conftest + old_content = f'"""\n{CONFTEST_MARKER}\nOLD VERSION\n"""\n' + conftest_path.write_text(old_content, encoding="utf-8") + + result = generate_conftest_for_header_injection(tmpdir) + + assert result is not None + content = conftest_path.read_text(encoding="utf-8") + # Should be updated to new template + assert content == CONFTEST_TEMPLATE + + def test_merges_with_user_conftest(self) -> None: + """Test generate_conftest_for_header_injection merges with user conftest.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + user_content = "# user conftest\nimport pytest\n\ndef my_fixture():\n pass\n" + conftest_path.write_text(user_content, encoding="utf-8") + + result = generate_conftest_for_header_injection(tmpdir) + + assert result is not None + content = conftest_path.read_text(encoding="utf-8") + # Should contain both marker and user content + assert CONFTEST_MARKER in content + assert "# user conftest" in content + assert "my_fixture" in content + + # Backup should exist + backup_path = conftest_path.with_suffix(".py.testsolar_backup") + assert backup_path.exists() + + +class TestCleanupGeneratedConftest: + """Test cases for cleanup_generated_conftest function.""" + + def test_removes_pure_testsolar_conftest(self) -> None: + """Test cleanup_generated_conftest removes pure testsolar conftest.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + conftest_path.write_text(CONFTEST_TEMPLATE, encoding="utf-8") + + cleanup_generated_conftest(tmpdir) + + assert not conftest_path.exists() + + def test_restores_from_backup(self) -> None: + """Test cleanup_generated_conftest restores from backup.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + backup_path = conftest_path.with_suffix(".py.testsolar_backup") + + user_content = "# user conftest\nimport pytest\n" + backup_path.write_text(user_content, encoding="utf-8") + conftest_path.write_text(CONFTEST_TEMPLATE + user_content, encoding="utf-8") + + cleanup_generated_conftest(tmpdir) + + assert conftest_path.exists() + assert not backup_path.exists() + assert conftest_path.read_text(encoding="utf-8") == user_content + + def test_does_not_remove_merged_conftest(self) -> None: + """Test cleanup_generated_conftest does not remove merged conftest without backup.""" + with tempfile.TemporaryDirectory() as tmpdir: + conftest_path = Path(tmpdir) / "conftest.py" + merged_content = ( + CONFTEST_TEMPLATE + + "\n\n# === Original conftest.py content below ===\n\n# user code" + ) + conftest_path.write_text(merged_content, encoding="utf-8") + + cleanup_generated_conftest(tmpdir) + + # Should still exist because it contains merged content + assert conftest_path.exists() diff --git a/pytest/tests/test_header_injection.py b/pytest/tests/test_header_injection.py new file mode 100644 index 0000000..a3ac7d6 --- /dev/null +++ b/pytest/tests/test_header_injection.py @@ -0,0 +1,240 @@ +"""Tests for header_injection module.""" + +import threading +from typing import Any, Dict, Optional +from unittest import mock + +import pytest + +from src.testsolar_pytestx.header_injection import ( + get_current_test_nodeid, + set_current_test_nodeid, + _inject_header_to_dict, + initialize_header_injection, +) + + +@pytest.fixture(autouse=True) +def reset_nodeid() -> None: + """Reset the test context before and after each test.""" + set_current_test_nodeid(None) + yield # type: ignore[misc] + set_current_test_nodeid(None) + + +class TestContextManagement: + """Test cases for test context management functions.""" + + def test_get_current_test_nodeid_returns_none_initially(self) -> None: + """Test get_current_test_nodeid returns None when not set.""" + result = get_current_test_nodeid() + assert result is None + + def test_set_and_get_current_test_nodeid(self) -> None: + """Test setting and getting the current test nodeid.""" + nodeid = "test_file.py::test_function" + set_current_test_nodeid(nodeid) + result = get_current_test_nodeid() + assert result == nodeid + + def test_set_current_test_nodeid_to_none(self) -> None: + """Test setting nodeid to None clears it.""" + set_current_test_nodeid("test_file.py::test_function") + set_current_test_nodeid(None) + result = get_current_test_nodeid() + assert result is None + + def test_nodeid_is_thread_local(self) -> None: + """Test that nodeid is thread-local.""" + main_nodeid = "main_thread::test" + set_current_test_nodeid(main_nodeid) + + child_nodeid: Optional[str] = None + child_set_result: Optional[str] = None + + def child_thread_func() -> None: + nonlocal child_nodeid, child_set_result + # Should be None in child thread initially + child_nodeid = get_current_test_nodeid() + # Set a different nodeid in child thread + set_current_test_nodeid("child_thread::test") + child_set_result = get_current_test_nodeid() + + thread = threading.Thread(target=child_thread_func) + thread.start() + thread.join() + + # Child thread should have seen None initially + assert child_nodeid is None + # Child thread should have its own nodeid + assert child_set_result == "child_thread::test" + # Main thread nodeid should be unchanged + assert get_current_test_nodeid() == main_nodeid + + +class TestInjectHeaderToDict: + """Test cases for _inject_header_to_dict function.""" + + def test_inject_header_to_none_headers(self) -> None: + """Test injecting header when headers is None.""" + result = _inject_header_to_dict(None, "test::nodeid") + assert result == {"X-Testsolar-Testcase": "test::nodeid"} + + def test_inject_header_to_empty_dict(self) -> None: + """Test injecting header to empty dict.""" + result = _inject_header_to_dict({}, "test::nodeid") + assert result == {"X-Testsolar-Testcase": "test::nodeid"} + + def test_inject_header_to_existing_dict(self) -> None: + """Test injecting header to dict with existing headers.""" + original: Dict[str, Any] = {"Content-Type": "application/json"} + result = _inject_header_to_dict(original, "test::nodeid") + + assert result == { + "Content-Type": "application/json", + "X-Testsolar-Testcase": "test::nodeid", + } + # Original should not be modified + assert "X-Testsolar-Testcase" not in original + + def test_inject_header_does_not_modify_original(self) -> None: + """Test that original headers dict is not modified.""" + original: Dict[str, str] = {"Authorization": "Bearer token"} + _ = _inject_header_to_dict(original, "test::nodeid") + + assert original == {"Authorization": "Bearer token"} + + +class TestInitializeHeaderInjection: + """Test cases for initialize_header_injection function.""" + + def test_can_be_called_multiple_times(self) -> None: + """Test that initialize_header_injection can be called multiple times safely.""" + # Should not raise any exceptions + initialize_header_injection() + initialize_header_injection() + initialize_header_injection() + + def test_patches_requests_if_available(self) -> None: + """Test that requests.Session.request is patched if available.""" + pytest.importorskip("requests") + + initialize_header_injection() + + # We verify by checking that the module-level original reference is set + from src.testsolar_pytestx.header_injection import _original_requests_request + + assert _original_requests_request is not None + + def test_patches_httpx_if_available(self) -> None: + """Test that httpx.Client.request is patched if available.""" + httpx = pytest.importorskip("httpx") + + initialize_header_injection() + + # The Client.request should be patched + assert callable(httpx.Client.request) + + +class TestRequestsIntegration: + """Integration tests for requests library patching.""" + + @pytest.fixture(autouse=True) + def setup_injection(self) -> None: + """Set up header injection.""" + initialize_header_injection() + + def test_requests_header_injection(self) -> None: + """Test that requests injects header when nodeid is set.""" + requests = pytest.importorskip("requests") + + set_current_test_nodeid("test_file.py::test_case") + + # Mock the actual HTTP call to capture the request + with mock.patch("requests.adapters.HTTPAdapter.send") as mock_send: + mock_response = mock.MagicMock() + mock_response.status_code = 200 + mock_send.return_value = mock_response + + session = requests.Session() + try: + session.get("http://example.com") + except Exception: + pass + + # Check that the header was injected + if mock_send.called: + request = mock_send.call_args[0][0] + assert "X-Testsolar-Testcase" in request.headers + assert request.headers["X-Testsolar-Testcase"] == "test_file.py::test_case" + + def test_requests_no_injection_when_nodeid_not_set(self) -> None: + """Test that requests does not inject header when nodeid is not set.""" + requests = pytest.importorskip("requests") + + set_current_test_nodeid(None) + + with mock.patch("requests.adapters.HTTPAdapter.send") as mock_send: + mock_response = mock.MagicMock() + mock_response.status_code = 200 + mock_send.return_value = mock_response + + session = requests.Session() + try: + session.get("http://example.com") + except Exception: + pass + + if mock_send.called: + request = mock_send.call_args[0][0] + assert "X-Testsolar-Testcase" not in request.headers + + +class TestHttpxIntegration: + """Integration tests for httpx library patching.""" + + @pytest.fixture(autouse=True) + def setup_injection(self) -> None: + """Set up header injection.""" + initialize_header_injection() + + def test_httpx_header_injection(self) -> None: + """Test that httpx injects header when nodeid is set.""" + httpx = pytest.importorskip("httpx") + + set_current_test_nodeid("test_file.py::test_httpx_case") + + # Mock the transport to capture the request + with mock.patch.object(httpx.HTTPTransport, "handle_request") as mock_handle: + mock_response = httpx.Response(200) + mock_handle.return_value = mock_response + + client = httpx.Client() + try: + client.get("http://example.com") + except Exception: + pass + finally: + client.close() + + if mock_handle.called: + request = mock_handle.call_args[0][0] + assert b"x-testsolar-testcase" in request.headers.raw + + def test_httpx_no_injection_when_nodeid_not_set(self) -> None: + """Test that httpx does not inject header when nodeid is not set.""" + httpx = pytest.importorskip("httpx") + + set_current_test_nodeid(None) + + with mock.patch.object(httpx.HTTPTransport, "handle_request") as mock_handle: + mock_response = httpx.Response(200) + mock_handle.return_value = mock_response + + client = httpx.Client() + try: + client.get("http://example.com") + except Exception: + pass + finally: + client.close() diff --git a/pytest/tests/test_run.py b/pytest/tests/test_run.py index 2d0f685..9076a7d 100644 --- a/pytest/tests/test_run.py +++ b/pytest/tests/test_run.py @@ -1,4 +1,5 @@ import json +import os import tempfile from pathlib import Path from unittest import TestCase @@ -15,6 +16,10 @@ class TestExecuteEntry(TestCase): testdata_dir = Path(__file__).parent.parent.absolute().joinpath("testdata") + def test_header_injection(self): + os.environ["ENABLE_API_COLLECTING"] = "1" + self.test_run_testcases_from_args() + def test_run_testcases_from_args(self): with tempfile.TemporaryDirectory() as tmpdir: report_dir = Path(tmpdir) diff --git a/pytest/testtool.yaml b/pytest/testtool.yaml index 79a4c81..73469a0 100644 --- a/pytest/testtool.yaml +++ b/pytest/testtool.yaml @@ -2,7 +2,7 @@ schemaVersion: 1.0 name: pytest nameZh: pytest自动化测试 lang: python -version: '0.1.58' +version: '0.1.69' langType: INTERPRETED description: |- pytest是一个成熟的全功能Python测试工具，可以帮助您编写更好的程序。此测试工具允许您在TestSolar上运行pytest。 diff --git a/pytest/uv.lock b/pytest/uv.lock index a5ab4eb..c274e10 100644 --- a/pytest/uv.lock +++ b/pytest/uv.lock @@ -463,7 +463,7 @@ wheels = [ [[package]] name = "testsolar-pytestx" -version = "0.1.58" +version = "0.1.69" source = { virtual = "." } dependencies = [ { name = "coverage" },