Show First 20 Lines • Show All 376 Lines • ▼ Show 20 Lines	assert (copied == expected_copied and in_dst == expected_in_dst), (
"Unexpected amount of objects copied, see the captured log for details"		"Unexpected amount of objects copied, see the captured log for details"
)		)

for (sha1, content) in contents.items():		for (sha1, content) in contents.items():
assert sha1 in objstorages['dst'], sha1		assert sha1 in objstorages['dst'], sha1
assert objstorages['dst'].get(sha1) == content		assert objstorages['dst'].get(sha1) == content


		class FlakyObjStorage(InMemoryObjStorage):
		def __init__(self, args, *kwargs):
		state = kwargs.pop('state')
		self.failures_left = Counter(kwargs.pop('failures'))
		super().__init__(args, *kwargs)
		if state:
		self.state = state

		def flaky_operation(self, op, obj_id):
		if self.failures_left[op, obj_id] > 0:
		self.failures_left[op, obj_id] -= 1
		raise RuntimeError(
		'Failed %s on %s' % (op, hash_to_hex(obj_id))
		)

		def get(self, obj_id):
		self.flaky_operation('get', obj_id)
		return super().get(obj_id)

		def add(self, data, obj_id=None, check_presence=True):
		self.flaky_operation('add', obj_id)
		return super().add(data, obj_id=obj_id,
		check_presence=check_presence)

		def __contains__(self, obj_id):
		self.flaky_operation('in', obj_id)
		return super().__contains__(obj_id)


@_patch_objstorages(['src', 'dst'])		@_patch_objstorages(['src', 'dst'])
def test_replay_content_check_dst_retry(		def test_replay_content_check_dst_retry(
objstorages,		objstorages,
storage,		storage,
kafka_prefix: str,		kafka_prefix: str,
kafka_consumer_group: str,		kafka_consumer_group: str,
kafka_server: Tuple[Popen, int]):		kafka_server: Tuple[Popen, int]):
(_, kafka_port) = kafka_server		(_, kafka_port) = kafka_server
kafka_prefix += '.swh.journal.objects'		kafka_prefix += '.swh.journal.objects'

contents = _fill_objstorage_and_kafka(		contents = _fill_objstorage_and_kafka(
kafka_port, kafka_prefix, objstorages)		kafka_port, kafka_prefix, objstorages)

class FlakyMemoryObjStorage(InMemoryObjStorage):		failures = {}
def __init__(self, args, *kwargs):
state = kwargs.pop('state')
super().__init__(args, *kwargs)
if state:
self.state = state
self.contains_failed = Counter()

def __contains__(self, id):
if self.contains_failed[id] == 0:
self.contains_failed[id] += 1
raise ValueError('This contains is flaky')

return super().__contains__(id)

for i, (sha1, content) in enumerate(contents.items()):		for i, (sha1, content) in enumerate(contents.items()):
if i >= NUM_CONTENTS_DST:		if i >= NUM_CONTENTS_DST:
break		break

objstorages['dst'].add(content, obj_id=sha1)		objstorages['dst'].add(content, obj_id=sha1)
		failures['in', sha1] = 1

dst_state = objstorages['dst'].state		orig_dst = objstorages['dst']
flaky_objstorage = FlakyMemoryObjStorage(state=dst_state)		objstorages['dst'] = FlakyObjStorage(state=orig_dst.state,
objstorages['dst'] = flaky_objstorage		failures=failures)
dst_contains_failed = flaky_objstorage.contains_failed

result = invoke(False, [		result = invoke(False, [
'content-replay',		'content-replay',
'--broker', '127.0.0.1:%d' % kafka_port,		'--broker', '127.0.0.1:%d' % kafka_port,
'--group-id', kafka_consumer_group,		'--group-id', kafka_consumer_group,
'--prefix', kafka_prefix,		'--prefix', kafka_prefix,
'--max-messages', str(NUM_CONTENTS),		'--max-messages', str(NUM_CONTENTS),
'--check-dst',		'--check-dst',
])		])
expected = r'Done.\n'		expected = r'Done.\n'
assert result.exit_code == 0, result.output		assert result.exit_code == 0, result.output
assert re.fullmatch(expected, result.output, re.MULTILINE), result.output		assert re.fullmatch(expected, result.output, re.MULTILINE), result.output

for (sha1, content) in contents.items():		for (sha1, content) in contents.items():
assert dst_contains_failed[sha1] >= 0		assert sha1 in objstorages['dst'], sha1
assert dst_state[sha1] == content		assert objstorages['dst'].get(sha1) == content


@_patch_objstorages(['src', 'dst'])		@_patch_objstorages(['src', 'dst'])
def test_replay_content_objnotfound(		def test_replay_content_objnotfound(
objstorages,		objstorages,
storage,		storage,
kafka_prefix: str,		kafka_prefix: str,
kafka_consumer_group: str,		kafka_consumer_group: str,
▲ Show 20 Lines • Show All 55 Lines • Show Last 20 Lines

Make the flaky object storage genericClosedPublicActions

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 9897

swh/journal/tests/test_cli.py

Make the flaky object storage generic
ClosedPublic
Actions

Revision Contents
Changeset List