Add create_unix_connection sock cancel tests

Mirror the TCP cancel/detach and data-leak tests for the
create_unix_connection(sock=) path, covering the fix in
both create_connection and create_unix_connection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: fantix

tests/test_unix.py

@@ -405,6 +405,114 @@ def test_create_unix_connection_6(self):
                     ssl_handshake_timeout=SSL_HANDSHAKE_TIMEOUT))
 
 
+    def test_create_unix_connection_sock_cancel_detaches(self):
+        async def test():
+            srv_path = os.path.join(tempfile.mkdtemp(), 'test.sock')
+            srv = await asyncio.start_unix_server(
+                lambda r, w: None, path=srv_path)
+
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            try:
+                sock.connect(srv_path)
+            except BlockingIOError:
+                pass
+            await asyncio.sleep(0.01)
+
+            task = asyncio.ensure_future(
+                self.loop.create_unix_connection(
+                    asyncio.Protocol, sock=sock))
+            await asyncio.sleep(0)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+
+            self.assertEqual(sock.fileno(), -1)
+
+            srv.close()
+            await srv.wait_closed()
+            if os.path.exists(srv_path):
+                os.unlink(srv_path)
+
+        self.loop.run_until_complete(test())
+
+    def test_create_unix_connection_sock_cancel_fd_leak(self):
+        # Same as test_create_connection_sock_cancel_fd_leak but for
+        # the create_unix_connection(sock=) path.
+
+        async def test():
+            srv_path = os.path.join(tempfile.mkdtemp(), 'test.sock')
+            srv = await asyncio.start_unix_server(
+                lambda r, w: None, path=srv_path)
+
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            sock.setblocking(False)
+            await self.loop.sock_connect(sock, srv_path)
+            stale_fd = sock.fileno()
+
+            task = self.loop.create_task(
+                self.loop.create_unix_connection(
+                    asyncio.Protocol, sock=sock))
+            await asyncio.sleep(0)
+            task.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await task
+
+            # Create victim that reuses the fd.
+            victim_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            victim_sock.setblocking(False)
+            await self.loop.sock_connect(victim_sock, srv_path)
+            victim_tr, _ = await self.loop.create_unix_connection(
+                asyncio.Protocol, sock=victim_sock)
+            victim_fd = victim_tr.get_extra_info('socket').fileno()
+            if victim_fd != stale_fd:
+                victim_tr.close()
+                sock.close()
+                srv.close()
+                await srv.wait_closed()
+                if os.path.exists(srv_path):
+                    os.unlink(srv_path)
+                raise unittest.SkipTest(
+                    f'fd not reused (got {victim_fd}, need {stale_fd})')
+
+            spy_a, spy_b = socket.socketpair()
+            spy_b.setblocking(False)
+
+            sock.close()
+
+            victim_broken = False
+            try:
+                os.fstat(victim_fd)
+            except OSError:
+                victim_broken = True
+
+            if victim_broken:
+                os.dup2(spy_a.fileno(), stale_fd)
+            spy_a.close()
+
+            victim_tr.write(b'LEAKED')
+
+            try:
+                leaked = spy_b.recv(4096)
+            except BlockingIOError:
+                leaked = b''
+
+            if victim_broken:
+                os.close(stale_fd)
+            spy_b.close()
+            victim_tr.close()
+            srv.close()
+            await srv.wait_closed()
+            if os.path.exists(srv_path):
+                os.unlink(srv_path)
+
+            self.assertEqual(leaked, b'',
+                             f"Data leaked to an unrelated socket: "
+                             f"got {leaked!r}")
+
+        self.loop.run_until_complete(test())
+
+
 class Test_UV_Unix(_TestUnix, tb.UVTestCase):
 
     @unittest.skipUnless(hasattr(os, 'fspath'), 'no os.fspath()')