kafka_prefix = 'kwsrblufgh', kafka_server = '127.0.0.1:54133' consumer = <cimpl.Consumer object at 0x7f9ed021e158> 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) expected_messages += len(objs) existing_topics = set( topic for topic in consumer.list_topics(timeout=10).topics.keys() if topic.startswith(kafka_prefix) ) > assert existing_topics == { f"{kafka_prefix}.{obj_type}" for obj_type in ( "content", "directory", "origin", "origin_visit", "release", "revision", "snapshot", "skipped_content", ) } | { f"{kafka_prefix}_privileged.{obj_type}" for obj_type in ("release", "revision",) } E AssertionError: assert {'kwsrblufgh....release', ...} == {'kwsrblufgh....evision', ...} E Extra items in the left set: E 'kwsrblufgh.origin_visit_status' E Full diff: E { E 'kwsrblufgh.content', E 'kwsrblufgh.directory', E 'kwsrblufgh.origin',... E E ...Full output truncated (10 lines hidden), use '-vv' to show .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_kafka_writer.py:123: AssertionError