Page MenuHomeSoftware Heritage
Differential D737 Diff 2348 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

# Copyright (C) 2015-2018 The Software Heritage developers # Copyright (C) 2015-2018 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 copy import copy
import datetime import datetime
import unittest import unittest
import itertools
from collections import defaultdict from collections import defaultdict
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
import pytest import pytest
from hypothesis import given, strategies from hypothesis import given, strategies
from swh.model import from_disk, identifiers from swh.model import from_disk, identifiers
▲ Show 20 Lines • Show All 151 Lines • ▼ Show 20 Lines def setUp(self):
'entries': [ 'entries': [
{ {
'name': b'foo', 'name': b'foo',
'type': 'file', 'type': 'file',
'target': self.cont['sha1_git'], 'target': self.cont['sha1_git'],
'perms': from_disk.DentryPerms.content, 'perms': from_disk.DentryPerms.content,
}, },
{ {
'name': b'bar', 'name': b'subdir',
'type': 'dir', 'type': 'dir',
'target': b'12345678901234560000', 'target': self.dir['id'],
'perms': from_disk.DentryPerms.directory, 'perms': from_disk.DentryPerms.directory,
}, },
{ {
'name': b'hello', 'name': b'hello',
'type': 'file', 'type': 'file',
'target': b'12345678901234567890', 'target': b'12345678901234567890',
'perms': from_disk.DentryPerms.content, 'perms': from_disk.DentryPerms.content,
}, },
▲ Show 20 Lines • Show All 491 Lines • ▼ Show 20 Lines def test_content_get_metadata_missing_sha1(self):
# All the metadata keys are None # All the metadata keys are None
missing_cont.pop('data') missing_cont.pop('data')
for key in list(missing_cont): for key in list(missing_cont):
if key != 'sha1': if key != 'sha1':
missing_cont[key] = None missing_cont[key] = None
self.assertEqual(list(gen), [missing_cont]) self.assertEqual(list(gen), [missing_cont])
def test_directory_add(self): @staticmethod
init_missing = list(self.storage.directory_missing([self.dir['id']])) def _transform_entries(dir_, *, prefix=b''):
self.assertEqual([self.dir['id']], init_missing) for ent in dir_['entries']:
yield {
self.storage.directory_add([self.dir]) 'dir_id': dir_['id'],
stored_data = list(self.storage.directory_ls(self.dir['id']))
data_to_store = []
for ent in self.dir['entries']:
data_to_store.append({
'dir_id': self.dir['id'],
'type': ent['type'], 'type': ent['type'],
'target': ent['target'], 'target': ent['target'],
'name': ent['name'], 'name': prefix + ent['name'],
'perms': ent['perms'], 'perms': ent['perms'],
'status': None, 'status': None,
'sha1': None, 'sha1': None,
'sha1_git': None, 'sha1_git': None,
'sha256': None, 'sha256': None,
'length': None, 'length': None,
}) }
self.assertCountEqual(data_to_store, stored_data) def test_directory_add(self):
init_missing = list(self.storage.directory_missing([self.dir['id']]))
self.assertEqual([self.dir['id']], init_missing)
self.storage.directory_add([self.dir])
actual_data = list(self.storage.directory_ls(self.dir['id']))
expected_data = list(self._transform_entries(self.dir))
self.assertCountEqual(expected_data, actual_data)
after_missing = list(self.storage.directory_missing([self.dir['id']])) after_missing = list(self.storage.directory_missing([self.dir['id']]))
self.assertEqual([], after_missing) self.assertEqual([], after_missing)
def test_directory_get_recursive(self):
init_missing = list(self.storage.directory_missing([self.dir['id']]))
self.assertEqual([self.dir['id']], init_missing)
self.storage.directory_add([self.dir, self.dir2, self.dir3])
actual_data = list(self.storage.directory_ls(
self.dir['id'], recursive=True))
expected_data = list(self._transform_entries(self.dir))
self.assertCountEqual(expected_data, actual_data)
actual_data = list(self.storage.directory_ls(
self.dir2['id'], recursive=True))
expected_data = list(self._transform_entries(self.dir2))
self.assertCountEqual(expected_data, actual_data)
actual_data = list(self.storage.directory_ls(
self.dir3['id'], recursive=True))
expected_data = list(itertools.chain(
self._transform_entries(self.dir3),
self._transform_entries(self.dir, prefix=b'subdir/')))
self.assertCountEqual(expected_data, actual_data)
def test_directory_entry_get_by_path(self): def test_directory_entry_get_by_path(self):
# given # given
init_missing = list(self.storage.directory_missing([self.dir3['id']])) init_missing = list(self.storage.directory_missing([self.dir3['id']]))
self.assertEqual([self.dir3['id']], init_missing) self.assertEqual([self.dir3['id']], init_missing)
self.storage.directory_add([self.dir3]) self.storage.directory_add([self.dir3])
expected_entries = [ expected_entries = [
{ {
'dir_id': self.dir3['id'], 'dir_id': self.dir3['id'],
'name': b'foo', 'name': b'foo',
'type': 'file', 'type': 'file',
'target': self.cont['sha1_git'], 'target': self.cont['sha1_git'],
'sha1': None, 'sha1': None,
'sha1_git': None, 'sha1_git': None,
'sha256': None, 'sha256': None,
'status': None, 'status': None,
'perms': from_disk.DentryPerms.content, 'perms': from_disk.DentryPerms.content,
'length': None, 'length': None,
}, },
{ {
'dir_id': self.dir3['id'], 'dir_id': self.dir3['id'],
'name': b'bar', 'name': b'subdir',
'type': 'dir', 'type': 'dir',
'target': b'12345678901234560000', 'target': self.dir['id'],
'sha1': None, 'sha1': None,
'sha1_git': None, 'sha1_git': None,
'sha256': None, 'sha256': None,
'status': None, 'status': None,
'perms': from_disk.DentryPerms.directory, 'perms': from_disk.DentryPerms.directory,
'length': None, 'length': None,
}, },
{ {
▲ Show 20 Lines • Show All 1,435 Lines • Show Last 20 Lines
About · Developer information · Legal