(self, salt, items, expected_order, disable_shuffling=False)
| 195 | class ShuffleTest(testing.TestCase): |
| 196 | |
| 197 | def _test_items(self, salt, items, expected_order, disable_shuffling=False): |
| 198 | shuffler = shuffle.Shuffler(self.get_temp_dir(), salt, disable_shuffling) |
| 199 | for key, item in items: |
| 200 | shuffler.add(key, item) |
| 201 | self.assertEqual(shuffler.size, _TOTAL_SIZE) |
| 202 | if not shuffler._in_memory: # Check size of temporary bucket files |
| 203 | expected_size = (16 + 8) * len(items) + sum(len(t[1]) for t in items) |
| 204 | size = 0 |
| 205 | for bucket in shuffler._buckets: |
| 206 | if not bucket._fobj: |
| 207 | continue |
| 208 | bucket._fobj.close() |
| 209 | with open(bucket._path, 'rb') as f: |
| 210 | size += len(f.read()) |
| 211 | self.assertEqual(size, expected_size) |
| 212 | # Check records can be read as expected: |
| 213 | records = list(ex for _, ex in shuffler) |
| 214 | self.assertEqual(records, expected_order) |
| 215 | |
| 216 | def test_all_mem(self): |
| 217 | self._test_items('split1', _ITEMS, _ORDERED_ITEMS_SPLIT1) |
no test coverage detected