Differential D6604 Diff 23995 swh/loader/svn/tests/test_loader.py

Changeset View

Standalone View

swh/loader/svn/tests/test_loader.py

# Copyright (C) 2016-2021 The Software Heritage developers		# Copyright (C) 2016-2021 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

from enum import Enum		from enum import Enum
from io import BytesIO		from io import BytesIO
import os		import os
import subprocess		import subprocess
from typing import Dict, List		from typing import Any, Dict, List

import pytest		import pytest
from subvertpy import SubversionException, delta, repos		from subvertpy import SubversionException, delta, repos
from subvertpy.ra import Auth, RemoteAccess, get_username_provider		from subvertpy.ra import Auth, RemoteAccess, get_username_provider
from typing_extensions import TypedDict		from typing_extensions import TypedDict

from swh.loader.svn.loader import (		from swh.loader.svn.loader import (
SvnLoader,		SvnLoader,
▲ Show 20 Lines • Show All 887 Lines • ▼ Show 20 Lines	assert get_stats(loader.storage) == {
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 3,		"revision": 3,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 1,		"snapshot": 1,
}		}


		def get_head_revision_paths_info(loader: SvnLoader) -> Dict[bytes, Dict[str, Any]]:
		assert loader.snapshot is not None
		root_dir = loader.snapshot.branches[b"HEAD"].target
		revision = loader.storage.revision_get([root_dir])[0]
		assert revision is not None

		paths = {}
		for entry in loader.storage.directory_ls(revision.directory, recursive=True):
		paths[entry["name"]] = entry
		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, 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}"

# first commit		# first commit
add_commit(		add_commit(
Show All 36 Lines	assert get_stats(loader.storage) == {
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 1,		"revision": 1,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 1,		"snapshot": 1,
}		}

root_dir = loader.snapshot.branches[b"HEAD"].target		paths = get_head_revision_paths_info(loader)
revision = loader.storage.revision_get([root_dir])[0]

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

assert (		assert (
loader.storage.content_get_data(paths[b"file_with_crlf_eol.txt"]["sha1"])		loader.storage.content_get_data(paths[b"file_with_crlf_eol.txt"]["sha1"])
== b"Hello world!\n"		== b"Hello world!\n"
)		)

assert paths[b"directory/file_with_crlf_eol.txt"]["perms"] == DentryPerms.symlink		assert paths[b"directory/file_with_crlf_eol.txt"]["perms"] == DentryPerms.symlink
assert (		assert (
▲ Show 20 Lines • Show All 73 Lines • ▼ Show 20 Lines	assert get_stats(loader.storage) == {
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 2,		"revision": 2,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 1,		"snapshot": 1,
}		}

root_dir = loader.snapshot.branches[b"HEAD"].target		paths = get_head_revision_paths_info(loader)
revision = loader.storage.revision_get([root_dir])[0]

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

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

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):
		# 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"
		file_content = b"Hello world!\r\n"

		# # first commit
		add_commit(
		repo_url,
		(
		"Add a file with CRLF end of line and set svn:eol-style property "
		"to an invalid value."
		),
		[
		CommitChange(
		change_type=CommitChangeType.AddOrUpdate,
		path=filename,
		properties={"svn:eol-style": "foo"},
		data=file_content,
		)
		],
		)

		# instantiate a svn loader checking after each processed revision that
		# the repository filesystem it reconstructed does not differ from a subversion
		# export of that revision
		loader = SvnLoader(
		swh_storage, repo_url, destination_path=tmp_path, check_revision=1
		)

		assert loader.load() == {"status": "eventful"}
		assert loader.visit_status() == "full"

		paths = get_head_revision_paths_info(loader)
		# end of lines should not have been processed
		assert (
		loader.storage.content_get_data(paths[filename.encode()]["sha1"])
		== file_content
		)