kafka_prefix = 'xhasbvnnlq', kafka_server = '127.0.0.1:44385' consumer = <cimpl.Consumer object at 0x7f50ff7d3a60> def test_storage_direct_writer(kafka_prefix: str, kafka_server, consumer: Consumer): writer_config = { "cls": "kafka", "brokers": [kafka_server], "client_id": "kafka_writer", "prefix": kafka_prefix, "anonymize": False, } 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(): method = getattr(storage, obj_type + "_add") if obj_type in ( "content", "skipped_content", "directory", "revision", "release", "snapshot", "origin", "origin_visit_status", ): method(objs) expected_messages += len(objs) elif obj_type in ("origin_visit",): for obj in objs: assert isinstance(obj, OriginVisit) storage.origin_add_one(Origin(url=obj.origin)) visit = method([obj])[0] expected_messages += 1 + 1 # 1 visit + 1 visit status obj_d = obj.to_dict() for k in ("visit", "origin", "date", "type"): del obj_d[k] > storage.origin_visit_update(obj.origin, visit.visit, **obj_d) .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_kafka_writer.py:62: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = <swh.storage.in_memory.InMemoryStorage object at 0x7f50f5965400> origin = 'https://somewhere.org/den/fox', visit_id = 1, status = 'ongoing' metadata = {'foo': 'bar'}, snapshot = None, date = None def origin_visit_update( self, origin: str, visit_id: int, status: str, metadata: Optional[Dict] = None, snapshot: Optional[bytes] = None, date: Optional[datetime.datetime] = None, ): origin_url = self._get_origin_url(origin) if origin_url is None: raise StorageArgumentException("Unknown origin.") try: visit = self._origin_visits[origin_url][visit_id - 1] except IndexError: > raise StorageArgumentException("Unknown visit_id for this origin") from None E swh.storage.exc.StorageArgumentException: Unknown visit_id for this origin .tox/py3/lib/python3.7/site-packages/swh/storage/in_memory.py:869: StorageArgumentException