Differential D6820 Diff 24712 swh/loader/svn/tests/test_loader.py

Changeset View

Standalone View

swh/loader/svn/tests/test_loader.py

Show First 20 Lines • Show All 1,070 Lines • ▼ Show 20 Lines	for change in changes:
if "data" in change:		if "data" in change:
txdelta = file.apply_textdelta()		txdelta = file.apply_textdelta()
delta.send_stream(BytesIO(change["data"]), txdelta)		delta.send_stream(BytesIO(change["data"]), txdelta)
file.close()		file.close()
root.close()		root.close()
editor.close()		editor.close()


def test_loader_eol_style_file_property_handling_edge_case(swh_storage, tmp_path):		@pytest.fixture
		def repo_url(tmp_path):
# create a repository		# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")		repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)		repos.create(repo_path)
repo_url = f"file://{repo_path}"		repo_url = f"file://{repo_path}"
		return repo_url


		def test_loader_eol_style_file_property_handling_edge_case(
		swh_storage, repo_url, tmp_path
		):

# # first commit		# # first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
"Add a directory containing a file with CRLF end of line "		"Add a directory containing a file with CRLF end of line "
"and set svn:eol-style property to native so CRLF will be "		"and set svn:eol-style property to native so CRLF will be "
"replaced by LF in the file when exporting the revision"		"replaced by LF in the file when exporting the revision"
▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines	def get_head_revision_paths_info(loader: SvnLoader) -> Dict[bytes, Dict[str, Any]]:
assert revision is not None		assert revision is not None

paths = {}		paths = {}
for entry in loader.storage.directory_ls(revision.directory, recursive=True):		for entry in loader.storage.directory_ls(revision.directory, recursive=True):
paths[entry["name"]] = entry		paths[entry["name"]] = entry
return paths		return paths


def test_loader_eol_style_on_svn_link_handling(swh_storage, tmp_path):		def test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
"Add a regular file, a directory and a link to the regular file "		"Add a regular file, a directory and a link to the regular file "
"in the directory. Set svn:eol-style property for the regular "		"in the directory. Set svn:eol-style property for the regular "
"file and the link. Set svn:special property for the link."		"file and the link. Set svn:special property for the link."
▲ Show 20 Lines • Show All 47 Lines • ▼ Show 20 Lines	def test_loader_eol_style_on_svn_link_handling(swh_storage, repo_url, tmp_path):
assert (		assert (
loader.storage.content_get_data(		loader.storage.content_get_data(
paths[b"directory/file_with_crlf_eol.txt"]["sha1"]		paths[b"directory/file_with_crlf_eol.txt"]["sha1"]
)		)
== b"../file_with_crlf_eol.txt"		== b"../file_with_crlf_eol.txt"
)		)


def test_loader_svn_special_property_unset(swh_storage, tmp_path):		def test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
"Create a regular file, a link to a file and a link to an "		"Create a regular file, a link to a file and a link to an "
"external file. Set the svn:special property on the links."		"external file. Set the svn:special property on the links."
),		),
▲ Show 20 Lines • Show All 68 Lines • ▼ Show 20 Lines	def test_loader_svn_special_property_unset(swh_storage, repo_url, tmp_path):

assert paths[b"external_link.txt"]["perms"] == DentryPerms.content		assert paths[b"external_link.txt"]["perms"] == DentryPerms.content
assert (		assert (
loader.storage.content_get_data(paths[b"external_link.txt"]["sha1"])		loader.storage.content_get_data(paths[b"external_link.txt"]["sha1"])
== b"link /home/user/data.txt"		== b"link /home/user/data.txt"
)		)


def test_loader_invalid_svn_eol_style_property_value(swh_storage, tmp_path):		def test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

filename = "file_with_crlf_eol.txt"		filename = "file_with_crlf_eol.txt"
file_content = b"Hello world!\r\n"		file_content = b"Hello world!\r\n"

