Differential D3773 Diff 13367 swh/storage/tests/test_replay.py

Changeset View

Standalone View

swh/storage/tests/test_replay.py

Show First 20 Lines • Show All 200 Lines • ▼ Show 20 Lines	):
"""Simple utility function to compare the content of 2 in_memory storages		"""Simple utility function to compare the content of 2 in_memory storages

"""		"""
expected_persons = set(src._persons.values())		expected_persons = set(src._persons.values())
got_persons = set(dst._persons.values())		got_persons = set(dst._persons.values())
assert got_persons == expected_persons		assert got_persons == expected_persons

for attr_ in (		for attr_ in (
"revisions",
"releases",		"releases",
"snapshots",		"snapshots",
"origins",		"origins",
"origin_visits",		"origin_visits",
"origin_visit_statuses",		"origin_visit_statuses",
):		):
if exclude and attr_ in exclude:		if exclude and attr_ in exclude:
continue		continue
expected_objects = sorted(getattr(src, f"_{attr_}").items())		expected_objects = sorted(getattr(src, f"_{attr_}").items())
got_objects = sorted(getattr(dst, f"_{attr_}").items())		got_objects = sorted(getattr(dst, f"_{attr_}").items())
assert got_objects == expected_objects, f"Mismatch object list for {attr_}"		assert got_objects == expected_objects, f"Mismatch object list for {attr_}"

for attr_ in (		for attr_ in (
"contents",		"contents",
"skipped_contents",		"skipped_contents",
"directories",		"directories",
		"revisions",
):		):
if exclude and attr_ in exclude:		if exclude and attr_ in exclude:
continue		continue
expected_objects = [		expected_objects = [
(id, nullify_ctime(obj))		(id, nullify_ctime(obj))
for id, obj in sorted(getattr(src._cql_runner, f"_{attr_}").iter_all())		for id, obj in sorted(getattr(src._cql_runner, f"_{attr_}").iter_all())
]		]
got_objects = [		got_objects = [
▲ Show 20 Lines • Show All 141 Lines • ▼ Show 20 Lines	def check_replayed(src, dst, expected_anonymized=False):

expected_persons = {		expected_persons = {
maybe_anonymize("persons", person) for person in src._persons.values()		maybe_anonymize("persons", person) for person in src._persons.values()
}		}
got_persons = set(dst._persons.values())		got_persons = set(dst._persons.values())
assert got_persons == expected_persons		assert got_persons == expected_persons

for attr_ in (		for attr_ in (
"revisions",
"releases",		"releases",
"snapshots",		"snapshots",
"origins",		"origins",
"origin_visit_statuses",		"origin_visit_statuses",
):		):
expected_objects = [		expected_objects = [
(id, maybe_anonymize(attr_, obj))		(id, maybe_anonymize(attr_, obj))
for id, obj in sorted(getattr(src, f"_{attr_}").items())		for id, obj in sorted(getattr(src, f"_{attr_}").items())
]		]
got_objects = [		got_objects = [
(id, obj) for id, obj in sorted(getattr(dst, f"_{attr_}").items())		(id, obj) for id, obj in sorted(getattr(dst, f"_{attr_}").items())
]		]
assert got_objects == expected_objects, f"Mismatch object list for {attr_}"		assert got_objects == expected_objects, f"Mismatch object list for {attr_}"

for attr_ in (		for attr_ in (
"contents",		"contents",
"skipped_contents",		"skipped_contents",
"directories",		"directories",
		"revisions",
):		):
expected_objects = [		expected_objects = [
(id, nullify_ctime(maybe_anonymize(attr_, obj)))		(id, nullify_ctime(maybe_anonymize(attr_, obj)))
for id, obj in sorted(getattr(src._cql_runner, f"_{attr_}").iter_all())		for id, obj in sorted(getattr(src._cql_runner, f"_{attr_}").iter_all())
]		]
got_objects = [		got_objects = [
(id, nullify_ctime(obj))		(id, nullify_ctime(obj))
for id, obj in sorted(getattr(dst._cql_runner, f"_{attr_}").iter_all())		for id, obj in sorted(getattr(dst._cql_runner, f"_{attr_}").iter_all())
]		]
assert got_objects == expected_objects, f"Mismatch object list for {attr_}"		assert got_objects == expected_objects, f"Mismatch object list for {attr_}"