Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_backfill.py
Show All 13 Lines | |||||
from swh.storage import get_storage | from swh.storage import get_storage | ||||
from swh.storage.backfill import ( | from swh.storage.backfill import ( | ||||
PARTITION_KEY, | PARTITION_KEY, | ||||
JournalBackfiller, | JournalBackfiller, | ||||
byte_ranges, | byte_ranges, | ||||
compute_query, | compute_query, | ||||
raw_extrinsic_metadata_target_ranges, | raw_extrinsic_metadata_target_ranges, | ||||
) | ) | ||||
from swh.storage.in_memory import InMemoryStorage | |||||
from swh.storage.replay import process_replay_objects | from swh.storage.replay import process_replay_objects | ||||
from swh.storage.tests.test_replay import check_replayed | from swh.storage.tests.test_replay import check_replayed | ||||
TEST_CONFIG = { | TEST_CONFIG = { | ||||
"journal_writer": { | "journal_writer": { | ||||
"brokers": ["localhost"], | "brokers": ["localhost"], | ||||
"prefix": "swh.tmp_journal.new", | "prefix": "swh.tmp_journal.new", | ||||
"client_id": "swh.journal.client.test", | "client_id": "swh.journal.client.test", | ||||
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Lines | replayer2 = JournalClient( | ||||
group_id=f"{kafka_consumer_group}-2", | group_id=f"{kafka_consumer_group}-2", | ||||
prefix=prefix2, | prefix=prefix2, | ||||
stop_on_eof=True, | stop_on_eof=True, | ||||
) | ) | ||||
worker_fn2 = functools.partial(process_replay_objects, storage=sto2) | worker_fn2 = functools.partial(process_replay_objects, storage=sto2) | ||||
replayer2.process(worker_fn2) | replayer2.process(worker_fn2) | ||||
# Compare storages | # Compare storages | ||||
assert isinstance(sto1, InMemoryStorage) # needed to help mypy | |||||
assert isinstance(sto2, InMemoryStorage) | |||||
check_replayed(sto1, sto2) | check_replayed(sto1, sto2) | ||||
for record in caplog.records: | for record in caplog.records: | ||||
assert ( | assert ( | ||||
"this should not happen" not in record.message | "this should not happen" not in record.message | ||||
), "Replayer ignored some message types, see captured logging" | ), "Replayer ignored some message types, see captured logging" |