Preserve Win32 last-error in winproxy

ctypes_generation/func_parser.py

@@ -24,7 +24,7 @@ def generate_comment_ctypes(self):
         return model.format(self.name, ctypes_param_str)
 
     def generate_prototype_ctypes(self):
-        model = "{0} = WINFUNCTYPE({1})"
+        model = "{0} = WINFUNCTYPE({1}, use_last_error=True)"
         if isinstance(self.return_type, tuple) and self.return_type[0] == "PTR":
             ctypes_param = ["POINTER({0})".format(self.return_type[1])]
         else:

docs/source/winproxy.rst

@@ -12,6 +12,7 @@ Here are the things to know about :mod:`windows.winproxy`
     * All parameters can be passed by ordinal or keyword
     * The call will fail if an argument with default value ``NeededParamater`` have been called without another value.
     * The call will raise a subclass of :class:`WindowsError` if it fails.
+    * Win32 last-error codes are captured by ctypes at the native call boundary before error checks run.
     * Some functions are 'transparent proxy' it means that all parameters are mandatory
 
 Example: ``VirtualAlloc``

tests/test_winproxy.py

@@ -1,9 +1,15 @@
+import ctypes
+import sys
+import threading
+
 import pytest
 
 import windows
 import windows.generated_def as gdef
 
 from .pfwtest import *
+from windows.winproxy.apis.advapi32 import Advapi32Proxy
+from windows.winproxy.error import fail_on_zero
 
 pytestmark = pytest.mark.usefixtures('check_for_gc_garbage')
 
@@ -38,3 +44,56 @@ def test_resolve():
     assert ntdll.name == "ntdll.dll"
     assert ntdll.pe.exports["NtCreateFile"] == windows.winproxy.resolve(windows.winproxy.NtCreateFile)
 
+
+def test_last_error_captured_before_error_check():
+    def check_after_native_error_is_overwritten(func_name, result, func, args):
+        ctypes.windll.kernel32.SetLastError(0)
+        return fail_on_zero(func_name, result, func, args)
+
+    @Advapi32Proxy("GetTokenInformation", error_check=check_after_native_error_is_overwritten)
+    def get_token_information(token, size):
+        return get_token_information.ctypes_function(
+            token.handle, gdef.TokenUser, None, 0, ctypes.byref(size)
+        )
+
+    token = windows.current_process.token
+    with pytest.raises(WindowsError) as exc_info:
+        get_token_information(token, gdef.DWORD())
+    assert exc_info.value.winerror == gdef.ERROR_INSUFFICIENT_BUFFER
+
+
+def test_last_error_preserved_during_gil_contention():
+    token = windows.current_process.token
+    size = gdef.DWORD()
+    stop = threading.Event()
+
+    def contend_for_gil():
+        while not stop.is_set():
+            sum(range(1000))
+
+    worker = threading.Thread(target=contend_for_gil)
+    if sys.version_info[0] == 2:
+        get_interval, set_interval, interval = sys.getcheckinterval, sys.setcheckinterval, 1
+    else:
+        get_interval, set_interval, interval = sys.getswitchinterval, sys.setswitchinterval, 0.00001
+    original_interval = get_interval()
+    set_interval(interval)
+    try:
+        worker.start()
+        for _ in range(256):
+            with pytest.raises(WindowsError) as exc_info:
+                windows.winproxy.GetTokenInformation(
+                    token.handle,
+                    gdef.TokenUser,
+                    None,
+                    0,
+                    ctypes.byref(size),
+                )
+            assert exc_info.value.winerror == gdef.ERROR_INSUFFICIENT_BUFFER
+            assert windows.winproxy.GetLastError() == gdef.ERROR_INSUFFICIENT_BUFFER
+    finally:
+        stop.set()
+        if worker.ident is not None:
+            worker.join()
+        set_interval(original_interval)
+