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.