tests/id_generator_test.py

   1 #!/usr/bin/env python3
   2
   3 # © Copyright 2021-2022, Scott Gasch
   4
   5 """Make sure id_generator is thread safe."""
   6
   7 import unittest
   8
   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
  13
  14
  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}")
  20         results = []
  21         for _ in range(10000):
  22             results.append(id_generator.get("TestSequence"))
  23         return results
  24
  25     def test_is_safe(self):
  26         results = []
  27         for i in range(10):
  28             results.append(self.get_some_ids())
  29
  30         smart_future.wait_all(results)
  31         already_seen = set()
  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?!")
  36                 else:
  37                     already_seen.add(identifier)
  38
  39
  40 if __name__ == "__main__":
  41     unittest.main()