replayer_storage_and_client = (<swh.storage.in_memory.InMemoryStorage object at 0x7f9ed020c438>, <swh.journal.client.JournalClient object at 0x7f9ed01ed358>) caplog = <_pytest.logging.LogCaptureFixture object at 0x7f9ed0210f28> def test_storage_replayer(replayer_storage_and_client, caplog): """Optimal replayer scenario. This: - writes objects to a source storage - replayer consumes objects from the topic and replays them - a destination storage is filled from this In the end, both storages should have the same content. """ src, replayer = replayer_storage_and_client # Fill Kafka using a source storage nb_sent = 0 for object_type, objects in TEST_OBJECTS.items(): if object_type == "origin_visit": # src.origin_visit_upsert(objects) for visit in objects: src.origin_visit_add( origin_url=visit.origin, date=visit.date, type=visit.type ) else: method = getattr(src, object_type + "_add") method(objects) nb_sent += len(objects) caplog.set_level(logging.ERROR, "swh.journal.replay") # Fill the destination storage from Kafka dst = get_storage(cls="memory") worker_fn = functools.partial(process_replay_objects, storage=dst) nb_inserted = replayer.process(worker_fn) > assert nb_sent == nb_inserted E assert 31 == 36 E +31 E -36 .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_replay.py:90: AssertionError