persist_data.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. #
  2. # BitBake Test for lib/bb/persist_data/
  3. #
  4. # Copyright (C) 2018 Garmin Ltd.
  5. #
  6. # SPDX-License-Identifier: GPL-2.0-only
  7. #
  8. import unittest
  9. import bb.data
  10. import bb.persist_data
  11. import tempfile
  12. import threading
  13. class PersistDataTest(unittest.TestCase):
  14. def _create_data(self):
  15. return bb.persist_data.persist('TEST_PERSIST_DATA', self.d)
  16. def setUp(self):
  17. self.d = bb.data.init()
  18. self.tempdir = tempfile.TemporaryDirectory()
  19. self.d['PERSISTENT_DIR'] = self.tempdir.name
  20. self.data = self._create_data()
  21. self.items = {
  22. 'A1': '1',
  23. 'B1': '2',
  24. 'C2': '3'
  25. }
  26. self.stress_count = 10000
  27. self.thread_count = 5
  28. for k,v in self.items.items():
  29. self.data[k] = v
  30. def tearDown(self):
  31. self.tempdir.cleanup()
  32. def _iter_helper(self, seen, iterator):
  33. with iter(iterator):
  34. for v in iterator:
  35. self.assertTrue(v in seen)
  36. seen.remove(v)
  37. self.assertEqual(len(seen), 0, '%s not seen' % seen)
  38. def test_get(self):
  39. for k, v in self.items.items():
  40. self.assertEqual(self.data[k], v)
  41. self.assertIsNone(self.data.get('D'))
  42. with self.assertRaises(KeyError):
  43. self.data['D']
  44. def test_set(self):
  45. for k, v in self.items.items():
  46. self.data[k] += '-foo'
  47. for k, v in self.items.items():
  48. self.assertEqual(self.data[k], v + '-foo')
  49. def test_delete(self):
  50. self.data['D'] = '4'
  51. self.assertEqual(self.data['D'], '4')
  52. del self.data['D']
  53. self.assertIsNone(self.data.get('D'))
  54. with self.assertRaises(KeyError):
  55. self.data['D']
  56. def test_contains(self):
  57. for k in self.items:
  58. self.assertTrue(k in self.data)
  59. self.assertTrue(self.data.has_key(k))
  60. self.assertFalse('NotFound' in self.data)
  61. self.assertFalse(self.data.has_key('NotFound'))
  62. def test_len(self):
  63. self.assertEqual(len(self.data), len(self.items))
  64. def test_iter(self):
  65. self._iter_helper(set(self.items.keys()), self.data)
  66. def test_itervalues(self):
  67. self._iter_helper(set(self.items.values()), self.data.itervalues())
  68. def test_iteritems(self):
  69. self._iter_helper(set(self.items.items()), self.data.iteritems())
  70. def test_get_by_pattern(self):
  71. self._iter_helper({'1', '2'}, self.data.get_by_pattern('_1'))
  72. def _stress_read(self, data):
  73. for i in range(self.stress_count):
  74. for k in self.items:
  75. data[k]
  76. def _stress_write(self, data):
  77. for i in range(self.stress_count):
  78. for k, v in self.items.items():
  79. data[k] = v + str(i)
  80. def _validate_stress(self):
  81. for k, v in self.items.items():
  82. self.assertEqual(self.data[k], v + str(self.stress_count - 1))
  83. def test_stress(self):
  84. self._stress_read(self.data)
  85. self._stress_write(self.data)
  86. self._validate_stress()
  87. def test_stress_threads(self):
  88. def read_thread():
  89. data = self._create_data()
  90. self._stress_read(data)
  91. def write_thread():
  92. data = self._create_data()
  93. self._stress_write(data)
  94. threads = []
  95. for i in range(self.thread_count):
  96. threads.append(threading.Thread(target=read_thread))
  97. threads.append(threading.Thread(target=write_thread))
  98. for t in threads:
  99. t.start()
  100. self._stress_read(self.data)
  101. for t in threads:
  102. t.join()
  103. self._validate_stress()