Changeset View
Changeset View
Standalone View
Standalone View
swh/model/tests/test_from_disk.py
# Copyright (C) 2017 The Software Heritage developers | # Copyright (C) 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 os | import os | ||||
import pytest | import pytest | ||||
import tarfile | import tarfile | ||||
import tempfile | import tempfile | ||||
import unittest | import unittest | ||||
from typing import ClassVar, Optional | from typing import ClassVar, Optional | ||||
from swh.model import from_disk | from swh.model import from_disk | ||||
from swh.model.from_disk import Content, DentryPerms, Directory | from swh.model.from_disk import ( | ||||
Content, DentryPerms, Directory, DiskBackedContent | |||||
) | |||||
from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes, hash_to_hex | from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes, hash_to_hex | ||||
from swh.model import model | |||||
TEST_DATA = os.path.join(os.path.dirname(__file__), 'data') | TEST_DATA = os.path.join(os.path.dirname(__file__), 'data') | ||||
class ModeToPerms(unittest.TestCase): | class ModeToPerms(unittest.TestCase): | ||||
def setUp(self): | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
Show All 18 Lines | def setUp(self): | ||||
else: | else: | ||||
self.perms_map[i] = DentryPerms.content | self.perms_map[i] = DentryPerms.content | ||||
def test_exhaustive_mode_to_perms(self): | def test_exhaustive_mode_to_perms(self): | ||||
for fmode, perm in self.perms_map.items(): | for fmode, perm in self.perms_map.items(): | ||||
self.assertEqual(perm, from_disk.mode_to_perms(fmode)) | self.assertEqual(perm, from_disk.mode_to_perms(fmode)) | ||||
class TestDiskBackedContent(unittest.TestCase): | |||||
def test_with_data(self): | |||||
expected_content = model.Content( | |||||
length=42, status='visible', data=b'foo bar', | |||||
sha1=b'foo', sha1_git=b'bar', sha256=b'baz', blake2s256=b'qux') | |||||
with tempfile.NamedTemporaryFile(mode='w+b') as fd: | |||||
content = DiskBackedContent( | |||||
length=42, status='visible', path=fd.name, | |||||
sha1=b'foo', sha1_git=b'bar', sha256=b'baz', blake2s256=b'qux') | |||||
fd.write(b'foo bar') | |||||
fd.seek(0) | |||||
content_with_data = content.with_data() | |||||
assert expected_content == content_with_data | |||||
def test_lazy_data(self): | |||||
with tempfile.NamedTemporaryFile(mode='w+b') as fd: | |||||
fd.write(b'foo') | |||||
fd.seek(0) | |||||
content = DiskBackedContent( | |||||
length=42, status='visible', path=fd.name, | |||||
sha1=b'foo', sha1_git=b'bar', sha256=b'baz', blake2s256=b'qux') | |||||
fd.write(b'bar') | |||||
fd.seek(0) | |||||
content_with_data = content.with_data() | |||||
fd.write(b'baz') | |||||
fd.seek(0) | |||||
assert content_with_data.data == b'bar' | |||||
def test_with_data_cannot_read(self): | |||||
olasd: I don't think the expected content is that useful. The test would be easier to follow with just… | |||||
with tempfile.NamedTemporaryFile(mode='w+b') as fd: | |||||
content = DiskBackedContent( | |||||
length=42, status='visible', path=fd.name, | |||||
sha1=b'foo', sha1_git=b'bar', sha256=b'baz', blake2s256=b'qux') | |||||
with pytest.raises(OSError): | |||||
content.with_data() | |||||
def test_missing_path(self): | |||||
with pytest.raises(TypeError): | |||||
DiskBackedContent( | |||||
length=42, status='visible', | |||||
sha1=b'foo', sha1_git=b'bar', sha256=b'baz', blake2s256=b'qux') | |||||
with pytest.raises(TypeError): | |||||
DiskBackedContent( | |||||
length=42, status='visible', path=None, | |||||
sha1=b'foo', sha1_git=b'bar', sha256=b'baz', blake2s256=b'qux') | |||||
class DataMixin: | class DataMixin: | ||||
maxDiff = None # type: ClassVar[Optional[int]] | maxDiff = None # type: ClassVar[Optional[int]] | ||||
def setUp(self): | def setUp(self): | ||||
self.tmpdir = tempfile.TemporaryDirectory( | self.tmpdir = tempfile.TemporaryDirectory( | ||||
prefix='swh.model.from_disk' | prefix='swh.model.from_disk' | ||||
) | ) | ||||
self.tmpdir_name = os.fsencode(self.tmpdir.name) | self.tmpdir_name = os.fsencode(self.tmpdir.name) | ||||
▲ Show 20 Lines • Show All 337 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
'b9ad6b44c9b383621faa482b990a367d' | 'b9ad6b44c9b383621faa482b990a367d' | ||||
), | ), | ||||
}, | }, | ||||
} | } | ||||
def tearDown(self): | def tearDown(self): | ||||
self.tmpdir.cleanup() | self.tmpdir.cleanup() | ||||
def assertContentEqual(self, left, right, *, check_data=False, # noqa | def assertContentEqual(self, left, right, *, # noqa | ||||
check_path=False): | check_path=False): | ||||
if not isinstance(left, Content): | if not isinstance(left, Content): | ||||
raise ValueError('%s is not a Content' % left) | raise ValueError('%s is not a Content' % left) | ||||
if isinstance(right, Content): | if isinstance(right, Content): | ||||
right = right.get_data() | right = right.get_data() | ||||
# Compare dictionaries | |||||
keys = DEFAULT_ALGORITHMS | { | keys = DEFAULT_ALGORITHMS | { | ||||
'length', | 'length', | ||||
'perms', | 'perms', | ||||
} | } | ||||
if check_data: | |||||
keys |= {'data'} | |||||
if check_path: | if check_path: | ||||
keys |= {'path'} | keys |= {'path'} | ||||
failed = [] | failed = [] | ||||
for key in keys: | for key in keys: | ||||
try: | try: | ||||
lvalue = left.data[key] | lvalue = left.data[key] | ||||
if key == 'perms' and 'perms' not in right: | if key == 'perms' and 'perms' not in right: | ||||
Show All 19 Lines | class DataMixin: | ||||
def assertDirectoryEqual(self, left, right): # NoQA | def assertDirectoryEqual(self, left, right): # NoQA | ||||
if not isinstance(left, Directory): | if not isinstance(left, Directory): | ||||
raise ValueError('%s is not a Directory' % left) | raise ValueError('%s is not a Directory' % left) | ||||
if isinstance(right, Directory): | if isinstance(right, Directory): | ||||
right = right.get_data() | right = right.get_data() | ||||
assert left.entries == right['entries'] | assert left.entries == right['entries'] | ||||
assert left.hash == right['id'] | |||||
assert left.to_model() == model.Directory.from_dict(right) | |||||
def make_contents(self, directory): | def make_contents(self, directory): | ||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
path = os.path.join(directory, filename) | path = os.path.join(directory, filename) | ||||
with open(path, 'wb') as f: | with open(path, 'wb') as f: | ||||
f.write(content['data']) | f.write(content['data']) | ||||
os.chmod(path, content['mode']) | os.chmod(path, content['mode']) | ||||
Show All 33 Lines | class SymlinkToContent(DataMixin, unittest.TestCase): | ||||
def test_symlink_to_content(self): | def test_symlink_to_content(self): | ||||
for filename, symlink in self.symlinks.items(): | for filename, symlink in self.symlinks.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
perms = 0o120000 | perms = 0o120000 | ||||
conv_content = Content.from_symlink(path=path, mode=perms) | conv_content = Content.from_symlink(path=path, mode=perms) | ||||
self.assertContentEqual(conv_content, symlink) | self.assertContentEqual(conv_content, symlink) | ||||
def test_symlink_to_base_model(self): | |||||
for filename, symlink in self.symlinks.items(): | |||||
path = os.path.join(self.tmpdir_name, filename) | |||||
perms = 0o120000 | |||||
model_content = \ | |||||
Content.from_symlink(path=path, mode=perms).to_model() | |||||
right = symlink.copy() | |||||
for key in ('perms', 'path', 'mode'): | |||||
right.pop(key, None) | |||||
right['status'] = 'visible' | |||||
assert model_content == model.Content.from_dict(right) | |||||
class FileToContent(DataMixin, unittest.TestCase): | class FileToContent(DataMixin, unittest.TestCase): | ||||
def setUp(self): | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
self.make_contents(self.tmpdir_name) | self.make_contents(self.tmpdir_name) | ||||
self.make_symlinks(self.tmpdir_name) | self.make_symlinks(self.tmpdir_name) | ||||
self.make_specials(self.tmpdir_name) | self.make_specials(self.tmpdir_name) | ||||
def test_symlink_to_content(self): | def test_symlink_to_content(self): | ||||
# Check whether loading the data works | |||||
for data in [True, False]: | |||||
for filename, symlink in self.symlinks.items(): | for filename, symlink in self.symlinks.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
conv_content = Content.from_file(path=path, data=data) | conv_content = Content.from_file(path=path) | ||||
self.assertContentEqual(conv_content, symlink, check_data=data) | self.assertContentEqual(conv_content, symlink) | ||||
def test_file_to_content(self): | def test_file_to_content(self): | ||||
for data in [True, False]: | |||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
conv_content = Content.from_file(path=path, data=data) | conv_content = Content.from_file(path=path) | ||||
self.assertContentEqual(conv_content, content, check_data=data) | self.assertContentEqual(conv_content, content) | ||||
def test_special_to_content(self): | def test_special_to_content(self): | ||||
for data in [True, False]: | |||||
for filename in self.specials: | for filename in self.specials: | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
conv_content = Content.from_file(path=path, data=data) | conv_content = Content.from_file(path=path) | ||||
self.assertContentEqual(conv_content, self.empty_content) | self.assertContentEqual(conv_content, self.empty_content) | ||||
for path in ['/dev/null', '/dev/zero']: | for path in ['/dev/null', '/dev/zero']: | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
conv_content = Content.from_file(path=path, data=data) | conv_content = Content.from_file(path=path) | ||||
self.assertContentEqual(conv_content, self.empty_content) | self.assertContentEqual(conv_content, self.empty_content) | ||||
def test_symlink_to_content_model(self): | |||||
for filename, symlink in self.symlinks.items(): | |||||
path = os.path.join(self.tmpdir_name, filename) | |||||
model_content = Content.from_file(path=path).to_model() | |||||
right = symlink.copy() | |||||
for key in ('perms', 'path', 'mode'): | |||||
right.pop(key, None) | |||||
right['status'] = 'visible' | |||||
assert model_content == model.Content.from_dict(right) | |||||
def test_file_to_content_model(self): | |||||
for filename, content in self.contents.items(): | |||||
path = os.path.join(self.tmpdir_name, filename) | |||||
model_content = Content.from_file(path=path).to_model() | |||||
right = content.copy() | |||||
for key in ('perms', 'mode'): | |||||
right.pop(key, None) | |||||
assert model_content.with_data() == model.Content.from_dict(right) | |||||
right['path'] = path | |||||
del right['data'] | |||||
assert model_content == DiskBackedContent.from_dict(right) | |||||
def test_special_to_content_model(self): | |||||
for filename in self.specials: | |||||
path = os.path.join(self.tmpdir_name, filename) | |||||
model_content = Content.from_file(path=path).to_model() | |||||
right = self.empty_content.copy() | |||||
for key in ('perms', 'path', 'mode'): | |||||
right.pop(key, None) | |||||
right['status'] = 'visible' | |||||
assert model_content == model.Content.from_dict(right) | |||||
for path in ['/dev/null', '/dev/zero']: | |||||
model_content = Content.from_file(path=path).to_model() | |||||
right = self.empty_content.copy() | |||||
for key in ('perms', 'path', 'mode'): | |||||
right.pop(key, None) | |||||
right['status'] = 'visible' | |||||
assert model_content == model.Content.from_dict(right) | |||||
def test_symlink_max_length(self): | def test_symlink_max_length(self): | ||||
for max_content_length in [4, 10]: | for max_content_length in [4, 10]: | ||||
for filename, symlink in self.symlinks.items(): | for filename, symlink in self.symlinks.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
content = Content.from_file(path=path, data=True) | content = Content.from_file(path=path) | ||||
if content.data['length'] > max_content_length: | if content.data['length'] > max_content_length: | ||||
with pytest.raises(Exception, match='too large'): | with pytest.raises(Exception, match='too large'): | ||||
Content.from_file( | Content.from_file( | ||||
path=path, data=True, | path=path, | ||||
max_content_length=max_content_length) | max_content_length=max_content_length) | ||||
else: | else: | ||||
limited_content = Content.from_file( | limited_content = Content.from_file( | ||||
path=path, data=True, | path=path, | ||||
max_content_length=max_content_length) | max_content_length=max_content_length) | ||||
assert content == limited_content | assert content == limited_content | ||||
def test_file_max_length(self): | def test_file_max_length(self): | ||||
for max_content_length in [2, 4]: | for max_content_length in [2, 4]: | ||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
content = Content.from_file(path=path, data=True) | content = Content.from_file(path=path) | ||||
limited_content = Content.from_file( | limited_content = Content.from_file( | ||||
path=path, data=True, | path=path, | ||||
max_content_length=max_content_length) | max_content_length=max_content_length) | ||||
assert content.data['length'] == limited_content.data['length'] | assert content.data['length'] == limited_content.data['length'] | ||||
assert content.data['status'] == 'visible' | assert content.data['status'] == 'visible' | ||||
if content.data['length'] > max_content_length: | if content.data['length'] > max_content_length: | ||||
assert limited_content.data['status'] == 'absent' | assert limited_content.data['status'] == 'absent' | ||||
assert limited_content.data['reason'] \ | assert limited_content.data['reason'] \ | ||||
== 'Content too large' | == 'Content too large' | ||||
else: | else: | ||||
assert limited_content.data['status'] == 'visible' | assert limited_content.data['status'] == 'visible' | ||||
assert limited_content.data['data'] == content.data['data'] | |||||
def test_special_file_max_length(self): | def test_special_file_max_length(self): | ||||
for max_content_length in [None, 0, 1]: | for max_content_length in [None, 0, 1]: | ||||
for filename in self.specials: | for filename in self.specials: | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
content = Content.from_file(path=path, data=True) | content = Content.from_file(path=path) | ||||
limited_content = Content.from_file( | limited_content = Content.from_file( | ||||
path=path, data=True, | path=path, | ||||
max_content_length=max_content_length) | max_content_length=max_content_length) | ||||
assert limited_content == content | assert limited_content == content | ||||
def test_file_to_content_with_path(self): | def test_file_to_content_with_path(self): | ||||
for filename, content in self.contents.items(): | for filename, content in self.contents.items(): | ||||
content_w_path = content.copy() | content_w_path = content.copy() | ||||
path = os.path.join(self.tmpdir_name, filename) | path = os.path.join(self.tmpdir_name, filename) | ||||
content_w_path['path'] = path | content_w_path['path'] = path | ||||
conv_content = Content.from_file(path=path, save_path=True) | conv_content = Content.from_file(path=path) | ||||
self.assertContentEqual(conv_content, content_w_path, | self.assertContentEqual(conv_content, content_w_path, | ||||
check_path=True) | check_path=True) | ||||
@pytest.mark.fs | @pytest.mark.fs | ||||
class DirectoryToObjects(DataMixin, unittest.TestCase): | class DirectoryToObjects(DataMixin, unittest.TestCase): | ||||
def setUp(self): | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
▲ Show 20 Lines • Show All 163 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
super().setUp() | super().setUp() | ||||
self.make_from_tarball(self.tmpdir_name) | self.make_from_tarball(self.tmpdir_name) | ||||
def test_contents_match(self): | def test_contents_match(self): | ||||
directory = Directory.from_disk( | directory = Directory.from_disk( | ||||
path=os.path.join(self.tmpdir_name, b'sample-folder') | path=os.path.join(self.tmpdir_name, b'sample-folder') | ||||
) | ) | ||||
for name, data in self.tarball_contents.items(): | for name, expected in self.tarball_contents.items(): | ||||
obj = directory[name] | obj = directory[name] | ||||
if isinstance(obj, Content): | if isinstance(obj, Content): | ||||
self.assertContentEqual(obj, data) | self.assertContentEqual(obj, expected) | ||||
elif isinstance(obj, Directory): | elif isinstance(obj, Directory): | ||||
self.assertDirectoryEqual(obj, data) | self.assertDirectoryEqual(obj, expected) | ||||
else: | else: | ||||
raise self.failureException('Unknown type for %s' % obj) | raise self.failureException('Unknown type for %s' % obj) | ||||
class DirectoryManipulation(DataMixin, unittest.TestCase): | class DirectoryManipulation(DataMixin, unittest.TestCase): | ||||
def test_directory_access_nested(self): | def test_directory_access_nested(self): | ||||
d = Directory() | d = Directory() | ||||
d[b'a'] = Directory() | d[b'a'] = Directory() | ||||
▲ Show 20 Lines • Show All 75 Lines • Show Last 20 Lines |
I don't think the expected content is that useful. The test would be easier to follow with just assert content_with_data.data == b'bar'.
Using explicit contents such as b'written before instantiation', b'written before with_data()', b'written after with_data()' instead of metasyntactic stuff would make the test even clearer.