Differential D5731 Diff 20521 swh/vault/tests/test_cookers.py

Changeset View

Standalone View

swh/vault/tests/test_cookers.py

Show First 20 Lines • Show All 194 Lines • ▼ Show 20 Lines	with cook_stream_revision_gitfast(storage, revision.id) as stream, test_repo as p:
processor = dulwich.fastexport.GitImportProcessor(test_repo.repo)		processor = dulwich.fastexport.GitImportProcessor(test_repo.repo)
processor.import_stream(stream)		processor.import_stream(stream)
test_repo.checkout(b"HEAD")		test_repo.checkout(b"HEAD")
shutil.rmtree(p / ".git")		shutil.rmtree(p / ".git")
yield p		yield p


@contextlib.contextmanager		@contextlib.contextmanager
def cook_extract_directory_git_bare(storage, obj_id, fsck=True):		def cook_extract_directory_git_bare(
		storage, obj_id, fsck=True, direct_objstorage=False
		):
		olasdUnsubmitted Not Done Inline Actions Maybe call this argument `use_objstorage` or `objstorage_direct`? olasd: Maybe call this argument `use_objstorage` or `objstorage_direct`?
"""Context manager that cooks a revision and extract it,		"""Context manager that cooks a revision and extract it,
using GitBareCooker"""		using GitBareCooker"""
backend = unittest.mock.MagicMock()		backend = unittest.mock.MagicMock()
backend.storage = storage		backend.storage = storage

# Cook the object		# Cook the object
cooker = GitBareCooker("directory", obj_id, backend=backend, storage=storage)		cooker = GitBareCooker(
		"directory",
		obj_id,
		backend=backend,
		storage=storage,
		objstorage=storage.objstorage if direct_objstorage else None,
		)
cooker.use_fsck = fsck # Some tests try edge-cases that git-fsck rejects		cooker.use_fsck = fsck # Some tests try edge-cases that git-fsck rejects
cooker.fileobj = io.BytesIO()		cooker.fileobj = io.BytesIO()
assert cooker.check_exists()		assert cooker.check_exists()
cooker.prepare_bundle()		cooker.prepare_bundle()
cooker.fileobj.seek(0)		cooker.fileobj.seek(0)

# Extract it		# Extract it
with tempfile.TemporaryDirectory(prefix="tmp-vault-extract-") as td:		with tempfile.TemporaryDirectory(prefix="tmp-vault-extract-") as td:
▲ Show 20 Lines • Show All 114 Lines • ▼ Show 20 Lines	def test_directory_simple(self, git_loader, cook_extract_directory):
obj_id_hex = repo.repo[c].tree.decode()		obj_id_hex = repo.repo[c].tree.decode()
obj_id = hashutil.hash_to_bytes(obj_id_hex)		obj_id = hashutil.hash_to_bytes(obj_id_hex)

with cook_extract_directory(loader.storage, obj_id) as p:		with cook_extract_directory(loader.storage, obj_id) as p:
assert (p / "file").stat().st_mode == 0o100644		assert (p / "file").stat().st_mode == 0o100644
assert (p / "file").read_text() == TEST_CONTENT		assert (p / "file").read_text() == TEST_CONTENT
assert (p / "executable").stat().st_mode == 0o100755		assert (p / "executable").stat().st_mode == 0o100755
assert (p / "executable").read_bytes() == TEST_EXECUTABLE		assert (p / "executable").read_bytes() == TEST_EXECUTABLE
assert (p / "link").is_symlink		assert (p / "link").is_symlink()
assert os.readlink(str(p / "link")) == "file"		assert os.readlink(str(p / "link")) == "file"
assert (p / "dir1/dir2/file").stat().st_mode == 0o100644		assert (p / "dir1/dir2/file").stat().st_mode == 0o100644
assert (p / "dir1/dir2/file").read_text() == TEST_CONTENT		assert (p / "dir1/dir2/file").read_text() == TEST_CONTENT

directory = from_disk.Directory.from_disk(path=bytes(p))		directory = from_disk.Directory.from_disk(path=bytes(p))
assert obj_id_hex == hashutil.hash_to_hex(directory.hash)		assert obj_id_hex == hashutil.hash_to_hex(directory.hash)

def test_directory_filtered_objects(self, git_loader, cook_extract_directory):		def test_directory_filtered_objects(self, git_loader, cook_extract_directory):
▲ Show 20 Lines • Show All 81 Lines • ▼ Show 20 Lines	def test_directory_bogus_perms(self, git_loader, cook_extract_directory):

obj_id_hex = repo.repo[c].tree.decode()		obj_id_hex = repo.repo[c].tree.decode()
obj_id = hashutil.hash_to_bytes(obj_id_hex)		obj_id = hashutil.hash_to_bytes(obj_id_hex)

