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: | |||||
assert patched['fdatasync'].call_count == 1 | |||||
assert patched['fsync'].call_count == 0 | |||||
else: | |||||
assert patched['fdatasync'].call_count == 0 | |||||
assert patched['fsync'].call_count == 1 | |||||
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) | |||||
assert patched['fdatasync'].call_count == 1 | |||||
assert patched['fsync'].call_count == 0 | |||||
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) | |||||
assert patched['fdatasync'].call_count == 0 | |||||
assert patched['fsync'].call_count == 1 |