# # first commit		# # first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
Show All 23 Lines	def test_loader_invalid_svn_eol_style_property_value(swh_storage, repo_url, tmp_path):
paths = get_head_revision_paths_info(loader)		paths = get_head_revision_paths_info(loader)
# end of lines should not have been processed		# end of lines should not have been processed
assert (		assert (
loader.storage.content_get_data(paths[filename.encode()]["sha1"])		loader.storage.content_get_data(paths[filename.encode()]["sha1"])
== file_content		== file_content
)		)


def test_loader_first_revision_is_not_number_one(swh_storage, mocker, tmp_path):		def test_loader_first_revision_is_not_number_one(
		swh_storage, mocker, repo_url, tmp_path
		):
class SvnRepoSkipFirstRevision(SvnRepo):		class SvnRepoSkipFirstRevision(SvnRepo):
def logs(self, revision_start, revision_end):		def logs(self, revision_start, revision_end):
"""Overrides logs method to skip revision number one in yielded revisions"""		"""Overrides logs method to skip revision number one in yielded revisions"""
yield from super().logs(revision_start + 1, revision_end)		yield from super().logs(revision_start + 1, revision_end)

from swh.loader.svn import loader		from swh.loader.svn import loader

mocker.patch.object(loader, "SvnRepo", SvnRepoSkipFirstRevision)		mocker.patch.object(loader, "SvnRepo", SvnRepoSkipFirstRevision)
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

for filename in ("foo", "bar", "baz"):		for filename in ("foo", "bar", "baz"):
add_commit(		add_commit(
repo_url,		repo_url,
f"Add {filename} file",		f"Add {filename} file",
[		[
CommitChange(		CommitChange(
change_type=CommitChangeType.AddOrUpdate,		change_type=CommitChangeType.AddOrUpdate,
Show All 19 Lines	assert get_stats(loader.storage) == {
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 2,		"revision": 2,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 1,		"snapshot": 1,
}		}


def test_loader_svn_special_property_on_binary_file(swh_storage, tmp_path):		def test_loader_svn_special_property_on_binary_file(swh_storage, repo_url, tmp_path):
"""When a file has the svn:special property set but is not a svn link,		"""When a file has the svn:special property set but is not a svn link,
it might be truncated under certain conditions when performing an export		it might be truncated under certain conditions when performing an export
operation."""		operation."""

# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

data = (		data = (
b"!<symlink>\xff\xfea\x00p\x00t\x00-\x00c\x00y\x00g\x00.\x00s\x00h\x00\x00\x00"		b"!<symlink>\xff\xfea\x00p\x00t\x00-\x00c\x00y\x00g\x00.\x00s\x00h\x00\x00\x00"
)		)

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Lines	assert_last_visit_matches(
loader.storage,		loader.storage,
repo_url,		repo_url,
status="partial",		status="partial",
type="svn",		type="svn",
snapshot=GOURMET_SNAPSHOT.id,		snapshot=GOURMET_SNAPSHOT.id,
)		)


def test_loader_delete_directory_while_file_has_same_prefix(swh_storage, tmp_path):		def test_loader_delete_directory_while_file_has_same_prefix(
# create a repository		swh_storage, repo_url, tmp_path
repo_path = os.path.join(tmp_path, "tmprepo")		):
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
"Add a file and a directory with same prefix",		"Add a file and a directory with same prefix",
[		[
CommitChange(		CommitChange(
change_type=CommitChangeType.AddOrUpdate,		change_type=CommitChangeType.AddOrUpdate,
Show All 28 Lines	):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)		loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)

assert loader.load() == {"status": "eventful"}		assert loader.load() == {"status": "eventful"}
assert_last_visit_matches(		assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn",		loader.storage, repo_url, status="full", type="svn",
)		)


def test_svn_loader_incremental(swh_storage, tmp_path):		def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
"Add a directory containing a file with CRLF end of line "		"Add a directory containing a file with CRLF end of line "
"and set svn:eol-style property to native so CRLF will be "		"and set svn:eol-style property to native so CRLF will be "
"replaced by LF in the file when exporting the revision"		"replaced by LF in the file when exporting the revision"
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines	def test_svn_loader_incremental(swh_storage, repo_url, tmp_path):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)		loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)
assert loader.load() == {"status": "eventful"}		assert loader.load() == {"status": "eventful"}
assert_last_visit_matches(		assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn",		loader.storage, repo_url, status="full", type="svn",
)		)


def test_svn_loader_incremental_replay_start_with_empty_directory(		def test_svn_loader_incremental_replay_start_with_empty_directory(
swh_storage, mocker, tmp_path		swh_storage, mocker, repo_url, tmp_path
):		):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
("Add a file"),		("Add a file"),
[		[
CommitChange(		CommitChange(
change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",		change_type=CommitChangeType.AddOrUpdate, path="foo.txt", data=b"foo\n",
Show All 38 Lines	):
# second load, incremental		# second load, incremental
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)		loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
loader.load()		loader.load()

# check work directory was empty before replaying revisions		# check work directory was empty before replaying revisions
assert loader.svnrepo.replay_dir_content_before_start == []		assert loader.svnrepo.replay_dir_content_before_start == []


def test_loader_svn_executable_property_on_svn_link_handling(swh_storage, tmp_path):		def test_loader_svn_executable_property_on_svn_link_handling(
# create a repository		swh_storage, repo_url, tmp_path
repo_path = os.path.join(tmp_path, "tmprepo")		):
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
(		(
"Add an executable file and a svn link to it."		"Add an executable file and a svn link to it."
"Set svn:executable property for both paths."		"Set svn:executable property for both paths."
),		),
Show All 38 Lines	):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)		loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)

