From b7ff6f0fd42ce45a48781d18dc872859bd73d516 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 03:07:46 +0800 Subject: [PATCH] fix(ci): isolate diagnostic writers from test subprocess doubles Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/development/testing-and-quality.md | 4 + scripts/ci/pytest_timing.py | 22 +++-- tests/test_ci_pytest_timing.py | 105 +++++++++++++++++++++--- 3 files changed, 113 insertions(+), 18 deletions(-) diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index 884526476d..bfc284e0d4 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -731,6 +731,8 @@ Each triggered snapshot uses an isolated same-interpreter writer with a 100ms self-exit deadline, plus a 500ms parent communication timeout with kill/reap. This bounds pipe backpressure without changing shared stderr flags or leaving a persistently blocked writer after pytest exits; native process startup can add latency. +The observer retains its own process constructor, so test replacements of public +`subprocess.run` or `subprocess.Popen` cannot capture its diagnostic writer. Snapshots may be dropped on IO failure or backpressure. Default `0` starts no observer timer or writer process; pytest's own results and reports remain authoritative. The reusable Sonar workflow consumes that same run's XML and never reruns @@ -758,6 +760,8 @@ JUnit、coverage 或必过门禁。归属与测量边界见 每次触发栈输出时，使用同解释器的隔离 writer，100ms 自行退出；父进程另设 500ms 通信期限，超时则终止并回收。管道背压不会改变共享 stderr 标志或留下 持续阻塞的 writer；原生进程启动仍可能增加延迟。IO 失败或背压时可丢弃栈输出。 +观察器保留自己的进程构造入口，测试替换公共 `subprocess.run` 或 `subprocess.Popen` +不会接管诊断 writer。 默认 `0` 不启动观察 timer 或 writer 进程；pytest 原生结果和报告仍是权威。 Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。缺少 token 仍成功跳过 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端 diff --git a/scripts/ci/pytest_timing.py b/scripts/ci/pytest_timing.py index c6c75c214b..576b177d6c 100644 --- a/scripts/ci/pytest_timing.py +++ b/scripts/ci/pytest_timing.py @@ -27,6 +27,9 @@ PREFIX = "LOOPX_PYTEST_TIMING " _STACK_FD = pytest.StashKey[int]() +# Tests may replace public subprocess helpers. The observer owns this transport; +# subprocess.run would still resolve a test's replacement Popen at call time. +_STACK_POPEN = subprocess.Popen _STACK_WRITE_PROGRAM = ( "import os,sys,threading;" "deadline=threading.Timer(0.1,lambda:os._exit(0));" @@ -62,13 +65,22 @@ def dump() -> None: # A duplicate shares stderr's flags: never change them or # write a potentially full pipe in this process. The isolated # writer has its own exit deadline, even if pytest is killed; - # run's timeout also kills/reaps it before releasing this FD. - subprocess.run( + # The parent also kills/reaps it before releasing this FD. + with _STACK_POPEN( [sys.executable, "-I", "-S", "-c", _STACK_WRITE_PROGRAM], - input=("\n" + "\n".join(rows) + "\n").encode("utf-8", errors="backslashreplace"), + stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=item.config.stash[_STACK_FD], - timeout=0.5, check=False, - ) + ) as writer: + try: + writer.communicate( + ("\n" + "\n".join(rows) + "\n").encode("utf-8", errors="backslashreplace"), + timeout=0.5, + ) + finally: + # Popen's context manager waits but does not kill on an + # IO error. Stop our child before that unconditional reap. + if writer.poll() is None: + writer.kill() except (OSError, subprocess.TimeoutExpired): # Optional diagnostics cannot replace pytest's result on IO failure. pass diff --git a/tests/test_ci_pytest_timing.py b/tests/test_ci_pytest_timing.py index 6fd5a8ffad..e84ebd655e 100644 --- a/tests/test_ci_pytest_timing.py +++ b/tests/test_ci_pytest_timing.py @@ -165,38 +165,116 @@ def test_stall_snapshot_requires_original_descriptor_only_when_enabled(tmp_path, @pytest.mark.parametrize("workers", [0, 2]) -def test_stall_snapshot_output_failure_preserves_native_test_results(tmp_path, workers): +@pytest.mark.parametrize("communication_failure", [False, pytest.param( + True, marks=pytest.mark.skipif(os.name == "nt", reason="POSIX process-group cleanup of a live fault child"), +)], ids=["spawn", "communication"]) +def test_stall_snapshot_output_failure_preserves_native_test_results(tmp_path, workers, communication_failure): (tmp_path / "conftest.py").write_text( - "import os, subprocess, sys\nfrom pathlib import Path\n" - "original_run = subprocess.run\n" + "import json, os, sys\nfrom pathlib import Path\n" + "from scripts.ci import pytest_timing as observer\n" + "original_popen = observer._STACK_POPEN\nwriters = []\n" "def broken_snapshot(command, *args, **kwargs):\n" " if command[:3] == [sys.executable, '-I', '-S']:\n" " Path('attempt-' + str(os.getpid())).touch()\n" + f" if {communication_failure}:\n" + " process = original_popen([*command[:-1], 'import time; time.sleep(30)'], *args, **kwargs)\n" + " writers.append(process)\n" + " def broken_communicate(*args, **kwargs): raise OSError('diagnostic pipe unavailable')\n" + " process.communicate = broken_communicate\n" + " return process\n" " raise OSError('diagnostic output unavailable')\n" - " return original_run(command, *args, **kwargs)\n" - "def pytest_configure(config): subprocess.run = broken_snapshot\n", + " return original_popen(command, *args, **kwargs)\n" + "def pytest_configure(config): observer._STACK_POPEN = broken_snapshot\n" + "def pytest_sessionfinish(session):\n" + " assert all(process.poll() is not None for process in writers)\n" + " Path('writers-' + str(os.getpid()) + '.json').write_text(json.dumps([p.returncode for p in writers]))\n", encoding="utf-8") (tmp_path / "test_subject.py").write_text( "import time\n" - "def test_pass(): time.sleep(0.03)\n" + "def test_pass(): time.sleep(0.1)\n" "def test_fail():\n" - " time.sleep(0.03)\n" + " time.sleep(0.1)\n" " assert False\n", encoding="utf-8") log = tmp_path / "io-failure.log" with log.open("w") as output: - result = subprocess.run( + process = subprocess.Popen( [sys.executable, "-m", "pytest", "-q", "-n", str(workers), "-p", "scripts.ci.pytest_timing", "--ci-stall-after=0.01"], cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT, - timeout=30, check=False) - assert result.returncode == 1 + start_new_session=os.name != "nt") + try: + assert process.wait(timeout=5) == 1 + finally: + if process.poll() is None: + if os.name == "nt": + process.kill() + else: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=10) assert list(tmp_path.glob("attempt-*")) + writers = [code for path in tmp_path.glob("writers-*.json") for code in json.loads(path.read_text())] + assert len(writers) == (2 if communication_failure else 0) + assert all(code < 0 for code in writers) assert "LOOPX_PYTEST_STACK " not in log.read_text() report = read_log(log) assert report["execution_complete"] and report["pytest_exit_code"] == 1 assert report["selected_cases"] == report["finished_cases"] == 2 +@pytest.mark.parametrize("workers", [0, 2]) +@pytest.mark.parametrize("target", ["run", "Popen"]) +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.skipif(os.name == "nt", reason="POSIX process-group cleanup of the failing baseline") +def test_test_owned_subprocess_doubles_cannot_capture_diagnostic_writers(tmp_path, workers, target, enabled): + (tmp_path / "conftest.py").write_text( + "import subprocess, sys, threading, pytest\nfrom pathlib import Path\n" + "from scripts.ci import pytest_timing as observer\n" + "original_popen = subprocess.Popen\nwriters = []\n" + "def tracked_writer(*args, **kwargs):\n" + " process = original_popen(*args, **kwargs)\n" + " writers.append(process)\n" + " return process\n" + "observer._STACK_POPEN = tracked_writer\n" + f"original = subprocess.{target}\n" + "def test_double(command, *args, **kwargs):\n" + " if command[:3] == [sys.executable, '-I', '-S']:\n" + " Path('shared-subprocess-entered').touch()\n" + " threading.Event().wait(30)\n" + " raise AssertionError('test double must not own diagnostic transport')\n" + " return original(command, *args, **kwargs)\n" + f"def pytest_configure(config): subprocess.{target} = test_double\n" + "@pytest.hookimpl(wrapper=True, tryfirst=True)\n" + "def pytest_runtest_protocol(item):\n" + " before = set(threading.enumerate())\n" + " result = yield\n" + " assert not any(isinstance(t, threading.Timer) and t not in before\n" + " for t in threading.enumerate())\n" + " assert all(p.poll() is not None for p in writers)\n" + " return result\n", encoding="utf-8") + (tmp_path / "test_subject.py").write_text( + "import time\n" + "def test_pass(): time.sleep(0.1)\n" + "def test_fail():\n time.sleep(0.1)\n assert False\n", encoding="utf-8") + log = tmp_path / "test-double.log" + with log.open("w") as output: + process = subprocess.Popen( + [sys.executable, "-m", "pytest", "-q", "-n", str(workers), + "-p", "scripts.ci.pytest_timing", "--ci-stall-after=" + str(0.01 if enabled else 0)], + cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT, + start_new_session=True) + try: + assert process.wait(timeout=5) == 1 + finally: + if process.poll() is None: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=10) + assert not (tmp_path / "shared-subprocess-entered").exists() + assert ("LOOPX_PYTEST_STACK " in log.read_text()) == enabled + report = read_log(log) + assert report["execution_complete"] and report["pytest_exit_code"] == 1 + assert report["selected_cases"] == report["finished_cases"] == 2 + + @pytest.mark.parametrize("workers", [0, 2]) def test_completed_tests_cancel_pending_stall_snapshot(tmp_path, workers): (tmp_path / "test_subject.py").write_text("def test_quick(): pass\n", encoding="utf-8") @@ -228,8 +306,9 @@ def unread_stderr_pipe(): def track_stack_writers(tmp_path, *, child_deadline=True): (tmp_path / "conftest.py").write_text( - "import json, subprocess, sys\nfrom pathlib import Path\n" - "original_popen = subprocess.Popen\nwriters = []\n" + "import json, sys\nfrom pathlib import Path\n" + "from scripts.ci import pytest_timing as observer\n" + "original_popen = observer._STACK_POPEN\nwriters = []\n" "def tracked_popen(command, *args, **kwargs):\n" " diagnostic = command[:3] == [sys.executable, '-I', '-S']\n" f" if diagnostic and {not child_deadline}:\n" @@ -239,7 +318,7 @@ def track_stack_writers(tmp_path, *, child_deadline=True): " writers.append(process)\n" " Path('writer.pid').write_text(str(process.pid))\n" " return process\n" - "subprocess.Popen = tracked_popen\n" + "observer._STACK_POPEN = tracked_popen\n" "def pytest_sessionfinish(session):\n" " assert all(process.poll() is not None for process in writers)\n" " Path('writers.json').write_text(json.dumps([p.returncode for p in writers]))\n",