kafka_prefix = 'jlcxfdfhuo', kafka_consumer_group = 'test-consumer-jlcxfdfhuo' kafka_server = '127.0.0.1:51795' def test_storage_play_anonymized( kafka_prefix: str, kafka_consumer_group: str, kafka_server: str ): """Optimal replayer scenario. This: - writes objects to the topic - replayer consumes objects from the topic and replay them """ writer_config = { "cls": "kafka", "brokers": [kafka_server], "client_id": "kafka_writer", "prefix": kafka_prefix, "anonymize": True, } src_config: Dict[str, Any] = {"cls": "memory", "journal_writer": writer_config} storage = get_storage(**src_config) # Fill the src storage nb_sent = 0 for obj_type, objs in TEST_OBJECTS.items(): if obj_type in ("origin_visit", "origin_visit_status"): # these are unrelated with what we want to test here continue method = getattr(storage, obj_type + "_add") method(objs) nb_sent += len(objs) # Fill a destination storage from Kafka **using anonymized topics** dst_storage = get_storage(cls="memory") replayer = JournalClient( brokers=kafka_server, group_id=kafka_consumer_group, prefix=kafka_prefix, stop_after_objects=nb_sent, privileged=False, ) worker_fn = functools.partial(process_replay_objects, storage=dst_storage) nb_inserted = replayer.process(worker_fn) assert nb_sent == nb_inserted > check_replayed(storage, dst_storage, expected_anonymized=True) .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_replay.py:335: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_replay.py:373: in check_replayed expected_persons = {maybe_anonymize(person) for person in src._persons.values()} _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .0 = <dict_valueiterator object at 0x7f3b015d0ef8> > expected_persons = {maybe_anonymize(person) for person in src._persons.values()} E TypeError: maybe_anonymize() missing 1 required positional argument: 'row' .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_replay.py:373: TypeError