note that there is a dependency loop between this diff and D2023, because we do have an import loop between swh-storage and swh-journal (even after D1993)

the dependency between this diff and D2023 is not declared here, because phabricator prevents loops

vlorentz accepted this revision.Sep 23 2019, 12:31 PM

This revision is now accepted and ready to land.Sep 23 2019, 12:31 PM

zack removed a child revision: D2023: typing: minimal changes to make a no-op mypy run pass.Sep 24 2019, 1:27 PM

zack added a parent revision: D2023: typing: minimal changes to make a no-op mypy run pass.

init.py: switch to documented way of extending path

Harbormaster failed remote builds in B7958: Diff 6836!Sep 27 2019, 10:42 AM

Build has FAILED

Link to build: https://jenkins.softwareheritage.org/job/DJNL/job/tox/274/
See console output for more information: https://jenkins.softwareheritage.org/job/DJNL/job/tox/274/console

typing: minimal changes to make a no-op mypy run pass
init.py: switch to documented way of extending path
tox: anticipate mypy run to just after flake8

Build has FAILED

Link to build: https://jenkins.softwareheritage.org/job/DJNL/job/tox/275/
See console output for more information: https://jenkins.softwareheritage.org/job/DJNL/job/tox/275/console

Harbormaster failed remote builds in B8023: Diff 6881!Oct 1 2019, 9:05 AM

Build is green
See https://jenkins.softwareheritage.org/job/DJNL/job/tox/276/ for more details.

Harbormaster completed remote builds in B8023: Diff 6881.Oct 1 2019, 10:18 AM

Closed by commit rDJNLd541a5562118: typing: minimal changes to make a no-op mypy run pass (authored by zack). · Explain WhyOct 1 2019, 10:18 AM

This revision was automatically updated to reflect the committed changes.

zack added a commit: rDJNLd541a5562118: typing: minimal changes to make a no-op mypy run pass.

zack added a commit: rDJNLdec930105642: init.py: switch to documented way of extending path.

zack added a commit: rDJNL045b142a3446: tox: anticipate mypy run to just after flake8.

Revision Contents
Changeset List

Path

Size

.gitignore

1 line

MANIFEST.in

1 line

mypy.ini

21 lines

swh/

__init__.py

5 lines

journal/

py.typed

1 line

tests/

conftest.py

4 lines

test_cli.py

4 lines

tox.ini

10 lines

Diff 6887

View Options

.gitignore

# 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):
		olasdUnsubmitted Not Done Inline Actions 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. 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

View Options

MANIFEST.in

# 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):
		olasdUnsubmitted Not Done Inline Actions 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. 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

View Options

mypy.ini

# 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):
		olasdUnsubmitted Not Done Inline Actions 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. 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

View Options

swh/init.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):
		olasdUnsubmitted Not Done Inline Actions 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. 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