3 # © Copyright 2021-2022, Scott Gasch
5 """Make sure id_generator is thread safe."""
9 from pyutils import id_generator
10 from pyutils import unittest_utils as uu
11 from pyutils.parallelize import parallelize as par
12 from pyutils.parallelize import smart_future, thread_utils
15 class TestIdGenerator(unittest.TestCase):
16 @par.parallelize(method=par.Method.THREAD)
17 def get_some_ids(self):
18 name = thread_utils.current_thread_id()
19 print(f"Hello from {name}")
21 for _ in range(10000):
22 results.append(id_generator.get("TestSequence"))
25 def test_is_safe(self):
28 results.append(self.get_some_ids())
30 smart_future.wait_all(results)
32 for result in results:
33 for identifier in result:
34 if identifier in already_seen:
35 self.fail(f"Saw the id {identifier} more than once?!")
37 already_seen.add(identifier)
40 if __name__ == "__main__":