Differential D6758 Diff 24543 swh/loader/cvs/tests/test_loader.py

Changeset View

Standalone View

swh/loader/cvs/tests/test_loader.py

Show First 20 Lines • Show All 145 Lines • ▼ Show 20 Lines	def test_loader_cvs_with_file_additions_and_deletions(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 8,		"content": 8,
"directory": 20,		"directory": 20,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 7,		"revision": 7,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 7,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT, loader.storage)		check_snapshot(GREEK_SNAPSHOT, loader.storage)


def test_loader_cvs_pserver_with_file_additions_and_deletions(		def test_loader_cvs_pserver_with_file_additions_and_deletions(
swh_storage, datadir, tmp_path		swh_storage, datadir, tmp_path
):		):
Show All 20 Lines	):
assert stats == {		assert stats == {
"content": 8,		"content": 8,
"directory": 20,		"directory": 20,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 7,		"revision": 7,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 7,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT, loader.storage)		check_snapshot(GREEK_SNAPSHOT, loader.storage)


GREEK_SNAPSHOT2 = Snapshot(		GREEK_SNAPSHOT2 = Snapshot(
id=hash_to_bytes("048885ae2145ffe81588aea95dcf75c536ecdf26"),		id=hash_to_bytes("048885ae2145ffe81588aea95dcf75c536ecdf26"),
branches={		branches={
Show All 25 Lines	def test_loader_cvs_2_visits_with_change(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 8,		"content": 8,
"directory": 20,		"directory": 20,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 7,		"revision": 7,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 7,		"snapshot": 1,
}		}

archive_name2 = "greek-repository2"		archive_name2 = "greek-repository2"
archive_path2 = os.path.join(datadir, f"{archive_name2}.tgz")		archive_path2 = os.path.join(datadir, f"{archive_name2}.tgz")
repo_url = prepare_repository_from_archive(archive_path2, archive_name, tmp_path)		repo_url = prepare_repository_from_archive(archive_path2, archive_name, tmp_path)
repo_url += "/greek-tree" # CVS module name		repo_url += "/greek-tree" # CVS module name

loader = CvsLoader(		loader = CvsLoader(
Show All 14 Lines	def test_loader_cvs_2_visits_with_change(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 10,		"content": 10,
"directory": 23,		"directory": 23,
"origin": 1,		"origin": 1,
"origin_visit": 2,		"origin_visit": 2,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 2,
}		}

check_snapshot(GREEK_SNAPSHOT2, loader.storage)		check_snapshot(GREEK_SNAPSHOT2, loader.storage)

assert visit_status1.date < visit_status2.date		assert visit_status1.date < visit_status2.date
assert visit_status1.snapshot != visit_status2.snapshot		assert visit_status1.snapshot != visit_status2.snapshot


▲ Show 20 Lines • Show All 76 Lines • ▼ Show 20 Lines	def test_loader_cvs_visit_pserver_no_eol(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 9,		"content": 9,
"directory": 23,		"directory": 23,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT3, loader.storage)		check_snapshot(GREEK_SNAPSHOT3, loader.storage)


GREEK_SNAPSHOT4 = Snapshot(		GREEK_SNAPSHOT4 = Snapshot(
id=hash_to_bytes("26e943053ea9c5f961336a72328cac22026ed3b5"),		id=hash_to_bytes("26e943053ea9c5f961336a72328cac22026ed3b5"),
branches={		branches={
Show All 31 Lines	def test_loader_cvs_visit_expand_id_keyword(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 12,		"content": 12,
"directory": 31,		"directory": 31,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 11,		"revision": 11,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 11,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT4, loader.storage)		check_snapshot(GREEK_SNAPSHOT4, loader.storage)


def test_loader_cvs_visit_pserver_expand_id_keyword(swh_storage, datadir, tmp_path):		def test_loader_cvs_visit_pserver_expand_id_keyword(swh_storage, datadir, tmp_path):
"""Visit to CVS pserver with file with an RCS Id keyword"""		"""Visit to CVS pserver with file with an RCS Id keyword"""
archive_name = "greek-repository4"		archive_name = "greek-repository4"
Show All 23 Lines	def test_loader_cvs_visit_pserver_expand_id_keyword(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 12,		"content": 12,
"directory": 31,		"directory": 31,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 11,		"revision": 11,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 11,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT4, loader.storage)		check_snapshot(GREEK_SNAPSHOT4, loader.storage)


GREEK_SNAPSHOT5 = Snapshot(		GREEK_SNAPSHOT5 = Snapshot(
id=hash_to_bytes("ee6faeaf50aa513c53c8ba29194116a5ef88add6"),		id=hash_to_bytes("ee6faeaf50aa513c53c8ba29194116a5ef88add6"),
branches={		branches={
Show All 31 Lines	def test_loader_cvs_with_file_deleted_and_readded(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 9,		"content": 9,
"directory": 22,		"directory": 22,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT5, loader.storage)		check_snapshot(GREEK_SNAPSHOT5, loader.storage)


def test_loader_cvs_pserver_with_file_deleted_and_readded(		def test_loader_cvs_pserver_with_file_deleted_and_readded(
swh_storage, datadir, tmp_path		swh_storage, datadir, tmp_path
):		):
Show All 25 Lines	):
assert stats == {		assert stats == {
"content": 9,		"content": 9,
"directory": 22,		"directory": 22,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 1,
}		}

check_snapshot(GREEK_SNAPSHOT5, loader.storage)		check_snapshot(GREEK_SNAPSHOT5, loader.storage)


DINO_SNAPSHOT = Snapshot(		DINO_SNAPSHOT = Snapshot(
id=hash_to_bytes("417021c16e17c5e0038cf0e73dbf48a6142c8304"),		id=hash_to_bytes("417021c16e17c5e0038cf0e73dbf48a6142c8304"),
branches={		branches={
Show All 31 Lines	def test_loader_cvs_readded_file_in_attic(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 38,		"content": 38,
"directory": 105,		"directory": 105,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 35,		"revision": 35,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 35,		"snapshot": 1,
}		}

check_snapshot(DINO_SNAPSHOT, loader.storage)		check_snapshot(DINO_SNAPSHOT, loader.storage)


def test_loader_cvs_pserver_readded_file_in_attic(swh_storage, datadir, tmp_path):		def test_loader_cvs_pserver_readded_file_in_attic(swh_storage, datadir, tmp_path):
"""Conversion over pserver with RCS files in the Attic"""		"""Conversion over pserver with RCS files in the Attic"""
# This repository has some file revisions marked "dead" in the Attic only.		# This repository has some file revisions marked "dead" in the Attic only.
Show All 25 Lines	def test_loader_cvs_pserver_readded_file_in_attic(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 38,		"content": 38,
"directory": 105,		"directory": 105,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 35,		"revision": 35,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 35,		"snapshot": 1,
}		}

check_snapshot(DINO_SNAPSHOT, loader.storage)		check_snapshot(DINO_SNAPSHOT, loader.storage)


DINO_SNAPSHOT2 = Snapshot(		DINO_SNAPSHOT2 = Snapshot(
id=hash_to_bytes("a9d6ce0b4f22dc4fd752ad4c25ec9ea71ed568d7"),		id=hash_to_bytes("a9d6ce0b4f22dc4fd752ad4c25ec9ea71ed568d7"),
branches={		branches={
Show All 31 Lines	def test_loader_cvs_split_commits_by_commitid(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 18,		"content": 18,
"directory": 36,		"directory": 36,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 18,		"revision": 18,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 18,		"snapshot": 1,
}		}


def test_loader_cvs_pserver_split_commits_by_commitid(swh_storage, datadir, tmp_path):		def test_loader_cvs_pserver_split_commits_by_commitid(swh_storage, datadir, tmp_path):
"""Conversion via pserver which needs to be split by commit ID"""		"""Conversion via pserver which needs to be split by commit ID"""
# This repository has some file revisions which use the same log message		# This repository has some file revisions which use the same log message
# and can only be told apart by commit IDs. Without commit IDs, these commits		# and can only be told apart by commit IDs. Without commit IDs, these commits
# would get merged into a single commit in our conversion result.		# would get merged into a single commit in our conversion result.
Show All 21 Lines	def test_loader_cvs_pserver_split_commits_by_commitid(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 18,		"content": 18,
"directory": 36,		"directory": 36,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 18,		"revision": 18,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 18,		"snapshot": 1,
}		}


GREEK_SNAPSHOT6 = Snapshot(		GREEK_SNAPSHOT6 = Snapshot(
id=hash_to_bytes("b4c9423b2711c181251deb458d4ab4a3172948ac"),		id=hash_to_bytes("b4c9423b2711c181251deb458d4ab4a3172948ac"),
branches={		branches={
b"HEAD": SnapshotBranch(		b"HEAD": SnapshotBranch(
target=hash_to_bytes("f317c720e1929fec0afce10e6a8cfd24ef76dfc7"),		target=hash_to_bytes("f317c720e1929fec0afce10e6a8cfd24ef76dfc7"),
Show All 31 Lines	def test_loader_cvs_empty_lines_in_log_message(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 9,		"content": 9,
"directory": 22,		"directory": 22,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 1,
}		}


def test_loader_cvs_pserver_empty_lines_in_log_message(swh_storage, datadir, tmp_path):		def test_loader_cvs_pserver_empty_lines_in_log_message(swh_storage, datadir, tmp_path):
"""Conversion via pserver with empty lines in a log message"""		"""Conversion via pserver with empty lines in a log message"""
archive_name = "greek-repository6"		archive_name = "greek-repository6"
extracted_name = "greek-repository"		extracted_name = "greek-repository"
archive_path = os.path.join(datadir, f"{archive_name}.tgz")		archive_path = os.path.join(datadir, f"{archive_name}.tgz")
Show All 23 Lines	def test_loader_cvs_pserver_empty_lines_in_log_message(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 9,		"content": 9,
"directory": 22,		"directory": 22,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 1,
}		}


def get_head_revision_paths_info(loader: CvsLoader) -> Dict[bytes, Dict[str, Any]]:		def get_head_revision_paths_info(loader: CvsLoader) -> Dict[bytes, Dict[str, Any]]:
assert loader.snapshot is not None		assert loader.snapshot is not None
root_dir = loader.snapshot.branches[b"HEAD"].target		root_dir = loader.snapshot.branches[b"HEAD"].target
revision = loader.storage.revision_get([root_dir])[0]		revision = loader.storage.revision_get([root_dir])[0]
assert revision is not None		assert revision is not None
Show All 30 Lines	def test_loader_cvs_with_header_keyword(swh_storage, datadir, tmp_path):
expected_stats = {		expected_stats = {
"content": 9,		"content": 9,
"directory": 22,		"directory": 22,
"origin": 2,		"origin": 2,
"origin_visit": 2,		"origin_visit": 2,
"release": 0,		"release": 0,
"revision": 8,		"revision": 8,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 8,		"snapshot": 1,
}		}
stats = get_stats(loader.storage)		stats = get_stats(loader.storage)
assert stats == expected_stats		assert stats == expected_stats
stats = get_stats(loader2.storage)		stats = get_stats(loader2.storage)
assert stats == expected_stats		assert stats == expected_stats

# Ensure that file 'alpha', which contains a $Header$ keyword,		# Ensure that file 'alpha', which contains a $Header$ keyword,
# was imported with equal content via file:// and fake:// URLs.		# was imported with equal content via file:// and fake:// URLs.
▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines	def test_loader_cvs_expand_log_keyword(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 14,		"content": 14,
"directory": 31,		"directory": 31,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 11,		"revision": 11,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 11,		"snapshot": 1,
}		}


def test_loader_cvs_pserver_expand_log_keyword(swh_storage, datadir, tmp_path):		def test_loader_cvs_pserver_expand_log_keyword(swh_storage, datadir, tmp_path):
"""Conversion of RCS history with Log keyword in files"""		"""Conversion of RCS history with Log keyword in files"""
archive_name = "greek-repository8"		archive_name = "greek-repository8"
extracted_name = "greek-repository"		extracted_name = "greek-repository"
archive_path = os.path.join(datadir, f"{archive_name}.tgz")		archive_path = os.path.join(datadir, f"{archive_name}.tgz")
Show All 23 Lines	def test_loader_cvs_pserver_expand_log_keyword(swh_storage, datadir, tmp_path):
assert stats == {		assert stats == {
"content": 14,		"content": 14,
"directory": 31,		"directory": 31,
"origin": 1,		"origin": 1,
"origin_visit": 1,		"origin_visit": 1,
"release": 0,		"release": 0,
"revision": 11,		"revision": 11,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 11,		"snapshot": 1,
}		}