Page MenuHomeSoftware Heritage
Differential D6632 Diff 24102 swh/loader/svn/tests/test_loader.py

Changeset View

Standalone View

View Options

swh/loader/svn/tests/test_loader.py

Show First 20 Lines • Show All 198 Lines • ▼ Show 20 Lines
def test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path): def test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
"""Visit multiple times a repository with no change should yield the same snapshot """Visit multiple times a repository with no change should yield the same snapshot
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = 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,
Show All 16 Linesdef test_loader_svn_2_visits_no_change(swh_storage, datadir, tmp_path):
assert stats["snapshot"] == 1 assert stats["snapshot"] == 1
# even starting from previous revision... # even starting from previous revision...
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin_visit"] == 2 + 1 assert stats["origin_visit"] == 2 + 1
# ... with no change in repository, this yields the same snapshot # ... with no change in repository, this yields the same snapshot
assert stats["snapshot"] == 1 assert stats["snapshot"] == 1
assert_last_visit_matches( assert_last_visit_matches(
Show All 21 Lines """In this scenario, the dump has been tampered with to modify the
svnadmin setlog . -r 6 log.txt --bypass-hooks svnadmin setlog . -r 6 log.txt --bypass-hooks
tar cvf pkg-gourmet-tampered-rev6-log.tgz pkg-gourmet/ tar cvf pkg-gourmet-tampered-rev6-log.tgz pkg-gourmet/
``` ```
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
check_snapshot(GOURMET_SNAPSHOT, loader.storage) check_snapshot(GOURMET_SNAPSHOT, loader.storage)
assert_last_visit_matches( 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,
) )
archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz") archive_path2 = os.path.join(datadir, "pkg-gourmet-tampered-rev6-log.tgz")
repo_tampered_url = prepare_repository_from_archive( repo_tampered_url = prepare_repository_from_archive(
archive_path2, archive_name, tmp_path archive_path2, archive_name, tmp_path
) )
loader2 = SvnLoader(swh_storage, repo_tampered_url, origin_url=repo_url) loader2 = SvnLoader(
swh_storage, repo_tampered_url, origin_url=repo_url, temp_directory=tmp_path
)
assert loader2.load() == {"status": "eventful"} assert loader2.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader2.storage, loader2.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=hash_to_bytes("c499eebc1e201024d47d24053ac0080049305897"), snapshot=hash_to_bytes("c499eebc1e201024d47d24053ac0080049305897"),
Show All 13 Linesdef test_loader_svn_visit_with_changes(swh_storage, datadir, tmp_path):
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_initial_url = prepare_repository_from_archive( repo_initial_url = prepare_repository_from_archive(
archive_path, archive_name, tmp_path archive_path, archive_name, tmp_path
) )
# repo_initial_url becomes the origin_url we want to visit some more below # repo_initial_url becomes the origin_url we want to visit some more below
loader = SvnLoader(swh_storage, repo_initial_url) loader = SvnLoader(swh_storage, repo_initial_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
loader = SvnLoader(swh_storage, repo_updated_url, origin_url=repo_initial_url,) loader = SvnLoader(
swh_storage,
repo_updated_url,
origin_url=repo_initial_url,
temp_directory=tmp_path,
)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_updated_url, repo_updated_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_UPDATES_SNAPSHOT.id, snapshot=GOURMET_UPDATES_SNAPSHOT.id,
Show All 13 Lines assert stats == {
"skipped_content": 0, "skipped_content": 0,
"snapshot": 2, "snapshot": 2,
} }
check_snapshot(GOURMET_UPDATES_SNAPSHOT, loader.storage) check_snapshot(GOURMET_UPDATES_SNAPSHOT, loader.storage)
# Let's start the ingestion from the start, this should yield the same result # Let's start the ingestion from the start, this should yield the same result
loader = SvnLoader( loader = SvnLoader(
swh_storage, repo_updated_url, origin_url=repo_initial_url, incremental=False, swh_storage,
repo_updated_url,
origin_url=repo_initial_url,
incremental=False,
temp_directory=tmp_path,
) )
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status3 = assert_last_visit_matches( visit_status3 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_updated_url, repo_updated_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_UPDATES_SNAPSHOT.id, snapshot=GOURMET_UPDATES_SNAPSHOT.id,
Show All 15 Linesdef test_loader_svn_visit_start_from_revision(swh_storage, datadir, tmp_path):
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_initial_url = prepare_repository_from_archive( repo_initial_url = prepare_repository_from_archive(
archive_path, archive_name, tmp_path archive_path, archive_name, tmp_path
) )
# repo_initial_url becomes the origin_url we want to visit some more below # repo_initial_url becomes the origin_url we want to visit some more below
loader = SvnLoader(swh_storage, repo_initial_url) loader = SvnLoader(swh_storage, repo_initial_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
visit_status1 = assert_last_visit_matches( visit_status1 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_initial_url, repo_initial_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
start_revision = loader.storage.revision_get( start_revision = loader.storage.revision_get(
[hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")] [hash_to_bytes("95edacc8848369d6fb1608e887d6d2474fd5224f")]
)[0] )[0]
assert start_revision is not None assert start_revision is not None
archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-updates.tgz")
repo_updated_url = prepare_repository_from_archive( repo_updated_url = prepare_repository_from_archive(
archive_path, "pkg-gourmet", tmp_path archive_path, "pkg-gourmet", tmp_path
) )
# we'll start from start_revision # we'll start from start_revision
loader = SvnLoader(swh_storage, repo_updated_url, origin_url=repo_initial_url,) loader = SvnLoader(
swh_storage,
repo_updated_url,
origin_url=repo_initial_url,
temp_directory=tmp_path,
)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
# nonetheless, we obtain the same snapshot (as previous tests on that repository) # nonetheless, we obtain the same snapshot (as previous tests on that repository)
visit_status2 = assert_last_visit_matches( visit_status2 = assert_last_visit_matches(
loader.storage, loader.storage,
repo_updated_url, repo_updated_url,
status="full", status="full",
Show All 25 Lines """Check that a svn repo containing a versioned file with CRLF line
violation of svn specification as the file should have been violation of svn specification as the file should have been
stored with LF line endings) can be loaded anyway. stored with LF line endings) can be loaded anyway.
""" """
archive_name = "mediawiki-repo-r407-eol-native-crlf" archive_name = "mediawiki-repo-r407-eol-native-crlf"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
mediawiki_snapshot = Snapshot( mediawiki_snapshot = Snapshot(
id=hash_to_bytes("d6d6e9703f157c5702d9a4a5dec878926ed4ab76"), id=hash_to_bytes("d6d6e9703f157c5702d9a4a5dec878926ed4ab76"),
branches={ branches={
b"HEAD": SnapshotBranch( b"HEAD": SnapshotBranch(
target=hash_to_bytes("7da4975c363101b819756d33459f30a866d01b1b"), target=hash_to_bytes("7da4975c363101b819756d33459f30a866d01b1b"),
target_type=TargetType.REVISION, target_type=TargetType.REVISION,
Show All 23 Linesdef test_loader_svn_visit_with_mixed_crlf_lf(swh_storage, datadir, tmp_path):
for textual content should not be stored when the svn:eol-style for textual content should not be stored when the svn:eol-style
property is set) can be loaded anyway. property is set) can be loaded anyway.
""" """
archive_name = "pyang-repo-r343-eol-native-mixed-lf-crlf" archive_name = "pyang-repo-r343-eol-native-mixed-lf-crlf"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
pyang_snapshot = Snapshot( pyang_snapshot = Snapshot(
id=hash_to_bytes("6d9590de11b00a5801de0ff3297c5b44bbbf7d24"), id=hash_to_bytes("6d9590de11b00a5801de0ff3297c5b44bbbf7d24"),
branches={ branches={
b"HEAD": SnapshotBranch( b"HEAD": SnapshotBranch(
target=hash_to_bytes("9c6962eeb9164a636c374be700672355e34a98a7"), target=hash_to_bytes("9c6962eeb9164a636c374be700672355e34a98a7"),
target_type=TargetType.REVISION, target_type=TargetType.REVISION,
Show All 15 Lines
def test_loader_svn_with_external_properties(swh_storage, datadir, tmp_path): def test_loader_svn_with_external_properties(swh_storage, datadir, tmp_path):
"""Repository with svn:external properties cannot be fully ingested yet """Repository with svn:external properties cannot be fully ingested yet
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, "pkg-gourmet-with-external-id.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-external-id.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
gourmet_externals_snapshot = Snapshot( gourmet_externals_snapshot = Snapshot(
id=hash_to_bytes("19cb68d0a3f22372e2b7017ea5e2a2ea5ae3e09a"), id=hash_to_bytes("19cb68d0a3f22372e2b7017ea5e2a2ea5ae3e09a"),
branches={ branches={
b"HEAD": SnapshotBranch( b"HEAD": SnapshotBranch(
target=hash_to_bytes("82a7a4a09f9549223429143ba36ad77375e33c5c"), target=hash_to_bytes("82a7a4a09f9549223429143ba36ad77375e33c5c"),
target_type=TargetType.REVISION, target_type=TargetType.REVISION,
Show All 29 Linesdef test_loader_svn_with_symlink(swh_storage, datadir, tmp_path):
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join( archive_path = os.path.join(
datadir, "pkg-gourmet-with-edge-case-links-and-files.tgz" datadir, "pkg-gourmet-with-edge-case-links-and-files.tgz"
) )
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
gourmet_edge_cases_snapshot = Snapshot( gourmet_edge_cases_snapshot = Snapshot(
id=hash_to_bytes("18e60982fe521a2546ab8c3c73a535d80462d9d0"), id=hash_to_bytes("18e60982fe521a2546ab8c3c73a535d80462d9d0"),
branches={ branches={
b"HEAD": SnapshotBranch( b"HEAD": SnapshotBranch(
target=hash_to_bytes("3f43af2578fccf18b0d4198e48563da7929dc608"), target=hash_to_bytes("3f43af2578fccf18b0d4198e48563da7929dc608"),
target_type=TargetType.REVISION, target_type=TargetType.REVISION,
Show All 24 Lines Edge case:
- wrong symbolic link - wrong symbolic link
- wrong symbolic link with empty space names - wrong symbolic link with empty space names
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, "pkg-gourmet-with-wrong-link-cases.tgz") archive_path = os.path.join(datadir, "pkg-gourmet-with-wrong-link-cases.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loader = SvnLoader(swh_storage, repo_url) loader = SvnLoader(swh_storage, repo_url, temp_directory=tmp_path)
assert loader.load() == {"status": "eventful"} assert loader.load() == {"status": "eventful"}
gourmet_wrong_links_snapshot = Snapshot( gourmet_wrong_links_snapshot = Snapshot(
id=hash_to_bytes("b17f38acabb90f066dedd30c29f01a02af88a5c4"), id=hash_to_bytes("b17f38acabb90f066dedd30c29f01a02af88a5c4"),
branches={ branches={
b"HEAD": SnapshotBranch( b"HEAD": SnapshotBranch(
target=hash_to_bytes("cf30d3bb9d5967d0a2bbeacc405f10a5dd9b138a"), target=hash_to_bytes("cf30d3bb9d5967d0a2bbeacc405f10a5dd9b138a"),
target_type=TargetType.REVISION, target_type=TargetType.REVISION,
Show All 24 Lines Edge case:
- wrong symbolic link - wrong symbolic link
- wrong symbolic link with empty space names - wrong symbolic link with empty space names
""" """
archive_name = "pkg-gourmet" archive_name = "pkg-gourmet"
archive_path = os.path.join(datadir, f"{archive_name}.tgz") archive_path = os.path.join(datadir, f"{archive_name}.tgz")
repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path) repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
loaderFromDump = SvnLoaderFromRemoteDump(swh_storage, repo_url) loaderFromDump = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path
)
assert loaderFromDump.load() == {"status": "eventful"} assert loaderFromDump.load() == {"status": "eventful"}
assert_last_visit_matches( assert_last_visit_matches(
loaderFromDump.storage, loaderFromDump.storage,
repo_url, repo_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
origin_url = repo_url + "2" # rename to another origin origin_url = repo_url + "2" # rename to another origin
loader = SvnLoader(swh_storage, repo_url, origin_url=origin_url) loader = SvnLoader(
swh_storage, repo_url, origin_url=origin_url, temp_directory=tmp_path
)
assert loader.load() == {"status": "eventful"} # because are working on new origin assert loader.load() == {"status": "eventful"} # because are working on new origin
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
origin_url, origin_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
check_snapshot(GOURMET_SNAPSHOT, loader.storage) check_snapshot(GOURMET_SNAPSHOT, loader.storage)
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 2 # created one more origin assert stats["origin"] == 2 # created one more origin
assert stats["origin_visit"] == 2 assert stats["origin_visit"] == 2
assert stats["snapshot"] == 1 assert stats["snapshot"] == 1
loader = SvnLoader(swh_storage, repo_url) # no change on the origin-url loader = SvnLoader(
swh_storage, repo_url, temp_directory=tmp_path
) # no change on the origin-url
assert loader.load() == {"status": "uneventful"} assert loader.load() == {"status": "uneventful"}
assert_last_visit_matches( assert_last_visit_matches(
loader.storage, loader.storage,
origin_url, origin_url,
status="full", status="full",
type="svn", type="svn",
snapshot=GOURMET_SNAPSHOT.id, snapshot=GOURMET_SNAPSHOT.id,
) )
stats = get_stats(loader.storage) stats = get_stats(loader.storage)
assert stats["origin"] == 2 assert stats["origin"] == 2
assert stats["origin_visit"] == 3 assert stats["origin_visit"] == 3
assert stats["snapshot"] == 1 assert stats["snapshot"] == 1
# second visit from the dump should be uneventful # second visit from the dump should be uneventful
loaderFromDump = SvnLoaderFromRemoteDump(swh_storage, repo_url) loaderFromDump = SvnLoaderFromRemoteDump(
swh_storage, repo_url, temp_directory=tmp_path
)
assert loaderFromDump.load() == {"status": "uneventful"} assert loaderFromDump.load() == {"status": "uneventful"}
def test_loader_user_defined_svn_properties(swh_storage, datadir, tmp_path): def test_loader_user_defined_svn_properties(swh_storage, datadir, tmp_path):
"""Edge cases: The repository held some user defined svn-properties with special """Edge cases: The repository held some user defined svn-properties with special
encodings, this prevented the repository from being loaded even though we do not encodings, this prevented the repository from being loaded even though we do not
ingest those information. ingest those information.
▲ Show 20 Lines • Show All 546 Lines • Show Last 20 Lines
About · Developer information · Legal