Changeset View
Changeset View
Standalone 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 ( | ||||
"skipped_contents", | |||||
"directories", | "directories", | ||||
"revisions", | "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 ("contents",): | for attr_ in ( | ||||
"contents", | |||||
"skipped_contents", | |||||
): | |||||
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 = [ | ||||
(id, nullify_ctime(obj)) | (id, nullify_ctime(obj)) | ||||
▲ Show 20 Lines • Show All 140 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 ( | ||||
"skipped_contents", | |||||
"directories", | "directories", | ||||
"revisions", | "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 ("contents",): | for attr_ in ( | ||||
"contents", | |||||
"skipped_contents", | |||||
): | |||||
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_}" |