Changeset View
Changeset View
Standalone 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 | |||||
) |