Add Windows ETW integration tests and drop mock-heavy unit tests

Replace mocked ETW lifecycle tests with subprocess integration checks that
spawn real Python, OpenMP, and BLAS threads while the kernel tracer is
attached. Keep lightweight payload parsing unit tests for all platforms.

Add windows_etw_admin_available() and a short flush delay on tracer stop to
improve reliability on CI runners.

Co-authored-by: Olivier Grisel <olivier.grisel@ensta.org>

Author: cursoragent

threadpoolctl/_thread_tracer/__init__.py

@@ -11,9 +11,22 @@ def windows_tracer_available():
     return sys.platform == "win32"
 
 
+def windows_etw_admin_available():
+    """Return whether the current process can start kernel ETW sessions."""
+    if not windows_tracer_available():
+        return False
+    try:
+        import ctypes
+
+        return bool(ctypes.windll.shell32.IsUserAnAdmin())
+    except (AttributeError, OSError):
+        return False
+
+
 __all__ = [
     "ThreadSpawnStats",
     "ThreadTracerError",
     "WindowsThreadSpawnTracer",
+    "windows_etw_admin_available",
     "windows_tracer_available",
 ]

threadpoolctl/_thread_tracer/_windows_etw.py

@@ -11,6 +11,7 @@
 import ctypes.wintypes as wt
 import sys
 import threading
