diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 2d7f182664..dfcb4926bf 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -570,7 +570,7 @@ jobs: needs: [changes, chat-bundle] if: needs.changes.outputs.python_tests == 'true' runs-on: ubuntu-latest - # Share four long-test lanes across current PRs and other workflow runs. + # Share six long-test lanes across current PRs and other workflow runs. # Do not replace another candidate's pending shard or cancel its active head. # Workflow-level concurrency still retires obsolete heads of the same PR. concurrency: @@ -628,9 +628,11 @@ jobs: # Split the whole collection, not a hand-maintained list of directories. # Without timing history least_duration alternates equal-weight tests. # Each runner retains the measured two-worker pool. + # Cooperative per-test Python stacks diagnose stalls without terminating tests. run: >- python -m pytest -v -n 2 -m "not stage2c_e2e" -p scripts.ci.pytest_timing + --ci-stall-after 120 --splits 6 --group ${{ matrix.shard }} --splitting-algorithm least_duration --durations=25 --durations-min=1 diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index 0c8baefac1..e8e808a3b8 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -649,7 +649,7 @@ network latency, provider availability, or a two-hour matrix. 它刻意不包含真实模型调用和 full smoke catalog，因此普通迭代不依赖凭证、网络 时延、模型服务可用性或两小时级测试矩阵。 -The Linux suite uses four hosted runners with two xdist workers each. +The Linux suite uses six hosted runners with two xdist workers each. `pytest-split` partitions the complete collection using `least_duration`; without a timing file, tests have equal weight and alternate between shards. Lint, type checks, and the CLI budget run separately. The required `pytest` @@ -664,13 +664,28 @@ the xdist controller. Read the downloaded log with `node scripts/ci/pytest_timing.ts JOB_LOG`; incomplete evidence remains incomplete and does not replace JUnit, coverage or the required gate. See the [timing observation boundary](ci-impact-selection.md#four-complete-python-shards). +Full Python shards also opt into `--ci-stall-after 120` on the existing pytest +observer: a long-running test attempts to print its executing thread's Python stack without +terminating it or changing its result. This cooperative diagnostic needs the +interpreter to schedule its observer thread; it does not diagnose native code +that holds the GIL indefinitely. The dump is diagnostic evidence, not proof that the last named test +caused a stall. Reproduce with `-p scripts.ci.pytest_timing --ci-stall-after 120`; omit +`--ci-stall-after` or set it to `0` to keep this default-off diagnostic disabled. This grants no authority to +inspect other processes. Large text-boundary fixtures use short parameter IDs +so verbose progress records their cases without printing the payload. +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. +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 pytest or reads cross-run artifacts. Missing Sonar tokens still skip analysis successfully; test jobs receive no Sonar secret. The trigger is the union of the former Python and Sonar paths. Client-only PRs use the exemption above; Sonar-configuration changes remain full, including on forks without a token. -Linux 全套测试分到四台 hosted runner，每台保留两个 xdist worker。`pytest-split` +Linux 全套测试分到六台 hosted runner，每台保留两个 xdist worker。`pytest-split` 按完整 collection 分片；没有历史耗时时，等权测试交替分配。lint、类型检查和 CLI 预算独立执行。必需的 `pytest` 汇总检查会拒绝失败／跳过的分片和缺失的 coverage， 合并后再执行原有 19.6% 门槛；不要求单个分片达到全套覆盖率。coverage 使用相对路径， @@ -680,11 +695,21 @@ CI 同时在 xdist controller 流式记录 pytest 原生 phase 耗时。下载 `node scripts/ci/pytest_timing.ts JOB_LOG` 回读；不完整证据保持不完整，不能替代 JUnit、coverage 或必过门禁。归属与测量边界见 [分片耗时记录](ci-impact-selection.md#four-complete-python-shards)。 +全量 Python 分片在既有 pytest observer 上显式启用 `--ci-stall-after 120`：测试 +超过该时长就尝试输出其执行线程的 Python 栈，不终止测试或改变结果。此协作式诊断 +需要解释器调度观察线程，无法诊断始终占有 GIL 的原生代码。栈用于诊断，不能只凭 +最后一个测试名认定卡顿根因。本地加 `-p scripts.ci.pytest_timing --ci-stall-after 120` +复现，省略此参数或设为 `0` 可保持默认关闭；它不授予检查其他进程的权限。大文本边界 fixture 使用简短参数 ID，详细进度保留用例身份， +不打印整个载荷。 +每次触发栈输出时，使用同解释器的隔离 writer，100ms 自行退出；父进程另设 +500ms 通信期限，超时则终止并回收。管道背压不会改变共享 stderr 标志或留下 +持续阻塞的 writer；原生进程启动仍可能增加延迟。IO 失败或背压时可丢弃栈输出。 +默认 `0` 不启动观察 timer 或 writer 进程；pytest 原生结果和报告仍是权威。 Sonar 只复用同一次 run 的 XML，不重复测试、不跨 run 取产物。缺少 token 仍成功跳过 Sonar，测试 job 不接收 Sonar secret。触发范围取原有两套 workflow 的并集；纯前端 PR 使用前述豁免，Sonar 配置变更仍全量运行，包括没有 token 的 fork。 -Full Python shards share four repository-wide job concurrency groups, one per +Full Python shards share six repository-wide job concurrency groups, one per `matrix.shard`, across PR, main, merge-queue and manual runs carrying this workflow version. Each group admits one running job and uses `queue: max` to retain up to 100 pending jobs instead of replacing another current @@ -699,16 +724,16 @@ unchanged: cancelled, skipped or incomplete dependencies cannot become a green aggregate. This adds one short trusted cleanup job per update without checkout, PR-code execution or new permissions on qualification jobs. Older workflow versions on the base branch retain only closed-event cleanup. Queue overflow, cancellation and missing coverage remain failures of -the required gate; collection, four partitions, two workers, reports and the +the required gate; collection, six partitions, two workers, reports and the coverage floor stay unchanged. Contributors can wait longer for a shard during -overlapping full runs. This limits long-shard occupancy to four runners; it +overlapping full runs. This limits long-shard occupancy to six runners; it does not reserve release capacity or prioritize releases over other jobs or repositories in the organization's shared hosted pool. Older workflow versions do not participate. Roll back by removing this job-level concurrency block, retaining workflow-level cancellation and the required gate. See GitHub's [concurrency queue contract](https://docs.github.com/en/actions/how-tos/write-workflows/choose-when-workflows-run/control-workflow-concurrency). -全量 Python 分片按 `matrix.shard` 共用四个仓库级 job concurrency 组，适用于 +全量 Python 分片按 `matrix.shard` 共用六个仓库级 job concurrency 组，适用于 携带本版 workflow 的 PR、main、merge queue 和手动运行。每组允许一个运行中 job， 用 `queue: max` 保留至多 100 个等待 job，不替换其他有效候选。workflow 级取消 请求清退同 PR 的旧 head；可信 base 分支的 `synchronize` 清理还会取消遗留 @@ -718,14 +743,14 @@ PR、当前 head 和无关联 run 均被保护。必过汇总条件不变，取 不能变绿。每次更新增加一个短清理 job，不 checkout 或执行 PR 代码，不增加普通 验证 job 的权限；base 分支旧版 workflow 仍只清理 closed 事件。队列溢出、取消 和缺失 coverage 仍不能通过必需检查。 -完整 collection、四分片、双 worker、报告和覆盖率门槛保持不变；重叠的全量运行 -会增加贡献者等待分片的时间。此限制只把长分片占用约束到四台 runner，不预留发布 +完整 collection、六分片、双 worker、报告和覆盖率门槛保持不变；重叠的全量运行 +会增加贡献者等待分片的时间。此限制只把长分片占用约束到六台 runner，不预留发布 容量，也不保证发布优先于组织共享 hosted 池中的其他 job 或仓库。旧版 workflow 不参与该限制。回退时移除此 job 的 concurrency 块，保留 workflow 级取消和必需 检查；队列行为遵循上述 GitHub 官方契约。 -Reproduce one shard locally with `uv run --extra test python -m pytest -v -n 2 --splits 4 --group 1 ---splitting-algorithm least_duration --cov=loopx`. Omit the split arguments to +Reproduce one shard locally with `uv run --extra test python -m pytest -v -n 2 --splits 6 --group 1 +--splitting-algorithm least_duration -p scripts.ci.pytest_timing --ci-stall-after 120 --cov=loopx`. Omit the split arguments to run the complete suite locally. 全量本地测试仍省略分片参数即可。 ## Smokes And Canary / Smoke 与 Canary diff --git a/scripts/ci/pytest_timing.py b/scripts/ci/pytest_timing.py index 66e4af3902..c6c75c214b 100644 --- a/scripts/ci/pytest_timing.py +++ b/scripts/ci/pytest_timing.py @@ -1,16 +1,22 @@ """Stream pytest-owned timings from the controller, including interrupted runs. This is a CI observation adapter, not a splitter, timing cache or test gate. -Load explicitly with ``-p scripts.ci.pytest_timing``. No configuration or test -selection changes; node ids and phase/outcome values come from pytest itself. +Load explicitly with ``-p scripts.ci.pytest_timing``. Optional cooperative +stacks do not change test selection or results; timing values come from pytest. """ from __future__ import annotations import json import hashlib import math +import os import re +import subprocess import time +import sys +import threading +from collections.abc import Generator, Sequence +from typing import Any import pytest from _pytest.subtests import SubtestReport @@ -20,10 +26,66 @@ PREFIX = "LOOPX_PYTEST_TIMING " +_STACK_FD = pytest.StashKey[int]() +_STACK_WRITE_PROGRAM = ( + "import os,sys,threading;" + "deadline=threading.Timer(0.1,lambda:os._exit(0));" + "deadline.daemon=True;deadline.start();" + "os.write(2,sys.stdin.buffer.read())" +) + + +def pytest_addoption(parser: pytest.Parser) -> None: + parser.addoption("--ci-stall-after", type=float, default=0, + help="Print the executing test thread's Python stack after this many seconds; never stop the test.") + + +@pytest.hookimpl(wrapper=True, trylast=True) +def pytest_runtest_protocol(item: pytest.Item) -> Generator[None, object, object]: + delay = item.config.getoption("ci_stall_after") + if not delay: + return (yield) + owner_thread = threading.get_ident() + finished = threading.Event() + + def dump() -> None: + # Strong frame references avoid asynchronous native frame traversal. + # Only this test's executing thread is inspected; no locals are read. + frame = sys._current_frames().get(owner_thread) + rows = ["LOOPX_PYTEST_STACK " + item.nodeid.partition("[")[0][:256]] + while frame is not None and len(rows) <= 24: + rows.append(f" {frame.f_code.co_filename}:{frame.f_lineno} in {frame.f_code.co_name}") + frame = frame.f_back + del frame + if not finished.is_set(): + try: + # 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( + [sys.executable, "-I", "-S", "-c", _STACK_WRITE_PROGRAM], + input=("\n" + "\n".join(rows) + "\n").encode("utf-8", errors="backslashreplace"), + stdout=subprocess.DEVNULL, stderr=item.config.stash[_STACK_FD], + timeout=0.5, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + # Optional diagnostics cannot replace pytest's result on IO failure. + pass + + timer = threading.Timer(delay, dump) + timer.daemon = True + timer.start() + try: + return (yield) + finally: + finished.set() + timer.cancel() + timer.join() @pytest.hookimpl(wrapper=True) -def pytest_runtest_makereport(item, call): +def pytest_runtest_makereport(item: pytest.Item, call: Any) -> Generator[None, Any, Any]: report = yield if call.excinfo is not None: try: @@ -34,20 +96,39 @@ def pytest_runtest_makereport(item, call): return report -def pytest_configure(config): +def pytest_configure(config: pytest.Config) -> None: + delay = config.getoption("ci_stall_after") + if not math.isfinite(delay) or delay < 0: + raise pytest.UsageError("--ci-stall-after must be finite and nonnegative") + if delay: + # Capture the original descriptor before per-test fd capture starts, + # so a blocked worker's diagnostic is visible before it returns. + if sys.__stderr__ is None: + raise pytest.UsageError("--ci-stall-after requires an original stderr file descriptor") + try: + config.stash[_STACK_FD] = os.dup(sys.__stderr__.fileno()) + except (AttributeError, OSError, ValueError) as error: + raise pytest.UsageError("--ci-stall-after requires an original stderr file descriptor") from error # xdist sends TestReports to the controller. Worker output would duplicate # reports and can be lost when a worker is killed. if not hasattr(config, "workerinput"): config.pluginmanager.register(TimingObserver(config), "loopx-ci-timing-observer") +def pytest_unconfigure(config: pytest.Config) -> None: + descriptor = config.stash.get(_STACK_FD, None) + if descriptor is not None: + os.close(descriptor) + del config.stash[_STACK_FD] + + class TimingObserver: - def __init__(self, config): + def __init__(self, config: pytest.Config) -> None: self.config = config self.started = time.perf_counter() - self.collection = None + self.collection: tuple[str, ...] | None = None - def emit(self, *, _prefix=PREFIX, **row): + def emit(self, *, _prefix: str = PREFIX, **row: Any) -> None: row["elapsed_seconds"] = round(time.perf_counter() - self.started, 6) line = _prefix + json.dumps(row, ensure_ascii=True, allow_nan=False, separators=(",", ":")) reporter = self.config.pluginmanager.get_plugin("terminalreporter") @@ -61,10 +142,10 @@ def emit(self, *, _prefix=PREFIX, **row): reporter._tw.flush() @pytest.hookimpl(trylast=True) - def pytest_sessionstart(self, session): + def pytest_sessionstart(self, session: pytest.Session) -> None: self.emit(event="start", version=1) - def record_collection(self, ids): + def record_collection(self, ids: Sequence[str]) -> None: if self.collection is None: self.collection = tuple(ids) self.emit(event="collection", selected=len(ids)) @@ -74,14 +155,14 @@ def record_collection(self, ids): self.emit(event="collection_mismatch") @pytest.hookimpl(optionalhook=True) - def pytest_xdist_node_collection_finished(self, node, ids): + def pytest_xdist_node_collection_finished(self, node: Any, ids: Sequence[str]) -> None: self.record_collection(ids) - def pytest_collection_finish(self, session): + def pytest_collection_finish(self, session: pytest.Session) -> None: if not getattr(self.config.option, "numprocesses", None): self.record_collection([item.nodeid for item in session.items]) - def pytest_runtest_logreport(self, report): + def pytest_runtest_logreport(self, report: Any) -> None: test_id = hashlib.sha256(report.nodeid.encode("utf-8", errors="surrogatepass")).hexdigest() display = report.nodeid.partition("[")[0][:256] if report.failed: @@ -108,5 +189,5 @@ def pytest_runtest_logreport(self, report): seconds=duration if math.isfinite(duration) and duration >= 0 else None) @pytest.hookimpl(trylast=True) - def pytest_sessionfinish(self, session, exitstatus): + def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None: self.emit(event="end", exit_code=int(session.exitstatus)) diff --git a/tests/extensions/test_lark_reply_handoff.py b/tests/extensions/test_lark_reply_handoff.py index 46d627159d..59956f5885 100644 --- a/tests/extensions/test_lark_reply_handoff.py +++ b/tests/extensions/test_lark_reply_handoff.py @@ -187,7 +187,8 @@ def sender(original, returned_session, returned_turn, text): controller.close() -@pytest.mark.parametrize("source", ["字" * 32000, "😀" * 24000, "\u0000" * 16000]) +@pytest.mark.parametrize("source", ["字" * 32000, "😀" * 24000, "\u0000" * 16000], + ids=["cjk", "emoji", "escaped-control"]) def test_source_context_encoded_boundary_still_reads_the_real_ingress_record(tmp_path: Path, source: str) -> None: register_ingress(tmp_path, session_id="session", client_turn_id="turn", channel="manager.external.fixture", sender_id="owner", message="Current request", source_id="fixture:current", source_message=source) @@ -196,7 +197,8 @@ def test_source_context_encoded_boundary_still_reads_the_real_ingress_record(tmp assert _read(path)["source_message"] == source -@pytest.mark.parametrize("source", ["x" * 32001, "😀" * 26000, "\u0000" * 17000]) +@pytest.mark.parametrize("source", ["x" * 32001, "😀" * 26000, "\u0000" * 17000], + ids=["ascii", "emoji", "escaped-control"]) def test_oversized_source_is_rejected_before_any_ingress_record(tmp_path: Path, source: str) -> None: with pytest.raises(ValueError, match="source context exceeds.*scoped artifact"): register_ingress(tmp_path, session_id="session", client_turn_id="turn", channel="manager.external.fixture", diff --git a/tests/extensions/test_lark_reply_size.py b/tests/extensions/test_lark_reply_size.py index dfd8dea0bb..a7740c2f64 100644 --- a/tests/extensions/test_lark_reply_size.py +++ b/tests/extensions/test_lark_reply_size.py @@ -8,13 +8,15 @@ ) -@pytest.mark.parametrize("value", ["x" * 149992, "测" * 49997, "😀" * 37498]) +@pytest.mark.parametrize("value", ["x" * 149992, "测" * 49997, "😀" * 37498], + ids=["ascii", "cjk", "emoji"]) def test_provider_byte_budget_accepts_large_text(value): # {"x":""} costs eight bytes; UTF-8 code points have different costs. validate_lark_text_request_size({"x": value}) -@pytest.mark.parametrize("value", ["x" * 149993, "测" * 49998, "😀" * 37499]) +@pytest.mark.parametrize("value", ["x" * 149993, "测" * 49998, "😀" * 37499], + ids=["ascii", "cjk", "emoji"]) def test_provider_byte_budget_rejects_oversize_utf8(value): with pytest.raises(LarkOutboundTextError): validate_lark_text_request_size({"x": value}) diff --git a/tests/test_ci_pytest_timing.py b/tests/test_ci_pytest_timing.py index 0ba61f2120..6fd5a8ffad 100644 --- a/tests/test_ci_pytest_timing.py +++ b/tests/test_ci_pytest_timing.py @@ -34,7 +34,8 @@ def read_log(path, *, success=True): @pytest.mark.parametrize("verbosity", ["-v", "-q"]) @pytest.mark.parametrize("workers", [0, 2]) -def test_real_xdist_timings_match_native_pytest_owner(tmp_path, verbosity, workers): +@pytest.mark.parametrize("stack_timeout", [0, 0.01], ids=["no-stack", "stack"]) +def test_real_xdist_timings_match_native_pytest_owner(tmp_path, verbosity, workers, stack_timeout): (tmp_path / "test_subject.py").write_text( "import time, pytest\n" "def test_pass(): time.sleep(0.03)\n" @@ -57,11 +58,13 @@ def test_real_xdist_timings_match_native_pytest_owner(tmp_path, verbosity, worke with log.open("w") as output: result = subprocess.run( [sys.executable, "-m", "pytest", verbosity, "-n", str(workers), + "--ci-stall-after", str(stack_timeout), "-p", "scripts.ci.pytest_timing", "--store-durations", "--clean-durations", "--durations-path", "native.json", "--junitxml=junit.xml"], cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT, timeout=30, check=False) assert result.returncode == 1 + assert ("LOOPX_PYTEST_STACK " in log.read_text()) == bool(stack_timeout) assert all(line.startswith(PREFIX) for line in log.read_text().splitlines() if PREFIX in line) report = read_log(log) @@ -129,9 +132,198 @@ def test_real_pytest_subtest_cost_is_included_once_in_parent_call(tmp_path): sum(row["duration"] for row in parents), abs=1e-6) +@pytest.mark.parametrize("delay", ["-1", "nan", "inf", "-inf"]) +def test_stall_snapshot_rejects_invalid_delay_before_running_tests(tmp_path, delay): + (tmp_path / "test_subject.py").write_text( + "from pathlib import Path\n" + "def test_unreachable(): Path('ran').touch()\n", encoding="utf-8") + result = subprocess.run( + [sys.executable, "-m", "pytest", "-q", "-p", "scripts.ci.pytest_timing", + "--ci-stall-after=" + delay], cwd=tmp_path, env=environment(), + capture_output=True, text=True, timeout=10, check=False) + assert result.returncode == 4 + assert "must be finite and nonnegative" in result.stderr + assert not (tmp_path / "ran").exists() + + +@pytest.mark.parametrize("stderr", ["None", "io.StringIO()"]) +@pytest.mark.parametrize("enabled", [False, True]) +def test_stall_snapshot_requires_original_descriptor_only_when_enabled(tmp_path, stderr, enabled): + (tmp_path / "test_subject.py").write_text( + "from pathlib import Path\n" + "def test_example(): Path('ran').touch()\n", encoding="utf-8") + result = subprocess.run( + [sys.executable, "-c", "import io, sys, pytest; sys.__stderr__ = " + stderr + + "; raise SystemExit(pytest.main(sys.argv[1:]))", "-q", + "-p", "scripts.ci.pytest_timing", "--ci-stall-after=" + str(int(enabled))], + cwd=tmp_path, env=environment(), capture_output=True, text=True, + timeout=10, check=False) + assert result.returncode == (4 if enabled else 0) + assert (tmp_path / "ran").exists() == (not enabled) + if enabled: + assert "requires an original stderr file descriptor" in result.stderr + + +@pytest.mark.parametrize("workers", [0, 2]) +def test_stall_snapshot_output_failure_preserves_native_test_results(tmp_path, workers): + (tmp_path / "conftest.py").write_text( + "import os, subprocess, sys\nfrom pathlib import Path\n" + "original_run = subprocess.run\n" + "def broken_snapshot(command, *args, **kwargs):\n" + " if command[:3] == [sys.executable, '-I', '-S']:\n" + " Path('attempt-' + str(os.getpid())).touch()\n" + " raise OSError('diagnostic output unavailable')\n" + " return original_run(command, *args, **kwargs)\n" + "def pytest_configure(config): subprocess.run = broken_snapshot\n", + encoding="utf-8") + (tmp_path / "test_subject.py").write_text( + "import time\n" + "def test_pass(): time.sleep(0.03)\n" + "def test_fail():\n" + " time.sleep(0.03)\n" + " assert False\n", encoding="utf-8") + log = tmp_path / "io-failure.log" + with log.open("w") as output: + result = subprocess.run( + [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 + assert list(tmp_path.glob("attempt-*")) + 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]) +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") + result = subprocess.run( + [sys.executable, "-m", "pytest", "-q", "-n", str(workers), + "-p", "scripts.ci.pytest_timing", "--ci-stall-after=60"], + cwd=tmp_path, env=environment(), capture_output=True, text=True, + timeout=10, check=False) + assert result.returncode == 0 + assert "LOOPX_PYTEST_STACK " not in result.stdout + result.stderr + + +@pytest.fixture +def unread_stderr_pipe(): + reader, writer = os.pipe() + try: + os.set_blocking(writer, False) + while True: + try: + os.write(writer, b"x" * 4096) + except BlockingIOError: + break + os.set_blocking(writer, True) + yield writer + finally: + os.close(reader) + os.close(writer) + + +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" + "def tracked_popen(command, *args, **kwargs):\n" + " diagnostic = command[:3] == [sys.executable, '-I', '-S']\n" + f" if diagnostic and {not child_deadline}:\n" + " command = [*command[:-1], 'import os,sys; os.write(2,sys.stdin.buffer.read())']\n" + " process = original_popen(command, *args, **kwargs)\n" + " if diagnostic:\n" + " writers.append(process)\n" + " Path('writer.pid').write_text(str(process.pid))\n" + " return process\n" + "subprocess.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", + encoding="utf-8") + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX pipe/process groups; hosted diagnostic lanes are Linux") +@pytest.mark.parametrize("enabled", [False, True]) +@pytest.mark.parametrize("child_deadline", [True, False], ids=["helper-deadline", "parent-timeout"]) +def test_full_unread_stderr_does_not_block_native_test_results( + tmp_path, unread_stderr_pipe, enabled, child_deadline, +): + track_stack_writers(tmp_path, child_deadline=child_deadline) + (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 / "job.log" + with log.open("w") as output: + process = subprocess.Popen( + [sys.executable, "-m", "pytest", "-q", "-p", "scripts.ci.pytest_timing", + "--ci-stall-after=" + str(0.01 if enabled else 0)], + cwd=tmp_path, env=environment(), stdout=output, stderr=unread_stderr_pipe, + start_new_session=True) + try: + assert process.wait(timeout=4) == 1 + assert os.get_blocking(unread_stderr_pipe) + writers = json.loads((tmp_path / "writers.json").read_text()) + assert len(writers) == (2 if enabled else 0) + assert all(code == 0 if child_deadline else code < 0 for code in writers) + finally: + if process.poll() is None: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=10) + report = read_log(log) + assert report["execution_complete"] and report["pytest_exit_code"] == 1 + assert report["selected_cases"] == report["finished_cases"] == 2 + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX parent kill and process inspection") +def test_blocked_stack_writer_exits_after_only_its_pytest_parent_is_killed(tmp_path, unread_stderr_pipe): + track_stack_writers(tmp_path) + (tmp_path / "test_subject.py").write_text( + "import time\ndef test_wait(): time.sleep(30)\n", encoding="utf-8") + with (tmp_path / "job.log").open("w") as output: + process = subprocess.Popen( + [sys.executable, "-m", "pytest", "-q", "-p", "scripts.ci.pytest_timing", "--ci-stall-after=0.01"], + cwd=tmp_path, env=environment(), stdout=output, stderr=unread_stderr_pipe, + start_new_session=True) + try: + deadline = time.monotonic() + 4 + writer_pid = "" + while not writer_pid.isdecimal(): + assert process.poll() is None + assert time.monotonic() < deadline, "diagnostic writer never started" + if (tmp_path / "writer.pid").exists(): + writer_pid = (tmp_path / "writer.pid").read_text() + time.sleep(0.01) + process.kill() # Leave the writer alive: its own deadline must stop it. + process.wait(timeout=4) + while True: + inspected = subprocess.run( + ["ps", "-p", writer_pid, "-o", "stat="], + capture_output=True, text=True, timeout=2, check=False) + assert inspected.returncode in (0, 1), inspected.stderr + state = inspected.stdout.strip() + if inspected.returncode == 1 or state.startswith("Z"): + break + assert state, "process inspection returned no state" + assert time.monotonic() < deadline, "orphan diagnostic writer remained blocked" + time.sleep(0.02) + assert os.get_blocking(unread_stderr_pipe) + finally: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=10) + + @pytest.mark.skipif(os.name == "nt", reason="POSIX process-group kill; hosted shards are Linux") @pytest.mark.parametrize("verbosity", ["-v", "-q"]) -def test_killed_controller_leaves_partial_readable_native_observations(tmp_path, verbosity): +@pytest.mark.parametrize("stack_timeout", [0, 0.1], ids=["no-stack", "stack"]) +def test_killed_controller_leaves_partial_readable_native_observations(tmp_path, verbosity, stack_timeout): (tmp_path / "test_subject.py").write_text( "import time\n" "def test_first(): time.sleep(0.03)\n" @@ -140,6 +332,7 @@ def test_killed_controller_leaves_partial_readable_native_observations(tmp_path, with log.open("w") as output: process = subprocess.Popen( [sys.executable, "-m", "pytest", verbosity, "-n", "2", + "--ci-stall-after", str(stack_timeout), "-p", "scripts.ci.pytest_timing", "--junitxml=junit.xml"], cwd=tmp_path, env=environment(), stdout=output, stderr=subprocess.STDOUT, start_new_session=True) @@ -148,7 +341,8 @@ def test_killed_controller_leaves_partial_readable_native_observations(tmp_path, while time.monotonic() < deadline: rows = [json.loads(line.removeprefix(PREFIX)) for line in log.read_text().splitlines() if line.startswith(PREFIX)] - if any(row.get("phase") == "teardown" for row in rows): + stack_visible = not stack_timeout or "in test_wait\n" in log.read_text() + if any(row.get("phase") == "teardown" for row in rows) and stack_visible: break assert process.poll() is None, log.read_text() time.sleep(0.02) @@ -164,6 +358,7 @@ def test_killed_controller_leaves_partial_readable_native_observations(tmp_path, assert report["selected_cases"] == 2 assert report["finished_cases"] == 1 assert report["measured_worker_seconds"] >= 0.03 + assert ("in test_wait\n" in log.read_text()) == bool(stack_timeout) assert not (tmp_path / "junit.xml").exists() diff --git a/tests/test_python_ci_workflow.py b/tests/test_python_ci_workflow.py index 31ab4b484d..3fdf8539d7 100644 --- a/tests/test_python_ci_workflow.py +++ b/tests/test_python_ci_workflow.py @@ -325,6 +325,9 @@ def test_six_shards_execute_each_test_once_and_merge_portable_coverage( ) (root / "test_subject.py").write_text( "from ci_subject import first, second\n" + "import pytest, time\n" + "@pytest.fixture(autouse=True)\n" + "def slow_setup(): time.sleep(0.03)\n" "def test_first_one():\n assert first() == 1\n" "def test_first_two():\n assert first() == 1\n" "def test_second_one():\n assert second() == 2\n" @@ -338,6 +341,10 @@ def test_six_shards_execute_each_test_once_and_merge_portable_coverage( encoding="utf-8", ) args = shlex.split(template.replace("${{ matrix.shard }}", str(shard))) + assert args[args.index("--ci-stall-after") + 1] == "120" + # Exercise the same diagnostic on a bounded real fixture. Dumping a + # stack must still permit complete collection, success and coverage. + args[args.index("--ci-stall-after") + 1] = "0.01" args[0] = sys.executable args[args.index("--cov=loopx")] = "--cov=ci_subject" result = subprocess.run( @@ -345,6 +352,7 @@ def test_six_shards_execute_each_test_once_and_merge_portable_coverage( capture_output=True, text=True, timeout=60, check=False, ) assert result.returncode == 0, result.stdout + result.stderr + assert "LOOPX_PYTEST_STACK " in result.stdout + result.stderr cases = ET.parse(root / "results.xml").findall(".//testcase") seen.append({case.attrib["name"] for case in cases}) partial = subprocess.run(