3 # © Copyright 2021-2022, Scott Gasch
5 """shared_dict unittest."""
10 from pyutils import unittest_utils
11 from pyutils.collectionz.shared_dict import SharedDict
12 from pyutils.parallelize import parallelize as p
13 from pyutils.parallelize import smart_future
16 class SharedDictTest(unittest.TestCase):
17 @p.parallelize(method=p.Method.PROCESS)
18 def doit(self, n: int, dict_name: str, parent_lock_id: int):
19 assert id(SharedDict.LOCK) == parent_lock_id
20 d = SharedDict(dict_name, None)
22 msg = f'Hello from shard {n}'
23 for x in range(0, 1000):
25 self.assertTrue(n in d)
26 self.assertEqual(msg, d[n])
27 y = d.get(random.randrange(0, 99), None)
32 def test_basic_operations(self):
33 dict_name = 'test_shared_dict'
34 d = SharedDict(dict_name, 4096)
36 self.assertEqual(dict_name, d.get_name())
39 f = self.doit(n, d.get_name(), id(SharedDict.LOCK))
41 smart_future.wait_all(results)
43 self.assertTrue(f.wrapped_future.done())
45 self.assertEqual(d[k], f'Hello from shard {k}')
51 @p.parallelize(method=p.Method.PROCESS)
52 def add_one(self, name: str, expected_lock_id: int):
54 self.assertEqual(id(SharedDict.LOCK), expected_lock_id)
62 def test_locking_works(self):
63 dict_name = 'test_shared_dict_lock'
64 d = SharedDict(dict_name, 4096)
69 f = self.add_one(d.get_name(), id(SharedDict.LOCK))
71 smart_future.wait_all(results)
72 self.assertEqual(10000, d["sum"])
78 if __name__ == '__main__':