Differential D5731 Diff 20464 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, 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 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 213 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

		@pytest.mark.parametrize("objstorage", [True, False])
		olasdUnsubmitted Not Done Inline Actions Same here. olasd: Same here.
		def test_directory_objstorage(self, swh_storage, git_loader, mocker, 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, objstorage=objstorage
		) as p:
		assert (p / "file").stat().st_mode == 0o100644
		assert (p / "file").read_text() == TEST_CONTENT
		assert (p / "executable").stat().st_mode == 0o100755
		assert (p / "executable").read_bytes() == TEST_EXECUTABLE
		assert (p / "link").is_symlink
		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 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 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 20 Lines • Show All 247 Lines • Show Last 20 Lines