From 909871a1ed4af10b98103bfe5b2576d84e27aaf2 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Wed, 2 Apr 2025 17:08:22 +0300 Subject: [PATCH 1/4] Add tests for Bluetooth RFCOMM, HCI and SCO --- Lib/test/test_socket.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index cd3497f0274cd6..854a166a9f8178 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -2670,6 +2670,33 @@ def testBindBrEdrL2capSocket(self): addr = f.getsockname() self.assertEqual(addr, (socket.BDADDR_ANY, psm)) + def testBindRfcommSocket(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s: + channel = 0 + s.bind((socket.BDADDR_ANY, channel)) + addr = s.getsockname() + self.assertEqual(addr, (socket.BDADDR_ANY, channel)) + + @unittest.skipUnless(hasattr(socket, 'BTPROTO_HCI'), 'Bluetooth HCI sockets required for this test') + def testBindHciSocket(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s: + if sys.platform.startswith(('netbsd', 'dragonfly', 'freebsd')): + s.bind(socket.BDADDR_ANY.encode()) + addr = s.getsockname() + self.assertEqual(addr, socket.BDADDR_ANY) + else: + dev = 0 + s.bind((dev,)) + addr = s.getsockname() + self.assertEqual(addr, dev) + + @unittest.skipUnless(hasattr(socket, 'BTPROTO_SCO'), 'Bluetooth SCO sockets required for this test') + def testBindScoSocket(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_SCO) as s: + s.bind(socket.BDADDR_ANY.encode()) + addr = s.getsockname() + self.assertEqual(addr, socket.BDADDR_ANY) + @unittest.skipUnless(HAVE_SOCKET_HYPERV, 'Hyper-V sockets required for this test.') From 94c928704446e3c56d6cba258fe426baec903bd3 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Wed, 2 Apr 2025 21:51:06 +0300 Subject: [PATCH 2/4] Add tests for invalid addresses. --- Lib/test/test_socket.py | 63 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index 854a166a9f8178..60ca1b60e66fa6 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -2670,6 +2670,19 @@ def testBindBrEdrL2capSocket(self): addr = f.getsockname() self.assertEqual(addr, (socket.BDADDR_ANY, psm)) + @unittest.skipUnless(HAVE_SOCKET_BLUETOOTH_L2CAP, 'Bluetooth L2CAP sockets required for this test') + def testBadL2capAddr(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_L2CAP) as f: + with self.assertRaises(OSError): + f.bind((socket.BDADDR_ANY, 0, 0, socket.BDADDR_BREDR, 0)) + with self.assertRaises(OSError): + f.bind((socket.BDADDR_ANY,)) + with self.assertRaises(OSError): + f.bind(socket.BDADDR_ANY) + with self.assertRaises(OSError): + f.bind((socket.BDADDR_ANY.encode(), 0x1001)) + + @unittest.skipIf(sys.platform == "win32", "test does not work on windows") def testBindRfcommSocket(self): with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s: channel = 0 @@ -2677,6 +2690,20 @@ def testBindRfcommSocket(self): addr = s.getsockname() self.assertEqual(addr, (socket.BDADDR_ANY, channel)) + def testBadRfcommAddr(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s: + channel = 0 + with self.assertRaises(OSError): + s.bind((socket.BDADDR_ANY.encode(), channel)) + with self.assertRaises(OSError): + s.bind((socket.BDADDR_ANY,)) + with self.assertRaises(OSError): + s.bind((socket.BDADDR_ANY, channel, 0)) + with self.assertRaises(OSError): + s.bind((socket.BDADDR_ANY + '\0', channel)) + with self.assertRaises(OSError): + s.bind(('invalid', channel)) + @unittest.skipUnless(hasattr(socket, 'BTPROTO_HCI'), 'Bluetooth HCI sockets required for this test') def testBindHciSocket(self): with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s: @@ -2690,6 +2717,32 @@ def testBindHciSocket(self): addr = s.getsockname() self.assertEqual(addr, dev) + @unittest.skipUnless(hasattr(socket, 'BTPROTO_HCI'), 'Bluetooth HCI sockets required for this test') + def testBadHciAddr(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_RAW, socket.BTPROTO_HCI) as s: + if sys.platform.startswith(('netbsd', 'dragonfly', 'freebsd')): + with self.assertRaises(OSError): + s.bind(socket.BDADDR_ANY) + with self.assertRaises(OSError): + s.bind((socket.BDADDR_ANY.encode(),)) + if sys.platform.startswith('freebsd'): + with self.assertRaises(ValueError): + s.bind(socket.BDADDR_ANY.encode() + b'\0') + with self.assertRaises(ValueError): + s.bind(socket.BDADDR_ANY.encode() + b' '*100) + with self.assertRaises(OSError): + s.bind(b'invalid') + else: + dev = 0 + with self.assertRaises(OSError): + s.bind(()) + with self.assertRaises(OSError): + s.bind((dev, 0)) + with self.assertRaises(OSError): + s.bind(dev) + with self.assertRaises(OSError): + s.bind(socket.BDADDR_ANY.encode()) + @unittest.skipUnless(hasattr(socket, 'BTPROTO_SCO'), 'Bluetooth SCO sockets required for this test') def testBindScoSocket(self): with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_SCO) as s: @@ -2697,6 +2750,16 @@ def testBindScoSocket(self): addr = s.getsockname() self.assertEqual(addr, socket.BDADDR_ANY) + @unittest.skipUnless(hasattr(socket, 'BTPROTO_SCO'), 'Bluetooth SCO sockets required for this test') + def testBadScoAddr(self): + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_SEQPACKET, socket.BTPROTO_SCO) as s: + with self.assertRaises(OSError): + s.bind(socket.BDADDR_ANY) + with self.assertRaises(OSError): + s.bind((socket.BDADDR_ANY.encode(),)) + with self.assertRaises(OSError): + s.bind(b'invalid') + @unittest.skipUnless(HAVE_SOCKET_HYPERV, 'Hyper-V sockets required for this test.') From 926644587258a881bad0482d4b94af611d7b758d Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Thu, 3 Apr 2025 14:42:44 +0300 Subject: [PATCH 3/4] Try to run the RFCOMM test on Windows if possible. --- Lib/test/test_socket.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index 60ca1b60e66fa6..3b352d7d919e62 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -1,4 +1,5 @@ import unittest +from unittest import mock from test import support from test.support import ( is_apple, os_helper, refleak_helper, socket_helper, threading_helper @@ -2682,13 +2683,23 @@ def testBadL2capAddr(self): with self.assertRaises(OSError): f.bind((socket.BDADDR_ANY.encode(), 0x1001)) - @unittest.skipIf(sys.platform == "win32", "test does not work on windows") def testBindRfcommSocket(self): with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s: channel = 0 - s.bind((socket.BDADDR_ANY, channel)) + try: + s.bind((socket.BDADDR_ANY, channel)) + except OSError as err: + if sys.platform == 'win32' and err.winerror == 10050: + self.skipTest(str(err)) addr = s.getsockname() - self.assertEqual(addr, (socket.BDADDR_ANY, channel)) + self.assertEqual(addr, (mock.ANY, channel)) + self.assertRegex(addr[0], r'(?i)[0-9a-f]{2}(?::[0-9a-f]{2}){4}') + if sys.platform != 'win32': + self.assertEqual(addr, (socket.BDADDR_ANY, channel)) + with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s: + s.bind(addr) + addr2 = s.getsockname() + self.assertEqual(addr2, addr) def testBadRfcommAddr(self): with socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM) as s: From fc992e3a17aecc707fcfc2b474208b7b83e10094 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Thu, 3 Apr 2025 15:08:35 +0300 Subject: [PATCH 4/4] Re-raise unexpected OSError. --- Lib/test/test_socket.py | 1 + 1 file changed, 1 insertion(+) diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py index 3b352d7d919e62..ab17d19484d358 100644 --- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -2691,6 +2691,7 @@ def testBindRfcommSocket(self): except OSError as err: if sys.platform == 'win32' and err.winerror == 10050: self.skipTest(str(err)) + raise addr = s.getsockname() self.assertEqual(addr, (mock.ANY, channel)) self.assertRegex(addr[0], r'(?i)[0-9a-f]{2}(?::[0-9a-f]{2}){4}')