Page MenuHomeSoftware Heritage
Differential D2804 Diff 9979 swh/journal/tests/test_cli.py

Changeset View

Standalone View

View Options

swh/journal/tests/test_cli.py

Show First 20 Lines • Show All 62 Lines • ▼ Show 20 Lines
@pytest.fixture @pytest.fixture
def monkeypatch_retry_sleep(monkeypatch): def monkeypatch_retry_sleep(monkeypatch):
from swh.journal.replay import copy_object, obj_in_objstorage from swh.journal.replay import copy_object, obj_in_objstorage
monkeypatch.setattr(copy_object.retry, 'sleep', lambda x: None) monkeypatch.setattr(copy_object.retry, 'sleep', lambda x: None)
monkeypatch.setattr(obj_in_objstorage.retry, 'sleep', lambda x: None) monkeypatch.setattr(obj_in_objstorage.retry, 'sleep', lambda x: None)
def invoke(catch_exceptions, args, env=None): def invoke(*args, env=None):
config = copy.deepcopy(CLI_CONFIG) config = copy.deepcopy(CLI_CONFIG)
runner = CliRunner() runner = CliRunner()
with tempfile.NamedTemporaryFile('a', suffix='.yml') as config_fd: with tempfile.NamedTemporaryFile('a', suffix='.yml') as config_fd:
yaml.dump(config, config_fd) yaml.dump(config, config_fd)
config_fd.seek(0) config_fd.seek(0)
args = ['-C' + config_fd.name] + args args = ['-C' + config_fd.name] + list(args)
result = runner.invoke( return runner.invoke(
cli, args, obj={'log_level': logging.DEBUG}, env=env, cli, args, obj={'log_level': logging.DEBUG}, env=env,
) )
if not catch_exceptions and result.exception:
print(result.output)
raise result.exception
return result
def test_replay( def test_replay(
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]):
(_, port) = kafka_server (_, port) = kafka_server
Show All 15 Lines producer.produce(
topic=kafka_prefix+'.snapshot', topic=kafka_prefix+'.snapshot',
key=key_to_kafka(snapshot['id']), key=key_to_kafka(snapshot['id']),
value=value_to_kafka(snapshot), value=value_to_kafka(snapshot),
) )
producer.flush() producer.flush()
logger.debug('Flushed producer') logger.debug('Flushed producer')
result = invoke(False, [ result = invoke(
'replay', 'replay',
'--broker', '127.0.0.1:%d' % port, '--broker', '127.0.0.1:%d' % port,
'--group-id', kafka_consumer_group, '--group-id', kafka_consumer_group,
'--prefix', kafka_prefix, '--prefix', kafka_prefix,
'--stop-after-objects', '1', '--stop-after-objects', '1',
]) )
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
assert storage.snapshot_get(snapshot['id']) == { assert storage.snapshot_get(snapshot['id']) == {
**snapshot, 'next_branch': None} **snapshot, 'next_branch': None}
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Linesdef test_replay_content(
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)
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
]) )
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 sha1 in objstorages['dst'], sha1 assert sha1 in objstorages['dst'], sha1
assert objstorages['dst'].get(sha1) == content assert objstorages['dst'].get(sha1) == content
Show All 11 Linesdef test_replay_content_structured_log(
contents = _fill_objstorage_and_kafka( contents = _fill_objstorage_and_kafka(
kafka_port, kafka_prefix, objstorages) kafka_port, kafka_prefix, objstorages)
caplog.set_level(logging.DEBUG, 'swh.journal.replay') caplog.set_level(logging.DEBUG, 'swh.journal.replay')
expected_obj_ids = set(hash_to_hex(sha1) for sha1 in contents) expected_obj_ids = set(hash_to_hex(sha1) for sha1 in contents)
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
]) )
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
copied = set() copied = set()
for record in caplog.records: for record in caplog.records:
logtext = record.getMessage() logtext = record.getMessage()
if 'copied' in logtext: if 'copied' in logtext:
Show All 16 Linesdef test_replay_content_static_group_id(
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)
# Setup log capture to fish the consumer settings out of the log messages # Setup log capture to fish the consumer settings out of the log messages
caplog.set_level(logging.DEBUG, 'swh.journal.client') caplog.set_level(logging.DEBUG, 'swh.journal.client')
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
], {'KAFKA_GROUP_INSTANCE_ID': 'static-group-instance-id'}) env={'KAFKA_GROUP_INSTANCE_ID': 'static-group-instance-id'})
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
consumer_settings = None consumer_settings = None
for record in caplog.records: for record in caplog.records:
if 'Consumer settings' in record.message: if 'Consumer settings' in record.message:
consumer_settings = record.args consumer_settings = record.args
Show All 26 Lines contents = _fill_objstorage_and_kafka(
kafka_port, kafka_prefix, objstorages) kafka_port, kafka_prefix, objstorages)
excluded_contents = list(contents)[0::2] # picking half of them excluded_contents = list(contents)[0::2] # picking half of them
with tempfile.NamedTemporaryFile(mode='w+b') as fd: with tempfile.NamedTemporaryFile(mode='w+b') as fd:
fd.write(b''.join(sorted(excluded_contents))) fd.write(b''.join(sorted(excluded_contents)))
fd.seek(0) fd.seek(0)
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
'--exclude-sha1-file', fd.name, '--exclude-sha1-file', fd.name,
]) )
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():
if sha1 in excluded_contents: if sha1 in excluded_contents:
assert sha1 not in objstorages['dst'], sha1 assert sha1 not in objstorages['dst'], sha1
else: else:
Show All 28 Linesdef test_replay_content_check_dst(
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)
caplog.set_level(logging.DEBUG, 'swh.journal.replay') caplog.set_level(logging.DEBUG, 'swh.journal.replay')
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
'--check-dst' if check_dst else '--no-check-dst', '--check-dst' if check_dst else '--no-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
copied = 0 copied = 0
in_dst = 0 in_dst = 0
for record in caplog.records: for record in caplog.records:
logtext = record.getMessage() logtext = record.getMessage()
▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines for i, (sha1, content) in enumerate(contents.items()):
objstorages['dst'].add(content, obj_id=sha1) objstorages['dst'].add(content, obj_id=sha1)
failures['in', sha1] = 1 failures['in', sha1] = 1
orig_dst = objstorages['dst'] orig_dst = objstorages['dst']
objstorages['dst'] = FlakyObjStorage(state=orig_dst.state, objstorages['dst'] = FlakyObjStorage(state=orig_dst.state,
failures=failures) failures=failures)
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', 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 sha1 in objstorages['dst'], sha1 assert sha1 in objstorages['dst'], sha1
assert objstorages['dst'].get(sha1) == content assert objstorages['dst'].get(sha1) == content
▲ Show 20 Lines • Show All 50 Lines • ▼ Show 20 Linesdef test_replay_content_failed_copy_retry(
) )
objstorages['src'] = FlakyObjStorage( objstorages['src'] = FlakyObjStorage(
state=objstorages['src'].state, state=objstorages['src'].state,
failures=get_failures, failures=get_failures,
) )
caplog.set_level(logging.DEBUG, 'swh.journal.replay') caplog.set_level(logging.DEBUG, 'swh.journal.replay')
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
]) )
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
copied = 0 copied = 0
actually_failed = set() actually_failed = set()
for record in caplog.records: for record in caplog.records:
logtext = record.getMessage() logtext = record.getMessage()
Show All 38 Lines for i, sha1 in enumerate(contents):
if i >= num_contents_deleted: if i >= num_contents_deleted:
break break
del objstorages['src'].state[sha1] del objstorages['src'].state[sha1]
contents_deleted.add(hash_to_hex(sha1)) contents_deleted.add(hash_to_hex(sha1))
caplog.set_level(logging.DEBUG, 'swh.journal.replay') caplog.set_level(logging.DEBUG, 'swh.journal.replay')
result = invoke(False, [ result = invoke(
'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,
'--stop-after-objects', str(NUM_CONTENTS), '--stop-after-objects', str(NUM_CONTENTS),
]) )
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
copied = 0 copied = 0
not_in_src = set() not_in_src = set()
for record in caplog.records: for record in caplog.records:
logtext = record.getMessage() logtext = record.getMessage()
Show All 20 Lines
About · Developer information · Legal