Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_pathslicing.py
| # Copyright (C) 2015-2017 The Software Heritage developers | # Copyright (C) 2015-2017 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import shutil | import shutil | ||||
| import tempfile | import tempfile | ||||
| import unittest | import unittest | ||||
| from unittest.mock import patch, DEFAULT | |||||
| import gzip | import gzip | ||||
| from swh.model import hashutil | from swh.model import hashutil | ||||
| from swh.objstorage import exc, get_objstorage, ID_HASH_LENGTH | from swh.objstorage import exc, get_objstorage, ID_HASH_LENGTH | ||||
| from .objstorage_testing import ObjStorageTestFixture | from .objstorage_testing import ObjStorageTestFixture | ||||
| ▲ Show 20 Lines • Show All 87 Lines • ▼ Show 20 Lines | def test_iterate_from(self): | ||||
| self.assertEqual(len(ids), 0) | self.assertEqual(len(ids), 0) | ||||
| ids = list(self.storage.iter_from(all_ids[-2], n_leaf=True)) | ids = list(self.storage.iter_from(all_ids[-2], n_leaf=True)) | ||||
| n_leaf = ids[-1] | n_leaf = ids[-1] | ||||
| ids = ids[:-1] | ids = ids[:-1] | ||||
| self.assertEqual(n_leaf, 2) # beware, this depends on the hash algo | self.assertEqual(n_leaf, 2) # beware, this depends on the hash algo | ||||
| self.assertEqual(len(ids), 1) | self.assertEqual(len(ids), 1) | ||||
| self.assertEqual(ids, all_ids[-1:]) | self.assertEqual(ids, all_ids[-1:]) | ||||
| def test_fdatasync_default(self): | |||||
| content, obj_id = self.hash_content(b'check_fdatasync') | |||||
| with patch.multiple('os', fsync=DEFAULT, fdatasync=DEFAULT) as patched: | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| if self.storage.use_fdatasync: | |||||
| patched['fdatasync'].assert_called_once() | |||||
| patched['fsync'].assert_not_called() | |||||
| else: | |||||
| patched['fdatasync'].assert_not_called() | |||||
| patched['fsync'].assert_called_once() | |||||
| def test_fdatasync_forced_on(self): | |||||
| self.storage.use_fdatasync = True | |||||
| content, obj_id = self.hash_content(b'check_fdatasync') | |||||
| with patch.multiple('os', fsync=DEFAULT, fdatasync=DEFAULT) as patched: | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| patched['fdatasync'].assert_called_once() | |||||
| patched['fsync'].assert_not_called() | |||||
| def test_fdatasync_forced_off(self): | |||||
| self.storage.use_fdatasync = False | |||||
| content, obj_id = self.hash_content(b'check_fdatasync') | |||||
| with patch.multiple('os', fsync=DEFAULT, fdatasync=DEFAULT) as patched: | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| patched['fdatasync'].assert_not_called() | |||||
| patched['fsync'].assert_called_once() | |||||