Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/svn/tests/test_loader.py
| Show All 13 Lines | |||||
| 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, | ||||
| SvnLoaderFromDumpArchive, | SvnLoaderFromDumpArchive, | ||||
| SvnLoaderFromRemoteDump, | SvnLoaderFromRemoteDump, | ||||
| ) | ) | ||||
| from swh.loader.svn.svn import SvnRepo | |||||
| from swh.loader.svn.utils import init_svn_repo_from_dump | from swh.loader.svn.utils import init_svn_repo_from_dump | ||||
| from swh.loader.tests import ( | from swh.loader.tests import ( | ||||
| assert_last_visit_matches, | assert_last_visit_matches, | ||||
| check_snapshot, | check_snapshot, | ||||
| get_stats, | get_stats, | ||||
| prepare_repository_from_archive, | prepare_repository_from_archive, | ||||
| ) | ) | ||||
| from swh.model.from_disk import DentryPerms | from swh.model.from_disk import DentryPerms | ||||
| ▲ Show 20 Lines • Show All 1,099 Lines • ▼ Show 20 Lines | def test_loader_invalid_svn_eol_style_property_value(swh_storage, tmp_path): | ||||
| assert loader.visit_status() == "full" | assert loader.visit_status() == "full" | ||||
| 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): | |||||
| class SvnRepoSkipFirstRevision(SvnRepo): | |||||
| def logs(self, revision_start, revision_end): | |||||
| """Overrides logs method to skip revision number one in yielded revisions""" | |||||
| yield from super().logs(revision_start + 1, revision_end) | |||||
| from swh.loader.svn import loader | |||||
| 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"): | |||||
| add_commit( | |||||
| repo_url, | |||||
| f"Add {filename} file", | |||||
| [ | |||||
| CommitChange( | |||||
| change_type=CommitChangeType.AddOrUpdate, | |||||
| path=filename, | |||||
| data=f"{filename}\n".encode(), | |||||
| ) | |||||
| ], | |||||
| ) | |||||
| loader = SvnLoader(swh_storage, repo_url, destination_path=tmp_path) | |||||
| assert loader.load() == {"status": "eventful"} | |||||
| assert loader.visit_status() == "full" | |||||
| assert get_stats(loader.storage) == { | |||||
| "content": 2, | |||||
| "directory": 2, | |||||
| "origin": 1, | |||||
| "origin_visit": 1, | |||||
| "release": 0, | |||||
| "revision": 2, | |||||
| "skipped_content": 0, | |||||
| "snapshot": 1, | |||||
| } | |||||