From e07b78c01a57c5921614388e287c584d19a3e40c Mon Sep 17 00:00:00 2001 From: "Miss Islington (bot)" <31488909+miss-islington@users.noreply.github.com> Date: Fri, 17 Oct 2025 17:29:15 +0200 Subject: [PATCH] [3.14] gh-116738: test `uuid` module thread safety in free-threading (GH-140068) (#140229) gh-116738: test `uuid` module thread safety in free-threading (GH-140068) (cherry picked from commit 9a87ce8b57f5d698900d84b4f6f5aa47b5f37a89) Co-authored-by: Alper --- Lib/test/test_free_threading/test_uuid.py | 60 +++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100755 Lib/test/test_free_threading/test_uuid.py diff --git a/Lib/test/test_free_threading/test_uuid.py b/Lib/test/test_free_threading/test_uuid.py new file mode 100755 index 00000000000..d794afc552a --- /dev/null +++ b/Lib/test/test_free_threading/test_uuid.py @@ -0,0 +1,60 @@ +import os +import unittest + +from test.support import import_helper, threading_helper +from test.support.threading_helper import run_concurrently +from uuid import SafeUUID + +c_uuid = import_helper.import_module("_uuid") + +NTHREADS = 10 +UUID_PER_THREAD = 1000 + + +@threading_helper.requires_working_threading() +class UUIDTests(unittest.TestCase): + @unittest.skipUnless(os.name == "posix", "POSIX only") + def test_generate_time_safe(self): + uuids = [] + + def worker(): + local_uuids = [] + for _ in range(UUID_PER_THREAD): + uuid, is_safe = c_uuid.generate_time_safe() + self.assertIs(type(uuid), bytes) + self.assertEqual(len(uuid), 16) + # Collect the UUID only if it is safe. If not, we cannot ensure + # UUID uniqueness. According to uuid_generate_time_safe() man + # page, it is theoretically possible for two concurrently + # running processes to generate the same UUID(s) if the return + # value is not 0. + if is_safe == SafeUUID.safe: + local_uuids.append(uuid) + + # Merge all safe uuids + uuids.extend(local_uuids) + + run_concurrently(worker_func=worker, nthreads=NTHREADS) + self.assertEqual(len(uuids), len(set(uuids))) + + @unittest.skipUnless(os.name == "nt", "Windows only") + def test_UuidCreate(self): + uuids = [] + + def worker(): + local_uuids = [] + for _ in range(UUID_PER_THREAD): + uuid = c_uuid.UuidCreate() + self.assertIs(type(uuid), bytes) + self.assertEqual(len(uuid), 16) + local_uuids.append(uuid) + + # Merge all uuids + uuids.extend(local_uuids) + + run_concurrently(worker_func=worker, nthreads=NTHREADS) + self.assertEqual(len(uuids), len(set(uuids))) + + +if __name__ == "__main__": + unittest.main()