persist_data.py 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144
  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. # This program is free software; you can redistribute it and/or modify
  11. # it under the terms of the GNU General Public License version 2 as
  12. # published by the Free Software Foundation.
  13. #
  14. # This program is distributed in the hope that it will be useful,
  15. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17. # GNU General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU General Public License along
  20. # with this program; if not, write to the Free Software Foundation, Inc.,
  21. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  22. #
  23. import unittest
  24. import bb.data
  25. import bb.persist_data
  26. import tempfile
  27. import threading
  28. class PersistDataTest(unittest.TestCase):
  29. def _create_data(self):
  30. return bb.persist_data.persist('TEST_PERSIST_DATA', self.d)
  31. def setUp(self):
  32. self.d = bb.data.init()
  33. self.tempdir = tempfile.TemporaryDirectory()
  34. self.d['PERSISTENT_DIR'] = self.tempdir.name
  35. self.data = self._create_data()
  36. self.items = {
  37. 'A1': '1',
  38. 'B1': '2',
  39. 'C2': '3'
  40. }
  41. self.stress_count = 10000
  42. self.thread_count = 5
  43. for k,v in self.items.items():
  44. self.data[k] = v
  45. def tearDown(self):
  46. self.tempdir.cleanup()
  47. def _iter_helper(self, seen, iterator):
  48. with iter(iterator):
  49. for v in iterator:
  50. self.assertTrue(v in seen)
  51. seen.remove(v)
  52. self.assertEqual(len(seen), 0, '%s not seen' % seen)
  53. def test_get(self):
  54. for k, v in self.items.items():
  55. self.assertEqual(self.data[k], v)
  56. self.assertIsNone(self.data.get('D'))
  57. with self.assertRaises(KeyError):
  58. self.data['D']
  59. def test_set(self):
  60. for k, v in self.items.items():
  61. self.data[k] += '-foo'
  62. for k, v in self.items.items():
  63. self.assertEqual(self.data[k], v + '-foo')
  64. def test_delete(self):
  65. self.data['D'] = '4'
  66. self.assertEqual(self.data['D'], '4')
  67. del self.data['D']
  68. self.assertIsNone(self.data.get('D'))
  69. with self.assertRaises(KeyError):
  70. self.data['D']
  71. def test_contains(self):
  72. for k in self.items:
  73. self.assertTrue(k in self.data)
  74. self.assertTrue(self.data.has_key(k))
  75. self.assertFalse('NotFound' in self.data)
  76. self.assertFalse(self.data.has_key('NotFound'))
  77. def test_len(self):
  78. self.assertEqual(len(self.data), len(self.items))
  79. def test_iter(self):
  80. self._iter_helper(set(self.items.keys()), self.data)
  81. def test_itervalues(self):
  82. self._iter_helper(set(self.items.values()), self.data.itervalues())
  83. def test_iteritems(self):
  84. self._iter_helper(set(self.items.items()), self.data.iteritems())
  85. def test_get_by_pattern(self):
  86. self._iter_helper({'1', '2'}, self.data.get_by_pattern('_1'))
  87. def _stress_read(self, data):
  88. for i in range(self.stress_count):
  89. for k in self.items:
  90. data[k]
  91. def _stress_write(self, data):
  92. for i in range(self.stress_count):
  93. for k, v in self.items.items():
  94. data[k] = v + str(i)
  95. def _validate_stress(self):
  96. for k, v in self.items.items():
  97. self.assertEqual(self.data[k], v + str(self.stress_count - 1))
  98. def test_stress(self):
  99. self._stress_read(self.data)
  100. self._stress_write(self.data)
  101. self._validate_stress()
  102. def test_stress_threads(self):
  103. def read_thread():
  104. data = self._create_data()
  105. self._stress_read(data)
  106. def write_thread():
  107. data = self._create_data()
  108. self._stress_write(data)
  109. threads = []
  110. for i in range(self.thread_count):
  111. threads.append(threading.Thread(target=read_thread))
  112. threads.append(threading.Thread(target=write_thread))
  113. for t in threads:
  114. t.start()
  115. self._stress_read(self.data)
  116. for t in threads:
  117. t.join()
  118. self._validate_stress()