Differential D1390 Diff 4486 swh/journal/tests/test_write_replay.py

Changeset View

Standalone View

swh/journal/tests/test_write_replay.py

	Show All 11 Lines
	from swh.storage.in_memory import Storage			from swh.storage.in_memory import Storage
	from swh.storage import HashCollision			from swh.storage import HashCollision

	from swh.journal.serializers import (			from swh.journal.serializers import (
	key_to_kafka, kafka_to_key, value_to_kafka, kafka_to_value)			key_to_kafka, kafka_to_key, value_to_kafka, kafka_to_value)
	from swh.journal.direct_writer import DirectKafkaWriter			from swh.journal.direct_writer import DirectKafkaWriter
	from swh.journal.replay import StorageReplayer, OBJECT_TYPES			from swh.journal.replay import StorageReplayer, OBJECT_TYPES

	FakeKafkaMessage = namedtuple('FakeKafkaMessage', 'topic key value')			FakeKafkaMessage = namedtuple('FakeKafkaMessage', 'key value')
				FakeKafkaPartition = namedtuple('FakeKafkaPartition', 'topic')


	class MockedDirectKafkaWriter(DirectKafkaWriter):			class MockedDirectKafkaWriter(DirectKafkaWriter):
	def __init__(self):			def __init__(self):
	self._prefix = 'prefix'			self._prefix = 'prefix'


	class MockedStorageReplayer(StorageReplayer):			class MockedStorageReplayer(StorageReplayer):
	def __init__(self, object_types=OBJECT_TYPES):			def __init__(self, object_types=OBJECT_TYPES):
	self._object_types = object_types			self._object_types = object_types


	@given(lists(object_dicts()))			@given(lists(object_dicts(), min_size=1))
	@settings(suppress_health_check=[HealthCheck.too_slow])			@settings(suppress_health_check=[HealthCheck.too_slow])
	def test_write_replay_same_order(objects):			def test_write_replay_same_order(objects):
				committed = False
	queue = []			queue = []

	def send(topic, key, value):			def send(topic, key, value):
	key = kafka_to_key(key_to_kafka(key))			key = kafka_to_key(key_to_kafka(key))
	value = kafka_to_value(value_to_kafka(value))			value = kafka_to_value(value_to_kafka(value))
	queue.append(FakeKafkaMessage(topic=topic, key=key, value=value))			queue.append({
				FakeKafkaPartition(topic):
				[FakeKafkaMessage(key=key, value=value)]
				})

	def poll():			def poll():
	yield from queue			return queue.pop(0)

				def commit():
				nonlocal committed
				if queue == []:
				committed = True

	storage1 = Storage()			storage1 = Storage()
	storage1.journal_writer = MockedDirectKafkaWriter()			storage1.journal_writer = MockedDirectKafkaWriter()
	storage1.journal_writer.send = send			storage1.journal_writer.send = send

	for (obj_type, obj) in objects:			for (obj_type, obj) in objects:
	obj = obj.copy()			obj = obj.copy()
	if obj_type == 'origin_visit':			if obj_type == 'origin_visit':
	origin_id = storage1.origin_add_one(obj.pop('origin'))			origin_id = storage1.origin_add_one(obj.pop('origin'))
	if 'visit' in obj:			if 'visit' in obj:
	del obj['visit']			del obj['visit']
	storage1.origin_visit_add(origin_id, **obj)			storage1.origin_visit_add(origin_id, **obj)
	else:			else:
	method = getattr(storage1, obj_type + '_add')			method = getattr(storage1, obj_type + '_add')
	try:			try:
	method([obj])			method([obj])
	except HashCollision:			except HashCollision:
	pass			pass

	storage2 = Storage()			storage2 = Storage()
	replayer = MockedStorageReplayer()			replayer = MockedStorageReplayer()
	replayer.poll = poll			replayer.poll = poll
	replayer.fill(storage2)			replayer.commit = commit
				replayer.fill(storage2, max_messages=len(queue))

				assert committed

	for attr_name in ('_contents', '_directories', '_revisions', '_releases',			for attr_name in ('_contents', '_directories', '_revisions', '_releases',
	'_snapshots', '_origin_visits', '_origins'):			'_snapshots', '_origin_visits', '_origins'):
	assert getattr(storage1, attr_name) == getattr(storage2, attr_name), \			assert getattr(storage1, attr_name) == getattr(storage2, attr_name), \
	attr_name			attr_name


	# TODO: add test for hash collision			# TODO: add test for hash collision