Differential D3161 Diff 11223 swh/storage/tests/test_replay.py

Changeset View

Standalone View

swh/storage/tests/test_replay.py

Show All 13 Lines
from confluent_kafka import Producer		from confluent_kafka import Producer

import pytest		import pytest

from swh.model.hashutil import hash_to_hex, MultiHash, DEFAULT_ALGORITHMS		from swh.model.hashutil import hash_to_hex, MultiHash, DEFAULT_ALGORITHMS
from swh.model.model import Content		from swh.model.model import Content

from swh.storage import get_storage		from swh.storage import get_storage
		from swh.storage.writer import anonymize
from swh.storage.replay import process_replay_objects		from swh.storage.replay import process_replay_objects

from swh.journal.serializers import key_to_kafka, value_to_kafka		from swh.journal.serializers import key_to_kafka, value_to_kafka
from swh.journal.client import JournalClient		from swh.journal.client import JournalClient
		from swh.journal.tests.journal_data import TEST_OBJECTS

from swh.journal.tests.utils import MockedJournalClient, MockedKafkaWriter		from swh.journal.tests.utils import MockedJournalClient, MockedKafkaWriter
from swh.journal.tests.conftest import (		from swh.journal.tests.conftest import (
TEST_OBJECT_DICTS,		TEST_OBJECT_DICTS,
DUPLICATE_CONTENTS,		DUPLICATE_CONTENTS,
)		)


▲ Show 20 Lines • Show All 424 Lines • ▼ Show 20 Lines	return [
"status": "absent",		"status": "absent",
"reason": "why not",		"reason": "why not",
"origin": f"https://somewhere/{i}",		"origin": f"https://somewhere/{i}",
"ctime": now,		"ctime": now,
},		},
)		)
for i in range(n)		for i in range(n)
]		]


		def test_storage_play_anonymized(
		kafka_prefix: str, kafka_consumer_group: str, kafka_server: str, caplog,
		):
		"""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 = {"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 == "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)
		nb_sent += len(objs)

		dst_storage = get_storage(cls="memory")
		caplog.set_level(logging.ERROR, "swh.journal.replay")
		# Fill the storage from Kafka using anonymized topics
		object_types = {
		"content",
		"directory",
		"origin",
		"origin_visit",
		"release",
		"revision",
		"snapshot",
		"skipped_content",
		}

		replayer = JournalClient(
		brokers=kafka_server,
		group_id=kafka_consumer_group,
		prefix=kafka_prefix,
		stop_after_objects=nb_sent,
		object_types=list(object_types),
		)
		worker_fn = functools.partial(process_replay_objects, storage=dst_storage)

		nb_inserted = replayer.process(worker_fn)
		assert nb_sent == nb_inserted

		expected_persons = {anonymize(person) for person in storage._persons}
		assert set(dst_storage._persons) == expected_persons