with cook_extract_directory(loader.storage, obj_id) as p:		with cook_extract_directory(loader.storage, obj_id) as p:
assert (p / "file").stat().st_mode == 0o100644		assert (p / "file").stat().st_mode == 0o100644
assert (p / "executable").stat().st_mode == 0o100755		assert (p / "executable").stat().st_mode == 0o100755
assert (p / "wat").stat().st_mode == 0o100644		assert (p / "wat").stat().st_mode == 0o100644

		olasdUnsubmitted Not Done Inline Actions Same here. olasd: Same here.
		@pytest.mark.parametrize("direct_objstorage", [True, False])
		def test_directory_objstorage(
		self, swh_storage, git_loader, mocker, direct_objstorage
		):
		"""Like test_directory_simple, but using swh_objstorage directly, without
		going through swh_storage.content_get_data()"""
		repo = TestRepo()
		with repo as rp:
		(rp / "file").write_text(TEST_CONTENT)
		(rp / "executable").write_bytes(TEST_EXECUTABLE)
		(rp / "executable").chmod(0o755)
		(rp / "link").symlink_to("file")
		(rp / "dir1/dir2").mkdir(parents=True)
		(rp / "dir1/dir2/file").write_text(TEST_CONTENT)
		c = repo.commit()
		loader = git_loader(str(rp))
		loader.load()

		obj_id_hex = repo.repo[c].tree.decode()
		obj_id = hashutil.hash_to_bytes(obj_id_hex)

		# Set-up spies
		storage_content_get_data = mocker.patch.object(
		swh_storage, "content_get_data", wraps=swh_storage.content_get_data
		)
		objstorage_content_batch = mocker.patch.object(
		swh_storage.objstorage, "get_batch", wraps=swh_storage.objstorage.get_batch
		)

		with cook_extract_directory_git_bare(
		loader.storage, obj_id, direct_objstorage=direct_objstorage
		) as p:
		assert (p / "file").stat().st_mode == 0o100644
		assert (p / "file").read_text() == TEST_CONTENT
		olasdUnsubmitted Not Done Inline Actions I think `Path.is_symlink` is a method rather than a property, so this assert is always true (although the next call would fail if it wasn't anyway). olasd: I think `Path.is_symlink` is a method rather than a property, so this assert is always true…
		assert (p / "executable").stat().st_mode == 0o100755
		assert (p / "executable").read_bytes() == TEST_EXECUTABLE
		assert (p / "link").is_symlink()
		assert os.readlink(str(p / "link")) == "file"
		assert (p / "dir1/dir2/file").stat().st_mode == 0o100644
		assert (p / "dir1/dir2/file").read_text() == TEST_CONTENT

		directory = from_disk.Directory.from_disk(path=bytes(p))
		assert obj_id_hex == hashutil.hash_to_hex(directory.hash)

		if direct_objstorage:
		storage_content_get_data.assert_not_called()
		objstorage_content_batch.assert_called()
		else:
		storage_content_get_data.assert_called()
		objstorage_content_batch.assert_not_called()

def test_directory_revision_data(self, swh_storage):		def test_directory_revision_data(self, swh_storage):
target_rev = "0e8a3ad980ec179856012b7eecf4327e99cd44cd"		target_rev = "0e8a3ad980ec179856012b7eecf4327e99cd44cd"

dir = Directory(		dir = Directory(
entries=(		entries=(
DirectoryEntry(		DirectoryEntry(
name=b"submodule",		name=b"submodule",
type="rev",		type="rev",
Show All 36 Lines	def test_revision_simple(self, git_loader, cook_extract_revision):
loader.load()		loader.load()
obj_id_hex = repo.repo.refs[b"HEAD"].decode()		obj_id_hex = repo.repo.refs[b"HEAD"].decode()
obj_id = hashutil.hash_to_bytes(obj_id_hex)		obj_id = hashutil.hash_to_bytes(obj_id_hex)

with cook_extract_revision(loader.storage, obj_id) as (ert, p):		with cook_extract_revision(loader.storage, obj_id) as (ert, p):
ert.checkout(b"HEAD")		ert.checkout(b"HEAD")
assert (p / "file1").stat().st_mode == 0o100644		assert (p / "file1").stat().st_mode == 0o100644
assert (p / "file1").read_text() == TEST_CONTENT		assert (p / "file1").read_text() == TEST_CONTENT
assert (p / "link1").is_symlink		assert (p / "link1").is_symlink()
assert os.readlink(str(p / "link1")) == "file1"		assert os.readlink(str(p / "link1")) == "file1"
assert (p / "bin").stat().st_mode == 0o100755		assert (p / "bin").stat().st_mode == 0o100755
assert (p / "bin").read_bytes() == TEST_EXECUTABLE		assert (p / "bin").read_bytes() == TEST_EXECUTABLE
assert (p / "dir1/dir2/file").read_text() == TEST_CONTENT		assert (p / "dir1/dir2/file").read_text() == TEST_CONTENT
assert (p / "dir1/dir2/file").stat().st_mode == 0o100644		assert (p / "dir1/dir2/file").stat().st_mode == 0o100644
assert ert.repo.refs[b"HEAD"].decode() == obj_id_hex		assert ert.repo.refs[b"HEAD"].decode() == obj_id_hex

def test_revision_two_roots(self, git_loader, cook_extract_revision):		def test_revision_two_roots(self, git_loader, cook_extract_revision):
▲ Show 20 Lines • Show All 194 Lines • Show Last 20 Lines