persist_data.py 3.7 KB

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