gh-119690: Adds Unicode support for named pipes in _winapi (GH-119717)

(cherry picked from commit 78d697b7d5ec2a6fa046b0e1c34e804f49e750b4)

Co-authored-by: Steve Dower <steve.dower@python.org>

3 files changed, 58 insertions, 1 deletions

diff --git a/Lib/test/audit-tests.py b/Lib/test/audit-tests.py
index de7d0da..b902146 100644
--- a/Lib/test/audit-tests.py
+++ b/Lib/test/audit-tests.py
@@ -556,6 +556,17 @@ def test_sys_monitoring_register_callback():
     sys.monitoring.register_callback(1, 1, None)
 
 
+def test_winapi_createnamedpipe(pipe_name):
+    import _winapi
+
+    def hook(event, args):
+        if event == "_winapi.CreateNamedPipe":
+            print(event, args)
+
+    sys.addaudithook(hook)
+    _winapi.CreateNamedPipe(pipe_name, _winapi.PIPE_ACCESS_DUPLEX, 8, 2, 0, 0, 0, 0)
+
+
 if __name__ == "__main__":
     from test.support import suppress_msvcrt_asserts
 
diff --git a/Lib/test/test_audit.py b/Lib/test/test_audit.py
index e163c7a..321d4f9 100644
--- a/Lib/test/test_audit.py
+++ b/Lib/test/test_audit.py
@@ -291,6 +291,20 @@ class AuditTest(unittest.TestCase):
 
         self.assertEqual(actual, expected)
 
+    def test_winapi_createnamedpipe(self):
+        winapi = import_helper.import_module("_winapi")
+
+        pipe_name = r"\\.\pipe\LOCAL\test_winapi_createnamed_pipe"
+        returncode, events, stderr = self.run_python("test_winapi_createnamedpipe", pipe_name)
+        if returncode:
+            self.fail(stderr)
+
+        if support.verbose:
+            print(*events, sep='\n')
+        actual = [(ev[0], ev[2]) for ev in events]
+        expected = [("_winapi.CreateNamedPipe", f"({pipe_name!r}, 3, 8)")]
+
+        self.assertEqual(actual, expected)
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/Lib/test/test_winapi.py b/Lib/test/test_winapi.py
index 2ac6f36..73b8228 100644
--- a/Lib/test/test_winapi.py
+++ b/Lib/test/test_winapi.py
@@ -7,7 +7,7 @@ import re
 import threading
 import time
 import unittest
-from test.support import import_helper
+from test.support import import_helper, os_helper
 
 _winapi = import_helper.import_module('_winapi', required_on=['win'])
 
@@ -127,3 +127,35 @@ class WinAPITests(unittest.TestCase):
 
         # Should contain "PROGRA~" but we can't predict the number
         self.assertIsNotNone(re.match(r".\:\\PROGRA~\d", actual.upper()), actual)
+
+    def test_namedpipe(self):
+        pipe_name = rf"\\.\pipe\LOCAL\{os_helper.TESTFN}"
+
+        # Pipe does not exist, so this raises
+        with self.assertRaises(FileNotFoundError):
+            _winapi.WaitNamedPipe(pipe_name, 0)
+
+        pipe = _winapi.CreateNamedPipe(
+            pipe_name,
+            _winapi.PIPE_ACCESS_DUPLEX,
+            8, # 8=PIPE_REJECT_REMOTE_CLIENTS
+            2, # two instances available
+            32, 32, 0, 0)
+        self.addCleanup(_winapi.CloseHandle, pipe)
+
+        # Pipe instance is available, so this passes
+        _winapi.WaitNamedPipe(pipe_name, 0)
+
+        with open(pipe_name, 'w+b') as pipe2:
+            # No instances available, so this times out
+            # (WinError 121 does not get mapped to TimeoutError)
+            with self.assertRaises(OSError):
+                _winapi.WaitNamedPipe(pipe_name, 0)
+
+            _winapi.WriteFile(pipe, b'testdata')
+            self.assertEqual(b'testdata', pipe2.read(8))
+
+            self.assertEqual((b'', 0), _winapi.PeekNamedPipe(pipe, 8)[:2])
+            pipe2.write(b'testdata')
+            pipe2.flush()
+            self.assertEqual((b'testdata', 8), _winapi.PeekNamedPipe(pipe, 8)[:2])