replayer_storage_and_client = (<swh.storage.in_memory.InMemoryStorage object at 0x7f56f7bae588>, <swh.journal.client.JournalClient object at 0x7f56fb5bcc88>) caplog = <_pytest.logging.LogCaptureFixture object at 0x7f56fef13748> redisdb = Redis<ConnectionPool<UnixDomainSocketConnection<path=/tmp/pytest-of-jenkins/pytest-0/pytest-redis-redis_proc0/redis.27429.sock,db=0>>> def test_storage_replayer_with_validation_ok( replayer_storage_and_client, caplog, redisdb ): """Optimal replayer scenario with validation activated and reporter set to a redis db. - writes objects to a source storage - replayer consumes objects from the topic and replays them - a destination storage is filled from this - nothing has been reported in the redis db - both storages should have the same content """ src, replayer = replayer_storage_and_client replayer.deserializer = ModelObjectDeserializer(validate=True, reporter=redisdb.set) # Fill Kafka using a source storage nb_sent = 0 for object_type, objects in TEST_OBJECTS.items(): method = getattr(src, object_type + "_add") method(objects) if object_type == "origin_visit": nb_sent += len(objects) # origin-visit-add adds origin-visit-status as well nb_sent += len(objects) # 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 58 == 53 .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_replay.py:418: AssertionError