From d61a56751ec841fa4aa46669a10e421440324ae5 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 01/14] feat: Implement `deprecated` decorator and apply to `setup_logging`. - Introduce `@deprecated` decorator for marking functions as deprecated. - Apply `@deprecated` to `comtypes.logutil.setup_logging` due to its excessive responsibility and limited use. - Add `test_logutil.py` to verify the `DeprecationWarning` mechanism. --- comtypes/logutil.py | 15 +++++++++++++++ comtypes/test/test_logutil.py | 17 +++++++++++++++++ 2 files changed, 32 insertions(+) create mode 100644 comtypes/test/test_logutil.py diff --git a/comtypes/logutil.py b/comtypes/logutil.py index ea1f881a3..ab932b207 100644 --- a/comtypes/logutil.py +++ b/comtypes/logutil.py @@ -1,5 +1,7 @@ # logutil.py +import functools import logging +import warnings from ctypes import WinDLL from ctypes.wintypes import LPCSTR, LPCWSTR @@ -31,6 +33,19 @@ def emit( logging.NTDebugHandler = NTDebugHandler +def deprecated(reason: str): + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + warnings.warn(reason, category=DeprecationWarning, stacklevel=2) + return func(*args, **kwargs) + + return wrapper + + return decorator + + +@deprecated("Deprecated. See https://github.com/enthought/comtypes/issues/920.") def setup_logging(*pathnames): import configparser diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py new file mode 100644 index 000000000..4c393ebe4 --- /dev/null +++ b/comtypes/test/test_logutil.py @@ -0,0 +1,17 @@ +import unittest as ut + +from comtypes.logutil import deprecated + + +class Test_deprecated(ut.TestCase): + def test_warning_is_raised(self): + reason_text = "This is deprecated." + + @deprecated(reason_text) + def test_func(): + return "success" + + with self.assertWarns(DeprecationWarning) as cm: + result = test_func() + self.assertEqual(result, "success") + self.assertEqual(reason_text, str(cm.warning)) From 4929cf061e663248d63dca954b221ad0ce744c0c Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 02/14] test: Add `OutputDebugStringW` capture test. - Introduce `capture_debug_strings` context manager to capture debug output. - Add `Test_OutputDebugStringW` to verify the functionality of `OutputDebugStringW`. - This test lays the groundwork for fixing `logutil.NTDebugHandler`'s Python 2/3 `str` vs `bytes` incompatibility. --- comtypes/test/test_logutil.py | 111 ++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 4c393ebe4..8af1d989e 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -1,5 +1,14 @@ +import contextlib +import ctypes +import threading import unittest as ut +from ctypes import POINTER, c_void_p +from ctypes import c_size_t as SIZE_T +from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR +from comtypes.logutil import ( + _OutputDebugStringW as OutputDebugStringW, +) from comtypes.logutil import deprecated @@ -15,3 +24,105 @@ def test_func(): result = test_func() self.assertEqual(result, "success") self.assertEqual(reason_text, str(cm.warning)) + + +_kernel32 = ctypes.windll.kernel32 + +_CreateEventW = _kernel32.CreateEventW +_CreateEventW.argtypes = [c_void_p, BOOL, BOOL, LPCWSTR] +_CreateEventW.restype = HANDLE + +_SetEvent = _kernel32.SetEvent +_SetEvent.argtypes = [HANDLE] +_SetEvent.restype = BOOL + +_WaitForSingleObject = _kernel32.WaitForSingleObject +_WaitForSingleObject.argtypes = [HANDLE, DWORD] +_WaitForSingleObject.restype = DWORD + +_CreateFileMappingW = _kernel32.CreateFileMappingW +_CreateFileMappingW.argtypes = [HANDLE, c_void_p, DWORD, DWORD, DWORD, LPCWSTR] +_CreateFileMappingW.restype = HANDLE + +_MapViewOfFile = _kernel32.MapViewOfFile +_MapViewOfFile.argtypes = [HANDLE, DWORD, DWORD, DWORD, SIZE_T] +_MapViewOfFile.restype = c_void_p + +_UnmapViewOfFile = _kernel32.UnmapViewOfFile +_UnmapViewOfFile.argtypes = [c_void_p] +_UnmapViewOfFile.restype = BOOL + +_CloseHandle = _kernel32.CloseHandle +_CloseHandle.argtypes = [HANDLE] +_CloseHandle.restype = BOOL + +_GetCurrentProcessId = _kernel32.GetCurrentProcessId +_GetCurrentProcessId.argtypes = [] +_GetCurrentProcessId.restype = DWORD + + +@contextlib.contextmanager +def create_file_mapping(hfile, security, flprotect, size_high, size_low, name): + handle = _CreateFileMappingW(hfile, security, flprotect, size_high, size_low, name) + try: + yield handle + finally: + _CloseHandle(handle) + + +@contextlib.contextmanager +def map_view_of_file(handle, access, offset_high, offset_low, size): + p_view = _MapViewOfFile(handle, access, offset_high, offset_low, size) + try: + yield p_view + finally: + _UnmapViewOfFile(p_view) + + +@contextlib.contextmanager +def create_event(security, manual, init, name): + handle = _CreateEventW(security, manual, init, name) + try: + yield handle + finally: + _CloseHandle(handle) + + +@contextlib.contextmanager +def capture_debug_strings(ready, *, interval): + captured = [] + finished = threading.Event() + pid = _GetCurrentProcessId() + + def _listener() -> None: + with ( + create_event(None, False, False, "DBWIN_BUFFER_READY") as h_buffer_ready, + create_event(None, False, False, "DBWIN_DATA_READY") as h_data_ready, + create_file_mapping(-1, None, 0x04, 0, 4096, "DBWIN_BUFFER") as h_mapping, + map_view_of_file(h_mapping, 0x04, 0, 0, 4096) as p_view, + ): + ready.set() + while not finished.is_set(): + _SetEvent(h_buffer_ready) + if _WaitForSingleObject(h_data_ready, interval) == 0x00000000: + if ctypes.cast(p_view, POINTER(DWORD)).contents.value == pid: + captured.append(ctypes.string_at(p_view + 4).strip(b"\x00")) + + th = threading.Thread(target=_listener, daemon=True) + th.start() + try: + yield captured + finally: + finished.set() + th.join() + + +class Test_OutputDebugStringW(ut.TestCase): + def test(self): + ready = threading.Event() + with capture_debug_strings(ready, interval=100) as cap: + ready.wait(timeout=5) # Wait for the listener to be ready + OutputDebugStringW("hello world") + OutputDebugStringW("test message") + self.assertEqual(cap[0], b"hello world") + self.assertEqual(cap[1], b"test message") From 840e995cbaf06113183fc7398ab99aa062eef2f6 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 03/14] test: Document Windows API calls in `test_logutil` with MS Learn links. - Add inline comments to `test_logutil.py` with direct links to Microsoft Learn documentation for relevant Windows API functions. --- comtypes/test/test_logutil.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 8af1d989e..853407a50 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -28,34 +28,42 @@ def test_func(): _kernel32 = ctypes.windll.kernel32 +# https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-createeventw _CreateEventW = _kernel32.CreateEventW _CreateEventW.argtypes = [c_void_p, BOOL, BOOL, LPCWSTR] _CreateEventW.restype = HANDLE +# https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-setevent _SetEvent = _kernel32.SetEvent _SetEvent.argtypes = [HANDLE] _SetEvent.restype = BOOL +# https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-waitforsingleobject _WaitForSingleObject = _kernel32.WaitForSingleObject _WaitForSingleObject.argtypes = [HANDLE, DWORD] _WaitForSingleObject.restype = DWORD +# https://learn.microsoft.com/en-us/windows/win32/api/memoryapi/nf-memoryapi-createfilemappingw _CreateFileMappingW = _kernel32.CreateFileMappingW _CreateFileMappingW.argtypes = [HANDLE, c_void_p, DWORD, DWORD, DWORD, LPCWSTR] _CreateFileMappingW.restype = HANDLE +# https://learn.microsoft.com/en-us/windows/win32/api/memoryapi/nf-memoryapi-mapviewoffile _MapViewOfFile = _kernel32.MapViewOfFile _MapViewOfFile.argtypes = [HANDLE, DWORD, DWORD, DWORD, SIZE_T] _MapViewOfFile.restype = c_void_p +# https://learn.microsoft.com/en-us/windows/win32/api/memoryapi/nf-memoryapi-unmapviewoffile _UnmapViewOfFile = _kernel32.UnmapViewOfFile _UnmapViewOfFile.argtypes = [c_void_p] _UnmapViewOfFile.restype = BOOL +# https://learn.microsoft.com/en-us/windows/win32/api/handleapi/nf-handleapi-closehandle _CloseHandle = _kernel32.CloseHandle _CloseHandle.argtypes = [HANDLE] _CloseHandle.restype = BOOL +# https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-getcurrentprocessid _GetCurrentProcessId = _kernel32.GetCurrentProcessId _GetCurrentProcessId.argtypes = [] _GetCurrentProcessId.restype = DWORD From 7fef5df1e795e9b1e65166c3c60479059540ea12 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 04/14] test: Add clarity and documentation to `test_logutil` context managers. - Introduce detailed comments to context managers. - Clarify their roles in handling Windows API objects and interprocess communication for `OutputDebugString` capture. - Enhance understanding of test utility functions and shared resources. --- comtypes/test/test_logutil.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 853407a50..1f84e625b 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -71,6 +71,7 @@ def test_func(): @contextlib.contextmanager def create_file_mapping(hfile, security, flprotect, size_high, size_low, name): + """Context manager to creates a Windows file mapping object.""" handle = _CreateFileMappingW(hfile, security, flprotect, size_high, size_low, name) try: yield handle @@ -80,6 +81,9 @@ def create_file_mapping(hfile, security, flprotect, size_high, size_low, name): @contextlib.contextmanager def map_view_of_file(handle, access, offset_high, offset_low, size): + """Context manager to map a view of a file mapping into the process's + address space. + """ p_view = _MapViewOfFile(handle, access, offset_high, offset_low, size) try: yield p_view @@ -89,6 +93,7 @@ def map_view_of_file(handle, access, offset_high, offset_low, size): @contextlib.contextmanager def create_event(security, manual, init, name): + """Context manager to creates a Windows event object.""" handle = _CreateEventW(security, manual, init, name) try: yield handle @@ -98,6 +103,9 @@ def create_event(security, manual, init, name): @contextlib.contextmanager def capture_debug_strings(ready, *, interval): + """Context manager to capture debug strings emitted via `OutputDebugString`. + Spawns a listener thread to monitor the debug channels. + """ captured = [] finished = threading.Event() pid = _GetCurrentProcessId() From f8d1b3e020053eb3077ee0571079f091e55b8bf7 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 05/14] test: Enhance `OutputDebugStringW` capture test clarity. - Add detailed comments to `capture_debug_strings` in `test_logutil.py`. - Clarify the role of shared events (`DBWIN_BUFFER_READY`, `DBWIN_DATA_READY`) and shared memory (`DBWIN_BUFFER`) in interprocess communication for capturing `OutputDebugStringW` output. - Improve understanding of the test's underlying mechanism. --- comtypes/test/test_logutil.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 1f84e625b..6da2770d5 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -111,17 +111,33 @@ def capture_debug_strings(ready, *, interval): pid = _GetCurrentProcessId() def _listener() -> None: + # Create/open named events and file mapping for interprocess communication. + # These objects are part of the Windows Debugging API contract. with ( + # "DBWIN_BUFFER_READY": An event signaled by the listener to indicate + # it's ready to receive debug output. `OutputDebugString` waits for this. create_event(None, False, False, "DBWIN_BUFFER_READY") as h_buffer_ready, + # "DBWIN_DATA_READY": An event signaled by `OutputDebugString` to + # indicate new data is written to the shared buffer. Listener waits. create_event(None, False, False, "DBWIN_DATA_READY") as h_data_ready, + # "DBWIN_BUFFER": A shared memory region where `OutputDebugString` + # writes the debug string data. create_file_mapping(-1, None, 0x04, 0, 4096, "DBWIN_BUFFER") as h_mapping, + # Map the shared memory region into the listener's address space + # for reading the debug strings. map_view_of_file(h_mapping, 0x04, 0, 0, 4096) as p_view, ): - ready.set() + ready.set() # Signal to the main thread that listener is ready. + # Loop until the main thread signals to finish. while not finished.is_set(): - _SetEvent(h_buffer_ready) + _SetEvent(h_buffer_ready) # Signal readiness to `OutputDebugString`. + # Wait for `OutputDebugString` to signal that data is ready. if _WaitForSingleObject(h_data_ready, interval) == 0x00000000: + # Debug string buffer format: [4 bytes: PID][N bytes: string]. + # Check if the process ID in the buffer matches the current PID. if ctypes.cast(p_view, POINTER(DWORD)).contents.value == pid: + # Extract the null-terminated string, skipping the PID, + # and put it into the queue. captured.append(ctypes.string_at(p_view + 4).strip(b"\x00")) th = threading.Thread(target=_listener, daemon=True) From b89e739af525a5de412cc5080b00b7309ee0ab00 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 06/14] test: Use `ctypes.WinDLL` for explicit DLL loading in `test_logutil`. - Replaces `ctypes.windll.kernel32` with `ctypes.WinDLL("kernel32")`. - This prevents unintended modification of shared `ctypes.windll` objects. - Aligns `test_logutil` with recommended `ctypes` practices. --- comtypes/test/test_logutil.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 6da2770d5..0a7cadb1f 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -2,7 +2,7 @@ import ctypes import threading import unittest as ut -from ctypes import POINTER, c_void_p +from ctypes import POINTER, WinDLL, c_void_p from ctypes import c_size_t as SIZE_T from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR @@ -26,7 +26,7 @@ def test_func(): self.assertEqual(reason_text, str(cm.warning)) -_kernel32 = ctypes.windll.kernel32 +_kernel32 = WinDLL("kernel32") # https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-createeventw _CreateEventW = _kernel32.CreateEventW From 10f0a1dab575e28473f22bc1f8e2823dfbe43e34 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 07/14] test: Add error handling assertions to resource creation functions. - Introduce `assert` statements with test helpers in `test_logutil.py`. - These assertions ensure that resource creation failures are immediately caught and reported with detailed error information. --- comtypes/test/test_logutil.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 0a7cadb1f..c15945212 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -26,7 +26,7 @@ def test_func(): self.assertEqual(reason_text, str(cm.warning)) -_kernel32 = WinDLL("kernel32") +_kernel32 = WinDLL("kernel32", use_last_error=True) # https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-createeventw _CreateEventW = _kernel32.CreateEventW @@ -73,6 +73,7 @@ def test_func(): def create_file_mapping(hfile, security, flprotect, size_high, size_low, name): """Context manager to creates a Windows file mapping object.""" handle = _CreateFileMappingW(hfile, security, flprotect, size_high, size_low, name) + assert handle, ctypes.FormatError(ctypes.get_last_error()) try: yield handle finally: @@ -85,6 +86,7 @@ def map_view_of_file(handle, access, offset_high, offset_low, size): address space. """ p_view = _MapViewOfFile(handle, access, offset_high, offset_low, size) + assert p_view, ctypes.FormatError(ctypes.get_last_error()) try: yield p_view finally: @@ -95,6 +97,7 @@ def map_view_of_file(handle, access, offset_high, offset_low, size): def create_event(security, manual, init, name): """Context manager to creates a Windows event object.""" handle = _CreateEventW(security, manual, init, name) + assert handle, ctypes.FormatError(ctypes.get_last_error()) try: yield handle finally: From 176edb28d767c985937df4ec5d7f201d515efa5e Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:22 +0900 Subject: [PATCH 08/14] refactor: Improve type safety for Windows API calls in `test_logutil.py`. - Update `argtypes` to use `POINTER(SECURITY_ATTRIBUTES)`. - This change enhances type safety and clarity by explicitly defining the expected argument types for these Windows API functions, making the test code more robust and easier to understand. --- comtypes/test/test_logutil.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index c15945212..3d4bdb081 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -6,6 +6,7 @@ from ctypes import c_size_t as SIZE_T from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR +from comtypes.client._events import SECURITY_ATTRIBUTES from comtypes.logutil import ( _OutputDebugStringW as OutputDebugStringW, ) @@ -30,7 +31,7 @@ def test_func(): # https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-createeventw _CreateEventW = _kernel32.CreateEventW -_CreateEventW.argtypes = [c_void_p, BOOL, BOOL, LPCWSTR] +_CreateEventW.argtypes = [POINTER(SECURITY_ATTRIBUTES), BOOL, BOOL, LPCWSTR] _CreateEventW.restype = HANDLE # https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-setevent @@ -45,7 +46,14 @@ def test_func(): # https://learn.microsoft.com/en-us/windows/win32/api/memoryapi/nf-memoryapi-createfilemappingw _CreateFileMappingW = _kernel32.CreateFileMappingW -_CreateFileMappingW.argtypes = [HANDLE, c_void_p, DWORD, DWORD, DWORD, LPCWSTR] +_CreateFileMappingW.argtypes = [ + HANDLE, + POINTER(SECURITY_ATTRIBUTES), + DWORD, + DWORD, + DWORD, + LPCWSTR, +] _CreateFileMappingW.restype = HANDLE # https://learn.microsoft.com/en-us/windows/win32/api/memoryapi/nf-memoryapi-mapviewoffile From dd98fd3fa2c2047828cbcd56db97f6add409f982 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:23 +0900 Subject: [PATCH 09/14] refactor: Add type hints to `capture_debug_strings` in `test_logutil.py`. - Introduce type hints to the `capture_debug_strings` context manager in `test_logutil.py`. - This improves code readability and maintainability by explicitly the expected types of its parameters and return value, facilitating better static analysis and developer understanding. --- comtypes/test/test_logutil.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 3d4bdb081..64412394e 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -2,6 +2,7 @@ import ctypes import threading import unittest as ut +from collections.abc import Iterator, Sequence from ctypes import POINTER, WinDLL, c_void_p from ctypes import c_size_t as SIZE_T from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR @@ -113,7 +114,9 @@ def create_event(security, manual, init, name): @contextlib.contextmanager -def capture_debug_strings(ready, *, interval): +def capture_debug_strings( + ready: threading.Event, *, interval: int +) -> Iterator[Sequence[bytes]]: """Context manager to capture debug strings emitted via `OutputDebugString`. Spawns a listener thread to monitor the debug channels. """ From 01453930e67e62e5b9beda882312db4c27ccb83c Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:23 +0900 Subject: [PATCH 10/14] refactor: Add type hints to `test_logutil.py` helper WinApi context managers. - Introduce type hints to `create_file_mapping`, `map_view_of_file`, and `create_event` in `test_logutil.py`. - This improves code readability and maintainability by making explicit the expected types of arguments and return values for these Windows API wrapper functions, enabling better static analysis. --- comtypes/test/test_logutil.py | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 64412394e..2192564c4 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -6,6 +6,8 @@ from ctypes import POINTER, WinDLL, c_void_p from ctypes import c_size_t as SIZE_T from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR +from typing import TYPE_CHECKING, Optional +from typing import Union as _UnionT from comtypes.client._events import SECURITY_ATTRIBUTES from comtypes.logutil import ( @@ -13,6 +15,9 @@ ) from comtypes.logutil import deprecated +if TYPE_CHECKING: + from ctypes import _CArgObject, _Pointer + class Test_deprecated(ut.TestCase): def test_warning_is_raised(self): @@ -79,7 +84,14 @@ def test_func(): @contextlib.contextmanager -def create_file_mapping(hfile, security, flprotect, size_high, size_low, name): +def create_file_mapping( + hfile: int, + security: _UnionT["_Pointer[SECURITY_ATTRIBUTES]", "_CArgObject", None], + flprotect: int, + size_high: int, + size_low: int, + name: Optional[str], +) -> Iterator[int]: """Context manager to creates a Windows file mapping object.""" handle = _CreateFileMappingW(hfile, security, flprotect, size_high, size_low, name) assert handle, ctypes.FormatError(ctypes.get_last_error()) @@ -90,7 +102,9 @@ def create_file_mapping(hfile, security, flprotect, size_high, size_low, name): @contextlib.contextmanager -def map_view_of_file(handle, access, offset_high, offset_low, size): +def map_view_of_file( + handle: int, access: int, offset_high: int, offset_low: int, size: int +) -> Iterator[int]: """Context manager to map a view of a file mapping into the process's address space. """ @@ -103,7 +117,12 @@ def map_view_of_file(handle, access, offset_high, offset_low, size): @contextlib.contextmanager -def create_event(security, manual, init, name): +def create_event( + security: _UnionT["_Pointer[SECURITY_ATTRIBUTES]", "_CArgObject", None], + manual: bool, + init: bool, + name: Optional[str], +) -> Iterator[int]: """Context manager to creates a Windows event object.""" handle = _CreateEventW(security, manual, init, name) assert handle, ctypes.FormatError(ctypes.get_last_error()) From 7ffc3e3d0d9de2b255bcff8f7777c19122c3a6e1 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:23 +0900 Subject: [PATCH 11/14] refactor: Use named constants for Windows API values in `test_logutil.py` - Replace magic numbers with descriptive named constants for Windows API calls within `test_logutil.py`. - This improves the readability, maintainability, and clarity of the test code, making the intent of API parameters more explicit. --- comtypes/test/test_logutil.py | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 2192564c4..3a19a8b98 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -132,6 +132,13 @@ def create_event( _CloseHandle(handle) +DBWIN_BUFFER_SIZE = 4096 # Longer messages are truncated at the source by the OS +WAIT_OBJECT_0 = 0x00000000 +PAGE_READWRITE = 0x04 +FILE_MAP_READ = 0x04 +INVALID_HANDLE_VALUE = -1 # Backed by the system paging file instead of a file on disk + + @contextlib.contextmanager def capture_debug_strings( ready: threading.Event, *, interval: int @@ -155,17 +162,26 @@ def _listener() -> None: create_event(None, False, False, "DBWIN_DATA_READY") as h_data_ready, # "DBWIN_BUFFER": A shared memory region where `OutputDebugString` # writes the debug string data. - create_file_mapping(-1, None, 0x04, 0, 4096, "DBWIN_BUFFER") as h_mapping, + create_file_mapping( + INVALID_HANDLE_VALUE, + None, + PAGE_READWRITE, + 0, + DBWIN_BUFFER_SIZE, + "DBWIN_BUFFER", + ) as h_mapping, # Map the shared memory region into the listener's address space # for reading the debug strings. - map_view_of_file(h_mapping, 0x04, 0, 0, 4096) as p_view, + map_view_of_file( + h_mapping, FILE_MAP_READ, 0, 0, DBWIN_BUFFER_SIZE + ) as p_view, ): ready.set() # Signal to the main thread that listener is ready. # Loop until the main thread signals to finish. while not finished.is_set(): _SetEvent(h_buffer_ready) # Signal readiness to `OutputDebugString`. # Wait for `OutputDebugString` to signal that data is ready. - if _WaitForSingleObject(h_data_ready, interval) == 0x00000000: + if _WaitForSingleObject(h_data_ready, interval) == WAIT_OBJECT_0: # Debug string buffer format: [4 bytes: PID][N bytes: string]. # Check if the process ID in the buffer matches the current PID. if ctypes.cast(p_view, POINTER(DWORD)).contents.value == pid: From 5d9277922a50ed7a610c0cf0634ac4d7b47435f8 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:23 +0900 Subject: [PATCH 12/14] refactor: Extract debug channel setup into `open_dbwin_debug_channels`. - Introduce `open_dbwin_debug_channels` context manager in `test_logutil.py` to encapsulate the creation and management of Windows debug output channels (events and shared memory). - This refactoring improves the readability and reusability of the debug string capturing mechanism, centralizing resource handling. --- comtypes/test/test_logutil.py | 55 ++++++++++++++++++++--------------- 1 file changed, 32 insertions(+), 23 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 3a19a8b98..e0def1cab 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -139,6 +139,37 @@ def create_event( INVALID_HANDLE_VALUE = -1 # Backed by the system paging file instead of a file on disk +@contextlib.contextmanager +def open_dbwin_debug_channels() -> Iterator[tuple[int, int, int]]: + """Context manager to open the standard Windows debug output channels + (events and shared memory). + Yields handles to `DBWIN_BUFFER_READY`, `DBWIN_DATA_READY`, and a pointer + to `DBWIN_BUFFER`. + """ + with ( + # "DBWIN_BUFFER_READY": An event signaled by the listener to indicate + # it's ready to receive debug output. `OutputDebugString` waits for this. + create_event(None, False, False, "DBWIN_BUFFER_READY") as h_buffer_ready, + # "DBWIN_DATA_READY": An event signaled by `OutputDebugString` to + # indicate new data is written to the shared buffer. Listener waits. + create_event(None, False, False, "DBWIN_DATA_READY") as h_data_ready, + # "DBWIN_BUFFER": A shared memory region where `OutputDebugString` + # writes the debug string data. + create_file_mapping( + INVALID_HANDLE_VALUE, + None, + PAGE_READWRITE, + 0, + DBWIN_BUFFER_SIZE, + "DBWIN_BUFFER", + ) as h_mapping, + # Map the shared memory region into the listener's address space + # for reading the debug strings. + map_view_of_file(h_mapping, FILE_MAP_READ, 0, 0, DBWIN_BUFFER_SIZE) as p_view, + ): + yield (h_buffer_ready, h_data_ready, p_view) + + @contextlib.contextmanager def capture_debug_strings( ready: threading.Event, *, interval: int @@ -153,29 +184,7 @@ def capture_debug_strings( def _listener() -> None: # Create/open named events and file mapping for interprocess communication. # These objects are part of the Windows Debugging API contract. - with ( - # "DBWIN_BUFFER_READY": An event signaled by the listener to indicate - # it's ready to receive debug output. `OutputDebugString` waits for this. - create_event(None, False, False, "DBWIN_BUFFER_READY") as h_buffer_ready, - # "DBWIN_DATA_READY": An event signaled by `OutputDebugString` to - # indicate new data is written to the shared buffer. Listener waits. - create_event(None, False, False, "DBWIN_DATA_READY") as h_data_ready, - # "DBWIN_BUFFER": A shared memory region where `OutputDebugString` - # writes the debug string data. - create_file_mapping( - INVALID_HANDLE_VALUE, - None, - PAGE_READWRITE, - 0, - DBWIN_BUFFER_SIZE, - "DBWIN_BUFFER", - ) as h_mapping, - # Map the shared memory region into the listener's address space - # for reading the debug strings. - map_view_of_file( - h_mapping, FILE_MAP_READ, 0, 0, DBWIN_BUFFER_SIZE - ) as p_view, - ): + with open_dbwin_debug_channels() as (h_buffer_ready, h_data_ready, p_view): ready.set() # Signal to the main thread that listener is ready. # Loop until the main thread signals to finish. while not finished.is_set(): From 98ac8ffcd76ce4b7626340bcd61ec15cc5af9aee Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:23 +0900 Subject: [PATCH 13/14] refactor: Use `queue.Queue` for thread-safe debug string capture. - Modify `capture_debug_strings` context manager in `test_logutil.py` to use `queue.Queue` instead of a list for collecting debug output. - This ensures thread-safe communication between the listener thread and the main test thread, preventing race conditions when capturing `OutputDebugString` messages. - Update `Test_OutputDebugStringW` assertions to correctly retrieve messages from the `Queue`. --- comtypes/test/test_logutil.py | 31 +++++++++++++++++-------------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index e0def1cab..0c5df8025 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -2,10 +2,11 @@ import ctypes import threading import unittest as ut -from collections.abc import Iterator, Sequence +from collections.abc import Iterator from ctypes import POINTER, WinDLL, c_void_p from ctypes import c_size_t as SIZE_T from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR +from queue import Queue from typing import TYPE_CHECKING, Optional from typing import Union as _UnionT @@ -171,23 +172,21 @@ def open_dbwin_debug_channels() -> Iterator[tuple[int, int, int]]: @contextlib.contextmanager -def capture_debug_strings( - ready: threading.Event, *, interval: int -) -> Iterator[Sequence[bytes]]: +def capture_debug_strings(ready: threading.Event, *, interval: int) -> Iterator[Queue]: """Context manager to capture debug strings emitted via `OutputDebugString`. Spawns a listener thread to monitor the debug channels. """ - captured = [] + captured = Queue() finished = threading.Event() - pid = _GetCurrentProcessId() - def _listener() -> None: + def _listener( + q: Queue, rdy: threading.Event, fin: threading.Event, pid: int + ) -> None: # Create/open named events and file mapping for interprocess communication. # These objects are part of the Windows Debugging API contract. with open_dbwin_debug_channels() as (h_buffer_ready, h_data_ready, p_view): - ready.set() # Signal to the main thread that listener is ready. - # Loop until the main thread signals to finish. - while not finished.is_set(): + rdy.set() # Signal to the main thread that listener is ready. + while not fin.is_set(): # Loop until the main thread signals to finish. _SetEvent(h_buffer_ready) # Signal readiness to `OutputDebugString`. # Wait for `OutputDebugString` to signal that data is ready. if _WaitForSingleObject(h_data_ready, interval) == WAIT_OBJECT_0: @@ -196,9 +195,13 @@ def _listener() -> None: if ctypes.cast(p_view, POINTER(DWORD)).contents.value == pid: # Extract the null-terminated string, skipping the PID, # and put it into the queue. - captured.append(ctypes.string_at(p_view + 4).strip(b"\x00")) + q.put(ctypes.string_at(p_view + 4).strip(b"\x00")) - th = threading.Thread(target=_listener, daemon=True) + th = threading.Thread( + target=_listener, + args=(captured, ready, finished, _GetCurrentProcessId()), + daemon=True, + ) th.start() try: yield captured @@ -214,5 +217,5 @@ def test(self): ready.wait(timeout=5) # Wait for the listener to be ready OutputDebugStringW("hello world") OutputDebugStringW("test message") - self.assertEqual(cap[0], b"hello world") - self.assertEqual(cap[1], b"test message") + self.assertEqual(cap.get(), b"hello world") + self.assertEqual(cap.get(), b"test message") From 50494f7a6424e634ae74a69e2722e74a527ee472 Mon Sep 17 00:00:00 2001 From: junkmd Date: Sun, 1 Feb 2026 18:58:23 +0900 Subject: [PATCH 14/14] fix: Fix `logutil.NTDebugHandler` Python 2/3 `str` vs `bytes` incompatibility. - Modify `logutil.NTDebugHandler.emit` to consistently use `_OutputDebugStringW`. - Remove `str` vs `bytes` type branching, simplifying implementation and resolving compatibility issues. - Add `Test_NTDebugHandler` to verify proper functioning of the fixed handler. --- comtypes/logutil.py | 6 +----- comtypes/test/test_logutil.py | 21 ++++++++++++++++++++- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/comtypes/logutil.py b/comtypes/logutil.py index ab932b207..a3db8ef1f 100644 --- a/comtypes/logutil.py +++ b/comtypes/logutil.py @@ -20,14 +20,10 @@ class NTDebugHandler(logging.Handler): def emit( self, record, - writeA=_OutputDebugStringA, writeW=_OutputDebugStringW, ): text = self.format(record) - if isinstance(text, str): - writeA(text + "\n") - else: - writeW(text + "\n") + writeW(text + "\n") logging.NTDebugHandler = NTDebugHandler diff --git a/comtypes/test/test_logutil.py b/comtypes/test/test_logutil.py index 0c5df8025..04f3bd32f 100644 --- a/comtypes/test/test_logutil.py +++ b/comtypes/test/test_logutil.py @@ -1,5 +1,6 @@ import contextlib import ctypes +import logging import threading import unittest as ut from collections.abc import Iterator @@ -11,10 +12,10 @@ from typing import Union as _UnionT from comtypes.client._events import SECURITY_ATTRIBUTES +from comtypes.logutil import NTDebugHandler, deprecated from comtypes.logutil import ( _OutputDebugStringW as OutputDebugStringW, ) -from comtypes.logutil import deprecated if TYPE_CHECKING: from ctypes import _CArgObject, _Pointer @@ -219,3 +220,21 @@ def test(self): OutputDebugStringW("test message") self.assertEqual(cap.get(), b"hello world") self.assertEqual(cap.get(), b"test message") + + +class Test_NTDebugHandler(ut.TestCase): + def test_emit(self): + ready = threading.Event() + handler = NTDebugHandler() + logger = logging.getLogger("test_ntdebug_handler") + # Clear existing handlers to prevent interference from other tests + logger.handlers = [] + logger.addHandler(handler) + logger.setLevel(logging.INFO) + with capture_debug_strings(ready, interval=100) as cap: + ready.wait(timeout=5) # Wait for the listener to be ready + msg = "This is a test message from NTDebugHandler." + logger.info(msg) + logger.removeHandler(handler) + handler.close() + self.assertEqual(cap.get(), msg.encode("utf-8") + b"\n")