+import time
 import uuid
 
 from threadpoolctl._thread_tracer._parsing import (
@@ -338,6 +339,7 @@ def stop(self):
             return self._stats.snapshot()
 
         self._stop_event.set()
+        time.sleep(0.2)
         if self._trace_handle.value:
             CloseTrace(self._trace_handle)
         if self._consumer_thread is not None:

threadpoolctl/tests/test_windows_thread_tracer.py

@@ -1,22 +1,16 @@
-import ctypes as ct
 import struct
 import sys
 
 import pytest
 
-from threadpoolctl._thread_tracer import (
-    ThreadSpawnStats,
-    ThreadTracerError,
-    WindowsThreadSpawnTracer,
-    windows_tracer_available,
-)
 from threadpoolctl._thread_tracer._parsing import (
     EVENT_TRACE_TYPE_DCSTART,
     EVENT_TRACE_TYPE_END,
     EVENT_TRACE_TYPE_START,
     classify_kernel_thread_event,
     parse_kernel_thread_payload,
 )
+from threadpoolctl._thread_tracer import windows_tracer_available
 
 
 def _pack_thread_payload(process_id, thread_id):
@@ -52,107 +46,3 @@ def test_parse_kernel_thread_payload_rejects_short_buffers():
 
 def test_windows_tracer_available_matches_platform():
     assert windows_tracer_available() == (sys.platform == "win32")
-
-
-def test_windows_tracer_start_requires_windows(monkeypatch):
-    if sys.platform == "win32":
-        pytest.skip("Platform-specific negative test")
-
-    monkeypatch.setattr(
-        "threadpoolctl._thread_tracer._windows_etw.sys.platform",
-        "linux",
-    )
-    monkeypatch.setattr(
-        "threadpoolctl._thread_tracer._windows_etw._advapi32",
-        None,
-    )
-    monkeypatch.setattr(
-        "threadpoolctl._thread_tracer._windows_etw.StartTraceW",
-        None,
-    )
-
-    tracer = WindowsThreadSpawnTracer(1234)
-    with pytest.raises(ThreadTracerError, match="only supported on Windows"):
-        tracer.start()
-
-
-def test_windows_tracer_lifecycle_with_mocked_etw(monkeypatch):
-    from threadpoolctl._thread_tracer import _windows_etw as etw
-
-    monkeypatch.setattr(etw, "_advapi32", object())
-    monkeypatch.setattr(etw.sys, "platform", "win32")
-
-    spawn_payload = _pack_thread_payload(9001, 12)
-    existing_payload = _pack_thread_payload(9001, 34)
-
-    def _make_record(opcode, payload):
-        record = etw.EVENT_RECORD()
-        record.EventHeader.EventDescriptor.Opcode = opcode
-        record.UserDataLength = len(payload)
-        buffer = ct.create_string_buffer(payload)
-        record._payload_buffer = buffer
-        record.UserData = ct.cast(buffer, ct.c_void_p)
-        return ct.pointer(record)
-
-    events = [
-        _make_record(etw.EVENT_TRACE_TYPE_DCSTART, existing_payload),
-        _make_record(etw.EVENT_TRACE_TYPE_START, spawn_payload),
-        _make_record(etw.EVENT_TRACE_TYPE_START, spawn_payload),
-    ]
-    event_state = {"index": 0, "callback": None, "stop": False}
-
-    def fake_start_trace(session_handle, logger_name, props):
-        session_handle._obj.value = 1
-        return etw.ERROR_SUCCESS
-
-    def fake_open_trace(trace_logfile):
-        event_state["callback"] = trace_logfile._obj.EventRecordCallback
-        return etw.TRACEHANDLE(2)
-
-    def fake_process_trace(trace_handle, count, start_time, end_time):
-        if event_state["stop"]:
-            return 1
-        callback = event_state["callback"]
-        while event_state["index"] < len(events):
-            callback(events[event_state["index"]])
-            event_state["index"] += 1
-        return 1
-
-    def fake_close_trace(trace_handle):
-        event_state["stop"] = True
-        return etw.ERROR_SUCCESS
-
-    def fake_control_trace(session_handle, logger_name, props, control_code):
-        return etw.ERROR_SUCCESS
-
-    monkeypatch.setattr(etw, "StartTraceW", fake_start_trace)
-    monkeypatch.setattr(etw, "OpenTraceW", fake_open_trace)
-    monkeypatch.setattr(etw, "ProcessTrace", fake_process_trace)
-    monkeypatch.setattr(etw, "CloseTrace", fake_close_trace)
-    monkeypatch.setattr(etw, "ControlTraceW", fake_control_trace)
-
-    tracer = WindowsThreadSpawnTracer(9001, include_existing_threads=True)
-    tracer.start()
-    stats = tracer.stop()
-
-    assert stats == ThreadSpawnStats(
-        spawn_count=2,
-        existing_thread_count=1,
-        thread_ids=frozenset({12, 34}),
-    )
-
-
-def test_windows_tracer_reports_existing_kernel_logger_session(monkeypatch):
-    from threadpoolctl._thread_tracer import _windows_etw as etw
-
-    monkeypatch.setattr(etw, "_advapi32", object())
-    monkeypatch.setattr(etw.sys, "platform", "win32")
-
-    def fake_start_trace(session_handle, logger_name, props):
-        return etw.ERROR_ALREADY_EXISTS
-
-    monkeypatch.setattr(etw, "StartTraceW", fake_start_trace)
-
-    tracer = WindowsThreadSpawnTracer(1234)
-    with pytest.raises(ThreadTracerError, match="already in use"):
-        tracer.start()

threadpoolctl/tests/test_windows_thread_tracer_integration.py

@@ -0,0 +1,144 @@
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from threadpoolctl._thread_tracer import (
+    ThreadTracerError,
+    WindowsThreadSpawnTracer,
+    windows_etw_admin_available,
+    windows_tracer_available,
+)
+
+pytestmark = [
+    pytest.mark.skipif(
+        not windows_tracer_available(),
+        reason="Windows ETW integration tests require Windows",
+    ),
+    pytest.mark.skipif(
+        not windows_etw_admin_available(),
+        reason="Windows ETW kernel tracing requires an elevated process",
+    ),
+]
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+TRACER_ATTACH_DELAY_SECONDS = 0.5
+TRACER_FLUSH_DELAY_SECONDS = 0.3
+SUBPROCESS_TIMEOUT_SECONDS = 60
+
+
+def _child_script(body):
+    return textwrap.dedent(
+        """
+        import time
+
+        time.sleep({attach_delay})
+        {body}
+        time.sleep({flush_delay})
+        """
+    ).format(
+        attach_delay=TRACER_ATTACH_DELAY_SECONDS,
+        body=textwrap.indent(textwrap.dedent(body).strip(), "    "),
+        flush_delay=TRACER_FLUSH_DELAY_SECONDS,
+    )
+
+
+def _run_traced_child(body, minimum_spawn_count):
+    env = os.environ.copy()
+    pythonpath = env.get("PYTHONPATH", "")
+    env["PYTHONPATH"] = (
+        str(REPO_ROOT)
+        if not pythonpath
+        else str(REPO_ROOT) + os.pathsep + pythonpath
+    )
+
+    proc = subprocess.Popen(
+        [sys.executable, "-c", _child_script(body)],
+        cwd=str(REPO_ROOT),
+        env=env,
+    )
+    tracer = WindowsThreadSpawnTracer(proc.pid)
+    try:
+        tracer.start()
+    except ThreadTracerError as exc:
+        proc.kill()
+        proc.wait(timeout=SUBPROCESS_TIMEOUT_SECONDS)
+        pytest.fail("failed to start Windows ETW tracer: {0}".format(exc))
+
+    try:
+        return_code = proc.wait(timeout=SUBPROCESS_TIMEOUT_SECONDS)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait(timeout=SUBPROCESS_TIMEOUT_SECONDS)
+        tracer.stop()
+        pytest.fail("timed out waiting for traced child process")
+
+    stats = tracer.stop()
+    assert return_code == 0, "traced child exited with code {0}".format(return_code)
+    assert stats.spawn_count >= minimum_spawn_count, (
+        "expected at least {expected} thread spawn events, got {actual} ({stats})"
+        .format(
+            expected=minimum_spawn_count,
+            actual=stats.spawn_count,
+            stats=stats,
+        )
+    )
+    return stats
+
+
+def test_tracer_counts_python_thread_spawns():
+    body = """
+    import threading
+
+    num_threads = 3
+    started = threading.Barrier(num_threads)
+
+    def work():
+        started.wait(timeout=5)
+
+    threads = [threading.Thread(target=work) for _ in range(num_threads)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=5)
+    """
+    _run_traced_child(body, minimum_spawn_count=3)
+
+
+def test_tracer_counts_openmp_thread_spawns():
+    try:
+        from threadpoolctl.tests._openmp_test_helper import check_openmp_n_threads  # noqa: F401
+    except ImportError:
+        pytest.skip("OpenMP test helper is not built")
+
+    body = """
+    import os
+
+    os.environ["OMP_NUM_THREADS"] = "4"
+    from threadpoolctl.tests._openmp_test_helper import check_openmp_n_threads
+
+    used = check_openmp_n_threads(1000)
+    assert used >= 1
+    """
+    _run_traced_child(body, minimum_spawn_count=4)
+
+
+def test_tracer_counts_blas_thread_spawns():
+    pytest.importorskip("numpy")
+    body = """
+    import os
+
+    os.environ["OMP_NUM_THREADS"] = "1"
+    os.environ["OPENBLAS_NUM_THREADS"] = "4"
+    os.environ["MKL_NUM_THREADS"] = "4"
+
+    import numpy as np
+
+    rng = np.random.RandomState(0)
+    a = rng.rand(2000, 2000)
+    np.dot(a, a)
+    """
+    _run_traced_child(body, minimum_spawn_count=2)