replayer_storage_and_client = (<swh.storage.in_memory.InMemoryStorage object at 0x7f8aa6b3e2b0>, <swh.journal.client.JournalClient object at 0x7f8aa6b3e4a8>) caplog = <_pytest.logging.LogCaptureFixture object at 0x7f8aa6c3ce48> def test_storage_play_with_collision(replayer_storage_and_client, caplog): """Another replayer scenario with collisions. This: - writes objects to the topic, including colliding contents - replayer consumes objects from the topic and replay them - This drops the colliding contents from the replay when detected """ src, replayer = replayer_storage_and_client # 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) # Create collision in input data # These should not be written in the destination producer = src.journal_writer.journal.producer prefix = src.journal_writer.journal._prefix for content in DUPLICATE_CONTENTS: topic = f"{prefix}.content" key = content["sha1"] producer.produce( topic=topic, key=key_to_kafka(key), value=value_to_kafka(content), ) nb_sent += 1 producer.flush() caplog.set_level(logging.ERROR, "swh.journal.replay") # 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) .tox/py3/lib/python3.7/site-packages/swh/storage/tests/test_replay.py:137: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ .tox/py3/lib/python3.7/site-packages/swh/journal/client.py:264: in process batch_processed, at_eof = self.handle_messages(messages, worker_fn) .tox/py3/lib/python3.7/site-packages/swh/journal/client.py:291: in handle_messages worker_fn(dict(objects)) .tox/py3/lib/python3.7/site-packages/swh/storage/replay.py:55: in process_replay_objects _insert_objects(object_type, objects, storage) .tox/py3/lib/python3.7/site-packages/swh/storage/replay.py:127: in _insert_objects model_objs.append(converter_fn(obj)) .tox/py3/lib/python3.7/site-packages/swh/model/model.py:73: in from_dict return cls(**d) <attrs generated init swh.model.model.OriginVisit>:12: in __init__ __attr_validator_date(self, __attr_date, self.date) .tox/py3/lib/python3.7/site-packages/attrs_strict/_type_validation.py:63: in _validator _validate_elements(attribute, field, attribute.type) _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ attribute = Attribute(name='date', default=NOTHING, validator=<function type_validator.<locals>._validator at 0x7f8c80fdbd90>, rep...True, hash=None, init=True, metadata=mappingproxy({}), type=<class 'datetime.datetime'>, converter=None, kw_only=False) value = '2018-11-27 17:20:39+00:00', expected_type = <class 'datetime.datetime'> def _validate_elements(attribute, value, expected_type): if expected_type is None: return base_type = _get_base_type(expected_type) if base_type == typing.Any: return if base_type != typing.Union and not isinstance( # type: ignore value, base_type ): > raise AttributeTypeError(value, attribute) E attrs_strict._error.AttributeTypeError: date must be <class 'datetime.datetime'> (got 2018-11-27 17:20:39+00:00 that is a <class 'str'>) .tox/py3/lib/python3.7/site-packages/attrs_strict/_type_validation.py:80: AttributeTypeError