assert loader.load() == {"status": "eventful"}		assert loader.load() == {"status": "eventful"}
assert_last_visit_matches(		assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn",		loader.storage, repo_url, status="full", type="svn",
)		)


def test_loader_svn_add_property_on_link(swh_storage, tmp_path):		def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
"Add an executable file and a svn link to it.",		"Add an executable file and a svn link to it.",
[		[
CommitChange(		CommitChange(
change_type=CommitChangeType.AddOrUpdate,		change_type=CommitChangeType.AddOrUpdate,
Show All 29 Lines	def test_loader_svn_add_property_on_link(swh_storage, repo_url, tmp_path):
loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)		loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path, check_revision=1)

assert loader.load() == {"status": "eventful"}		assert loader.load() == {"status": "eventful"}
assert_last_visit_matches(		assert_last_visit_matches(
loader.storage, repo_url, status="full", type="svn",		loader.storage, repo_url, status="full", type="svn",
)		)


def test_loader_svn_link_parsing(swh_storage, tmp_path):		def test_loader_svn_link_parsing(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
"Add an executable file and a svn link to it.",		"Add an executable file and a svn link to it.",
[		[
CommitChange(		CommitChange(
change_type=CommitChangeType.AddOrUpdate,		change_type=CommitChangeType.AddOrUpdate,
▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines	assert_last_visit_matches(
loader.storage,		loader.storage,
repo_url,		repo_url,
status="full",		status="full",
type="svn",		type="svn",
snapshot=GOURMET_SNAPSHOT.id,		snapshot=GOURMET_SNAPSHOT.id,
)		)


def test_loader_svn_add_property_on_directory_link(swh_storage, tmp_path):		def test_loader_svn_add_property_on_directory_link(swh_storage, repo_url, tmp_path):
# create a repository
repo_path = os.path.join(tmp_path, "tmprepo")
repos.create(repo_path)
repo_url = f"file://{repo_path}"

# first commit		# first commit
add_commit(		add_commit(
repo_url,		repo_url,
"Add an executable file in a directory and a svn link to the directory.",		"Add an executable file in a directory and a svn link to the directory.",
[		[
CommitChange(		CommitChange(
change_type=CommitChangeType.AddOrUpdate,		change_type=CommitChangeType.AddOrUpdate,
Show All 35 Lines