kafka_prefix = 'zxqtrkdvpr', kafka_server = '127.0.0.1:46949' consumer = <cimpl.Consumer object at 0x7f165a715950> def test_storage_direct_writer_anonymized( kafka_prefix: str, kafka_server, consumer: Consumer ): writer_config = { "cls": "kafka", "brokers": [kafka_server], "client_id": "kafka_writer", "prefix": kafka_prefix, "anonymize": True, } storage_config = { "cls": "pipeline", "steps": [{"cls": "memory", "journal_writer": writer_config},], } storage = get_storage(**storage_config) expected_messages = 0 for obj_type, objs in TEST_OBJECTS.items(): if obj_type == "origin_visit": # these have non-consistent API and are unrelated with what we # want to test here continue method = getattr(storage, obj_type + "_add") > method(objs) .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_kafka_writer.py:112: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .tox/py3/lib/python3.7/site-packages/swh/storage/in_memory.py:503: in release_add self.journal_writer.release_add(releases) .tox/py3/lib/python3.7/site-packages/swh/storage/writer.py:95: in release_add self.write_additions("release", releases) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <swh.storage.writer.JournalWriter object at 0x7f165a6fdcc0> obj_type = 'release' values = [Release(name=b'v0.0.1', message=b'foo', target=b'\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04\x04...offset=120, negative_utc=False), metadata=None, id=b'\xd8\x1c\xc0q\x0e\xb6\xcf\x9e\xfd[\x92\n\x84S\xe1\xe0qW\xb6\xcd')] def write_additions(self, obj_type, values) -> None: if self.journal: if obj_type not in self.anonymizable or not self.anonymize: self.journal.write_additions(obj_type, values) else: self.journal.write_additions( > obj_type, [anonymize(value) for value in values], privileged=False ) E TypeError: write_additions() got an unexpected keyword argument 'privileged' .tox/py3/lib/python3.7/site-packages/swh/storage/writer.py:67: TypeError